Extend the IP-level ban to Router-bound connections

node/bft/Cargo.toml

@@ -19,6 +19,7 @@ edition = "2021"
 [features]
 default = [ ]
 metrics = [ "dep:metrics", "snarkos-node-bft-events/metrics", "snarkos-node-bft-ledger-service/metrics" ]
+test = [ ]
 
 [dependencies.aleo-std]
 workspace = true
@@ -152,6 +153,10 @@ version = "0.4"
 [dev-dependencies.rayon]
 version = "1"
 
+[dev-dependencies.snarkos-node-bft]
+path = "."
+features = [ "test" ]
+
 [dev-dependencies.snarkos-node-bft-ledger-service]
 path = "./ledger-service"
 default-features = false

node/bft/src/gateway.rs

@@ -64,6 +64,8 @@ use futures::SinkExt;
 use indexmap::{IndexMap, IndexSet};
 use parking_lot::{Mutex, RwLock};
 use rand::seq::{IteratorRandom, SliceRandom};
+#[cfg(not(any(test, feature = "test")))]
+use std::net::IpAddr;
 use std::{collections::HashSet, future::Future, io, net::SocketAddr, sync::Arc, time::Duration};
 use tokio::{
  net::TcpStream,
@@ -88,6 +90,12 @@ const MIN_CONNECTED_VALIDATORS: usize = 175;
 /// The maximum number of validators to send in a validators response event.
 const MAX_VALIDATORS_TO_SEND: usize = 200;
 
+/// The minimum permitted interval between connection attempts for an IP; anything shorter is considered malicious.
+#[cfg(not(any(test, feature = "test")))]
+const MIN_CONNECTION_INTERVAL_IN_SECS: u64 = 10;
+/// The amount of time an IP address is prohibited from connecting.
+const IP_BAN_TIME_IN_SECS: u64 = 30;
+
 /// Part of the Gateway API that deals with networking.
 /// This is a separate trait to allow for easier testing/mocking.
 #[async_trait]
@@ -459,6 +467,18 @@ impl<N: Network> Gateway<N> {
  Ok(())
  }
 
+ /// Check whether the given IP address is currently banned.
+ #[cfg(not(any(test, feature = "test")))]
+ fn is_ip_banned(&self, ip: IpAddr) -> bool {
+ self.tcp.banned_peers().is_ip_banned(ip)
+ }
+
+ /// Insert or update a banned IP.
+ #[cfg(not(any(test, feature = "test")))]
+ fn update_ip_ban(&self, ip: IpAddr) {
+ self.tcp.banned_peers().update_ip_ban(ip);
+ }
+
  #[cfg(feature = "metrics")]
  fn update_metrics(&self) {
  metrics::gauge(metrics::bft::CONNECTED, self.connected_peers.read().len() as f64);
@@ -875,6 +895,8 @@ impl<N: Network> Gateway<N> {
  self.handle_unauthorized_validators();
  // If the number of connected validators is less than the minimum, send a `ValidatorsRequest`.
  self.handle_min_connected_validators();
+ // Unban any addresses whose ban time has expired.
+ self.handle_banned_ips();
  }
 
  /// Logs the connected validators.
@@ -955,6 +977,11 @@ impl<N: Network> Gateway<N> {
  }
  }
  }
+
+ // Remove addresses whose ban time has expired.
+ fn handle_banned_ips(&self) {
+ self.tcp.banned_peers().remove_old_bans(IP_BAN_TIME_IN_SECS);
+ }
 }
 
 #[async_trait]
@@ -1105,9 +1132,29 @@ impl<N: Network> OnConnect for Gateway<N> {
 impl<N: Network> Handshake for Gateway<N> {
  /// Performs the handshake protocol.
  async fn perform_handshake(&self, mut connection: Connection) -> io::Result<Connection> {
- // Perform the handshake.
  let peer_addr = connection.addr();
  let peer_side = connection.side();
+
+ // Check (or impose) IP-level bans.
+ #[cfg(not(any(test, feature = "test")))]
+ if self.dev().is_none() && peer_side == ConnectionSide::Initiator {
+ // If the IP is already banned, update the ban timestamp and reject the connection.
+ if self.is_ip_banned(peer_addr.ip()) {
+ self.update_ip_ban(peer_addr.ip());
+ trace!("{CONTEXT} Gateway rejected a connection request from banned IP '{}'", peer_addr.ip());
+ return Err(error(format!("'{}' is a banned IP address", peer_addr.ip())));
+ }
+
+ // Check the previous low-level connection timestamp.
+ if let Some(peer_stats) = self.tcp.known_peers().get(peer_addr.ip()) {
+ if peer_stats.timestamp().elapsed().as_secs() <= MIN_CONNECTION_INTERVAL_IN_SECS {
+ self.update_ip_ban(peer_addr.ip());
+ trace!("{CONTEXT} Gateway rejected a consecutive connection request from IP '{}'", peer_addr.ip());
+ return Err(error(format!("'{}' appears to be spamming connections", peer_addr.ip())));
+ }
+ }
+ }
+
  let stream = self.borrow_stream(&mut connection);
 
  // If this is an inbound connection, we log it, but don't know the listening address yet.

node/router/src/handshake.rs

@@ -100,6 +100,26 @@ impl<N: Network> Router<N> {
  Some(peer_addr)
  };
 
+ // Check (or impose) IP-level bans.
+ #[cfg(not(any(test, feature = "test")))]
+ if !self.is_dev() && peer_side == ConnectionSide::Initiator {
+ // If the IP is already banned, update the ban timestamp and reject the connection.
+ if self.is_ip_banned(peer_addr.ip()) {
+ self.update_ip_ban(peer_addr.ip());
+ trace!("Rejected a connection request from banned IP '{}'", peer_addr.ip());
+ return Err(error(format!("'{}' is a banned IP address", peer_addr.ip())));
+ }
+
+ // Check the previous low-level connection timestamp.
+ if let Some(peer_stats) = self.tcp.known_peers().get(peer_addr.ip()) {
+ if peer_stats.timestamp().elapsed().as_secs() <= Router::<N>::MIN_CONNECTION_INTERVAL_IN_SECS {
+ self.update_ip_ban(peer_addr.ip());
+ trace!("Rejected a consecutive connection request from IP '{}'", peer_addr.ip());
+ return Err(error(format!("'{}' appears to be spamming connections", peer_addr.ip())));
+ }
+ }
+ }
+
  // Perform the handshake; we pass on a mutable reference to peer_ip in case the process is broken at any point in time.
  let handshake_result = if peer_side == ConnectionSide::Responder {
  self.handshake_inner_initiator(peer_addr, &mut peer_ip, stream, genesis_header, restrictions_id).await

node/router/src/heartbeat.rs

@@ -42,6 +42,8 @@ pub trait Heartbeat<N: Network>: Outbound<N> {
  const MAXIMUM_NUMBER_OF_PEERS: usize = 21;
  /// The maximum number of provers to maintain connections with.
  const MAXIMUM_NUMBER_OF_PROVERS: usize = Self::MAXIMUM_NUMBER_OF_PEERS / 4;
+ /// The amount of time an IP address is prohibited from connecting.
+ const IP_BAN_TIME_IN_SECS: u64 = 30;
 
  /// Handles the heartbeat request.
  fn heartbeat(&self) {
@@ -60,6 +62,8 @@ pub trait Heartbeat<N: Network>: Outbound<N> {
  self.handle_trusted_peers();
  // Keep the puzzle request up to date.
  self.handle_puzzle_request();
+ // Unban any addresses whose ban time has expired.
+ self.handle_banned_ips();
  }
 
  /// TODO (howardwu): Consider checking minimum number of validators, to exclude clients and provers.
@@ -283,4 +287,9 @@ pub trait Heartbeat<N: Network>: Outbound<N> {
  fn handle_puzzle_request(&self) {
  // No-op
  }
+
+ // Remove addresses whose ban time has expired.
+ fn handle_banned_ips(&self) {
+ self.tcp().banned_peers().remove_old_bans(Self::IP_BAN_TIME_IN_SECS);
+ }
 }

node/router/src/lib.rs

@@ -45,6 +45,8 @@ use snarkvm::prelude::{Address, Network, PrivateKey, ViewKey};
 
 use anyhow::{bail, Result};
 use parking_lot::{Mutex, RwLock};
+#[cfg(not(any(test, feature = "test")))]
+use std::net::IpAddr;
 use std::{
  collections::{HashMap, HashSet},
  future::Future,
@@ -107,6 +109,9 @@ impl<N: Network> Router<N> {
  /// The duration in seconds after which a connected peer is considered inactive or
  /// disconnected if no message has been received in the meantime.
  const RADIO_SILENCE_IN_SECS: u64 = 150; // 2.5 minutes
+ /// The minimum permitted interval between connection attempts for an IP; anything shorter is considered malicious.
+ #[cfg(not(any(test, feature = "test")))]
+ const MIN_CONNECTION_INTERVAL_IN_SECS: u64 = 10;
 }
 
 impl<N: Network> Router<N> {
@@ -427,6 +432,18 @@ impl<N: Network> Router<N> {
  self.connected_peers.read().iter().map(|(ip, peer)| (*ip, peer.node_type())).collect()
  }
 
+ /// Check whether the given IP address is currently banned.
+ #[cfg(not(any(test, feature = "test")))]
+ fn is_ip_banned(&self, ip: IpAddr) -> bool {
+ self.tcp.banned_peers().is_ip_banned(ip)
+ }
+
+ /// Insert or update a banned IP.
+ #[cfg(not(any(test, feature = "test")))]
+ fn update_ip_ban(&self, ip: IpAddr) {
+ self.tcp.banned_peers().update_ip_ban(ip);
+ }
+
  #[cfg(feature = "metrics")]
  fn update_metrics(&self) {
  metrics::gauge(metrics::router::CONNECTED, self.connected_peers.read().len() as f64);

node/tcp/src/helpers/banned_peers.rs

@@ -0,0 +1,39 @@
+// Copyright (C) 2019-2023 Aleo Systems Inc.
+// This file is part of the snarkOS library.
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+// http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{collections::HashMap, net::IpAddr, time::Instant};
+
+use parking_lot::RwLock;
+
+/// Contains the set of peers currently banned by IP.
+#[derive(Default)]
+pub struct BannedPeers(RwLock<HashMap<IpAddr, Instant>>);
+
+impl BannedPeers {
+ /// Check whether the given IP address is currently banned.
+ pub fn is_ip_banned(&self, ip: IpAddr) -> bool {
+ self.0.read().contains_key(&ip)
+ }
+
+ /// Insert or update a banned IP.
+ pub fn update_ip_ban(&self, ip: IpAddr) {
+ let timestamp = Instant::now();
+ self.0.write().insert(ip, timestamp);
+ }
+
+ /// Remove the expired entries.
+ pub fn remove_old_bans(&self, ban_time_in_secs: u64) {
+ self.0.write().retain(|_, timestamp| timestamp.elapsed().as_secs() < ban_time_in_secs);
+ }
+}

node/tcp/src/helpers/known_peers.rs

@@ -12,53 +12,66 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{collections::HashMap, net::SocketAddr, sync::Arc};
+use std::{
+ collections::{hash_map::Entry, HashMap},
+ net::IpAddr,
+ sync::Arc,
+ time::Instant,
+};
 
 use parking_lot::RwLock;
 
 use crate::Stats;
 
 /// Contains statistics related to Tcp's peers, currently connected or not.
 #[derive(Default)]
-pub struct KnownPeers(RwLock<HashMap<SocketAddr, Arc<Stats>>>);
+pub struct KnownPeers(RwLock<HashMap<IpAddr, Arc<Stats>>>);
 
 impl KnownPeers {
  /// Adds an address to the list of known peers.
- pub fn add(&self, addr: SocketAddr) {
- self.0.write().entry(addr).or_default();
+ pub fn add(&self, addr: IpAddr) {
+ let timestamp = Instant::now();
+ match self.0.write().entry(addr) {
+ Entry::Vacant(entry) => {
+ entry.insert(Arc::new(Stats::new(timestamp)));
+ }
+ Entry::Occupied(entry) => {
+ *entry.get().timestamp.write() = timestamp;
+ }
+ }
  }
 
  /// Returns the stats for the given peer.
- pub fn get(&self, addr: SocketAddr) -> Option<Arc<Stats>> {
+ pub fn get(&self, addr: IpAddr) -> Option<Arc<Stats>> {
  self.0.read().get(&addr).map(Arc::clone)
  }
 
  /// Removes an address from the list of known peers.
- pub fn remove(&self, addr: SocketAddr) -> Option<Arc<Stats>> {
+ pub fn remove(&self, addr: IpAddr) -> Option<Arc<Stats>> {
  self.0.write().remove(&addr)
  }
 
  /// Returns the list of all known peers and their stats.
- pub fn snapshot(&self) -> HashMap<SocketAddr, Arc<Stats>> {
+ pub fn snapshot(&self) -> HashMap<IpAddr, Arc<Stats>> {
  self.0.read().clone()
  }
 
  /// Registers a submission of a message to the given address.
- pub fn register_sent_message(&self, to: SocketAddr, size: usize) {
+ pub fn register_sent_message(&self, to: IpAddr, size: usize) {
  if let Some(stats) = self.0.read().get(&to) {
  stats.register_sent_message(size);
  }
  }
 
  /// Registers a receipt of a message to the given address.
- pub fn register_received_message(&self, from: SocketAddr, size: usize) {
+ pub fn register_received_message(&self, from: IpAddr, size: usize) {
  if let Some(stats) = self.0.read().get(&from) {
  stats.register_received_message(size);
  }
  }
 
  /// Registers a failure associated with the given address.
- pub fn register_failure(&self, addr: SocketAddr) {
+ pub fn register_failure(&self, addr: IpAddr) {
  if let Some(stats) = self.0.read().get(&addr) {
  stats.register_failure();
  }

node/tcp/src/helpers/mod.rs

@@ -15,6 +15,9 @@
 mod config;
 pub use config::Config;
 
+mod banned_peers;
+pub use banned_peers::BannedPeers;
+
 pub mod connections;
 pub use connections::{Connection, ConnectionSide};
 

node/tcp/src/helpers/stats.rs

@@ -12,11 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
+use parking_lot::RwLock;
+use std::{
+ sync::atomic::{AtomicU64, Ordering::Relaxed},
+ time::Instant,
+};
 
 /// Contains statistics related to Tcp.
-#[derive(Default)]
 pub struct Stats {
+ /// The timestamp of the creation (for the node) or connection (to a peer).
+ pub(crate) timestamp: RwLock<Instant>,
  /// The number of all messages sent.
  msgs_sent: AtomicU64,
  /// The number of all messages received.
@@ -30,6 +35,23 @@ pub struct Stats {
 }
 
 impl Stats {
+ /// Creates a new instance of the object.
+ pub fn new(timestamp: Instant) -> Self {
+ Self {
+ timestamp: RwLock::new(timestamp),
+ msgs_sent: Default::default(),
+ msgs_received: Default::default(),
+ bytes_sent: Default::default(),
+ bytes_received: Default::default(),
+ failures: Default::default(),
+ }
+ }
+
+ /// Returns the creation or connection timestamp.
+ pub fn timestamp(&self) -> Instant {
+ *self.timestamp.read()
+ }
+
  /// Returns the number of sent messages and their collective size in bytes.
  pub fn sent(&self) -> (u64, u64) {
  let msgs = self.msgs_sent.load(Relaxed);