p2p: limit the number of pending inbound connections

Author: ImplOfAnImpl

CHANGELOG.md

@@ -35,6 +35,8 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
       - Adding an already existing relayable local transaction to the mempool will cause p2p to re-announce it.
       - Transactions are now announced in batches at irregular intervals (previously, a random delay was added
         before each individual transaction announcement).
+    - Security improvements:
+      - The number of pending inbound connections is now limited.
 
   - Mempool:
     - Various optimizations were made.

networking/src/transport/impls/stream_adapter/wrapped_transport/wrapped_listener.rs

@@ -21,6 +21,8 @@ use futures::{
     stream::{FuturesUnordered, StreamExt},
 };
 
+use logging::log;
+
 use crate::{
     Result,
     transport::{TransportListener, TransportSocket, impls::stream_adapter::traits::StreamAdapter},
@@ -63,7 +65,10 @@ impl<S: StreamAdapter<T::Stream>, T: TransportSocket> TransportListener for Adap
                     match handshake_res {
                         (Ok(handshake), addr) => return Ok((handshake, addr)),
                         (Err(err), addr) => {
-                            logging::log::warn!("Handshake with {addr} failed: {err}");
+                            // Note: failed transport handshakes are peer-triggerable. In particular,
+                            // a malicious peer can deliberately and repeatedly fail the Noise handshake,
+                            // so we keep this at debug level to avoid flooding default logs.
+                            log::debug!("Handshake with {addr} failed: {err}");
                             continue;
                         },
                     }
@@ -80,7 +85,12 @@ impl<S: StreamAdapter<T::Stream>, T: TransportSocket> TransportListener for Adap
                             self.handshakes.push(handshake_with_addr);
                         },
                         Err(err) => {
-                            logging::log::error!("Accept failed unexpectedly: {err}");
+                            // Note: this can also be peer-triggered, though less reliably than the
+                            // handshake error above. E.g. a malicious peer may open TCP connections
+                            // and close them immediately, attempting to flood the node's logs.
+                            // But in any case, this error is propagated to the p2p backend, which logs
+                            // it again, so there is no point in using a level higher than debug here.
+                            log::debug!("Accept failed unexpectedly: {err}");
                             return Err(err);
                         },
                     }

node-lib/src/config_files/p2p.rs

@@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize};
 use common::primitives::{semver::SemVer, user_agent::mintlayer_core_user_agent};
 use p2p::{
     ban_config::BanConfig,
-    config::{BackendTimeoutsConfig, NodeType, P2pConfig},
+    config::{BackendConfig, NodeType, P2pConfig},
     peer_manager::config::PeerManagerConfig,
 };
 use utils_networking::IpOrSocketAddress;
@@ -193,6 +193,7 @@ impl From<P2pConfigFile> for P2pConfig {
                 peer_handshake_timeout: Default::default(),
                 disconnection_timeout: Default::default(),
                 socket_write_timeout: Default::default(),
+                max_pending_inbound_connections: Default::default(),
             },
             protocol_config: Default::default(),
             custom_disconnection_reason_for_banning,

p2p/src/config.rs

@@ -37,6 +37,7 @@ make_config_setting!(SyncStallingTimeout, Duration, Duration::from_secs(25));
 make_config_setting!(PeerHandshakeTimeout, Duration, Duration::from_secs(10));
 make_config_setting!(DisconnectionTimeout, Duration, Duration::from_secs(10));
 make_config_setting!(SoketWriteTimeout, Duration, Duration::from_secs(60));
+make_config_setting!(MaxPendingInboundConnections, usize, 100);
 
 /// A node type.
 #[derive(Debug, Copy, Clone)]
@@ -144,12 +145,21 @@ pub struct BackendConfig {
     /// The outbound connection timeout value.
     pub outbound_connection_timeout: OutboundConnectionTimeout,
 
-    /// Timeout for initial peer handshake
+    /// Timeout for initial peer handshake.
     pub peer_handshake_timeout: PeerHandshakeTimeout,
 
-    /// Timeout for disconnection
+    /// Timeout for disconnection.
     pub disconnection_timeout: DisconnectionTimeout,
 
-    /// Timeout for the socket write call
+    /// Timeout for the socket write call.
     pub socket_write_timeout: SoketWriteTimeout,
+
+    /// The maximum number of pending inbound connections that can exist at the same time.
+    ///
+    /// Note: the presence of this limit is important for security, but its specific value is not -
+    /// it just defines backend's burst tolerance for transport-accepted inbound connections that
+    /// have not completed the P2P handshake yet. But it makes sense to keep it in the same order
+    /// as `MAX_CONCURRENT_HANDSHAKES` used by wrapped transport layer's `AdaptedListener`, which
+    /// limits the number of simultaneous transport-level inbound handshakes.
+    pub max_pending_inbound_connections: MaxPendingInboundConnections,
 }

p2p/src/net/default_backend/backend.rs

@@ -39,8 +39,8 @@ use networking::{
 use p2p_types::socket_address::SocketAddress;
 use randomness::{RngExt as _, make_pseudo_rng};
 use utils::{
-    atomics::SeqCstAtomicBool, eventhandler::EventsController, set_flag::SetFlag,
-    shallow_clone::ShallowClone, tokio_spawn_in_current_tracing_span,
+    atomics::SeqCstAtomicBool, debug_panic_or_log, eventhandler::EventsController,
+    set_flag::SetFlag, shallow_clone::ShallowClone, tokio_spawn_in_current_tracing_span,
 };
 
 use crate::{
@@ -108,6 +108,7 @@ struct PeerContext {
 }
 
 /// Pending peer data (until handshake message is received)
+#[derive(Debug)]
 struct PendingPeerContext {
     handle: tokio::task::JoinHandle<()>,
 
@@ -149,6 +150,9 @@ pub struct Backend<T: TransportSocket> {
     /// Pending connections
     pending_peers: HashMap<PeerId, PendingPeerContext>,
 
+    /// The number of inbound peers in `pending_peers`
+    pending_inbound_peer_count: usize,
+
     /// Map of streams for receiving events from peers.
     peer_event_stream_map: StreamMap<PeerId, ReceiverStream<PeerEvent>>,
 
@@ -210,6 +214,7 @@ where
             syncing_event_sender,
             peers: HashMap::new(),
             pending_peers: HashMap::new(),
+            pending_inbound_peer_count: 0,
             peer_event_stream_map: StreamMap::new(),
             command_queue: FuturesUnordered::new(),
             shutdown,
@@ -347,21 +352,51 @@ where
                 res = self.socket.accept() => {
                     match res {
                         Ok((stream, address)) => {
+                            // Note: this execution path is peer-triggerable, so we use debug level logging
+                            // until after `pending_inbound_peer_count` has been checked, to prevent malicious
+                            // peers from flooding the default logs.
                             if !self.networking_enabled {
-                                log::info!("Ignoring incoming connection from {address:?} because networking is disabled");
+                                log::debug!("Ignoring incoming connection from {address:?} because networking is disabled");
                             } else {
-                                self.create_pending_peer(
-                                    stream,
-                                    PeerId::new(),
-                                    ConnectionInfo::Inbound,
-                                    address.into(),
-                                )?;
+                                let max_pending_inbound_conn_count = *self.p2p_config.backend_config.max_pending_inbound_connections;
+
+                                if self.pending_inbound_peer_count >= max_pending_inbound_conn_count {
+                                    log::debug!(
+                                        concat!(
+                                            "Ignoring incoming connection from {:?} because the maximum number ",
+                                            "of pending incoming connections has been reached {}"
+                                        ),
+                                        address,
+                                        max_pending_inbound_conn_count
+                                    );
+                                } else {
+                                    self.create_pending_peer(
+                                        stream,
+                                        PeerId::new(),
+                                        ConnectionInfo::Inbound,
+                                        address.into(),
+                                    )?;
+                                }
                             }
                         },
                         Err(err) => {
-                            // Just log the error and let the node continue working
+                            // Note: this execution path is also peer-triggerable, though less reliably than the
+                            // successful path above, so we use debug-level logging here too. See also a similar
+                            // comment in `AdaptedListener::accept` in the transport layer.
+                            // TODO: malicious peers can still flood the default logs making successful connections
+                            // and dropping them immediately (though it will be less severe and more costly for the
+                            // attacker). Consider:
+                            // a) Using debug-level logging specifically for incoming connections (while keeping it
+                            //    at "info" for outbound ones) in all (or most) connectivity-related messages (e.g.
+                            //    "Peer disconnected", "Assigning peer id", "New peer accepted").
+                            // b) Implementing a rate-limiter for the log (e.g. using a custom tracing layer or filter),
+                            //    either based on the source code location (Bitcoin does this) or on the specified
+                            //    log target.
+                            // c) (In the case when some errors have been omitted or printed via debug!), printing
+                            //    some kind of info/warn-level summary at regular intervals, e.g. "X incoming connections
+                            //    ignored in the last Y seconds".
                             if self.networking_enabled {
-                                log::error!("Accepting a new connection failed unexpectedly: {err}")
+                                log::debug!("Accepting a new connection failed unexpectedly: {err}")
                             } else {
                                 log::debug!(
                                     "Ignoring failed incoming connection because networking is disabled (err = {err})",
@@ -426,7 +461,7 @@ where
             &format!("Peer[id={peer_id}]"),
         );
 
-        self.pending_peers.insert(
+        self.insert_new_pending_peer(
             peer_id,
             PendingPeerContext {
                 handle,
@@ -456,7 +491,7 @@ where
             bind_address,
             connection_info,
             backend_event_sender,
-        } = match self.pending_peers.remove(&peer_id) {
+        } = match self.remove_pending_peer(peer_id) {
             Some(pending_peer) => pending_peer,
             // Could be removed if self-connection was detected earlier
             None => return Ok(()),
@@ -568,7 +603,7 @@ where
                 .map(|(peer_id, _pending)| *peer_id);
 
             if let Some(peer_id) = pending_outbound_peer_id {
-                let peer_ctx = self.pending_peers.remove(&peer_id).expect("peer must exist");
+                let peer_ctx = self.remove_pending_peer(peer_id).expect("peer must exist");
 
                 log::info!(
                     "self-connection detected on address {:?}",
@@ -649,7 +684,7 @@ where
             }
 
             PeerEvent::ConnectionClosed => {
-                if let Some(pending_peer) = self.pending_peers.remove(&peer_id) {
+                if let Some(pending_peer) = self.remove_pending_peer(peer_id) {
                     // Note: we'll get here if handshake has failed, so no need to use log levels
                     // higher that debug, because the error should have been logged properly already.
                     match pending_peer.connection_info {
@@ -802,6 +837,57 @@ where
             Err(_) => log::error!("sending syncing event from the backend failed unexpectedly"),
         }
     }
+
+    fn insert_new_pending_peer(&mut self, peer_id: PeerId, peer_context: PendingPeerContext) {
+        let is_inbound = peer_context.connection_info.is_inbound();
+
+        if let Some(old_context) = self.pending_peers.insert(peer_id, peer_context) {
+            debug_panic_or_log!(
+                "Pending peer context already exists for a new peer {peer_id}: {old_context:?}"
+            );
+        } else if is_inbound {
+            self.pending_inbound_peer_count += 1;
+
+            #[cfg(test)]
+            self.assert_pending_inbound_peer_count_consistency();
+        }
+
+        if let Some(observer) = &self.observer {
+            observer.on_pending_peer_created(peer_id);
+        }
+    }
+
+    fn remove_pending_peer(&mut self, peer_id: PeerId) -> Option<PendingPeerContext> {
+        let peer_context = self.pending_peers.remove(&peer_id);
+
+        if let Some(peer_context) = &peer_context {
+            if peer_context.connection_info.is_inbound() {
+                self.pending_inbound_peer_count -= 1;
+
+                #[cfg(test)]
+                self.assert_pending_inbound_peer_count_consistency();
+            }
+
+            if let Some(observer) = &self.observer {
+                observer.on_pending_peer_removed(peer_id);
+            }
+        }
+
+        peer_context
+    }
+
+    #[cfg(test)]
+    fn assert_pending_inbound_peer_count_consistency(&self) {
+        let actual_pending_inbound_peer_count = self
+            .pending_peers
+            .values()
+            .filter(|context| context.connection_info.is_inbound())
+            .count();
+        assert_eq!(
+            self.pending_inbound_peer_count,
+            actual_pending_inbound_peer_count
+        );
+    }
 }
 
 // Some boilerplate types and a function for blocking tasks handling

p2p/src/net/default_backend/peer/mod.rs

@@ -73,6 +73,13 @@ impl ConnectionInfo {
             ConnectionInfo::Outbound { .. } => ConnectionDirection::Outbound,
         }
     }
+
+    pub fn is_inbound(&self) -> bool {
+        match self {
+            ConnectionInfo::Inbound => true,
+            ConnectionInfo::Outbound { .. } => false,
+        }
+    }
 }
 
 pub struct Peer<T: TransportSocket> {

p2p/src/net/default_backend/types.rs

@@ -417,6 +417,15 @@ pub trait BackendObserver {
 
     /// Called after the message has been read from the socket.
     fn on_message_read(&self, peer_id: PeerId, msg: &Message);
+
+    /// Called after a new peer has been created (i.e. after the peer task has been spawned and
+    /// `PendingPeerContext` added to the corresponding collection).
+    fn on_pending_peer_created(&self, peer_id: PeerId);
+
+    /// Called when a `PendingPeerContext` has been removed from the corresponding collection
+    /// (i.e. either the peer has completed the handshake successfully and is no longer pending,
+    /// or it failed the handshake and got disconnected).
+    fn on_pending_peer_removed(&self, peer_id: PeerId);
 }
 
 #[cfg(test)]

p2p/src/peer_manager/tests/connections.rs

@@ -1009,6 +1009,7 @@ async fn connection_timeout_rpc_notified<T>(
             peer_handshake_timeout: Default::default(),
             disconnection_timeout: Default::default(),
             socket_write_timeout: Default::default(),
+            max_pending_inbound_connections: Default::default(),
         },
 
         bind_addresses: Default::default(),

p2p/src/sync/tests/network_sync.rs

@@ -434,6 +434,7 @@ async fn send_block_from_the_future_again(#[case] seed: Seed) {
                 outbound_connection_timeout: Default::default(),
                 disconnection_timeout: Default::default(),
                 socket_write_timeout: Default::default(),
+                max_pending_inbound_connections: Default::default(),
             },
 
             bind_addresses: Default::default(),

p2p/src/tests/connection_lockup_when_socket_not_read.rs

@@ -37,7 +37,7 @@ use test_utils::{
 };
 
 use crate::{
-    config::{BackendTimeoutsConfig, P2pConfig},
+    config::{BackendConfig, P2pConfig},
     message::{HeaderList, HeaderListRequest},
     net::{
         default_backend::types::{HandshakeMessage, Message, MessageTag, P2pTimestamp},
@@ -291,6 +291,7 @@ async fn timeout_when_socket_not_read(
 
                 outbound_connection_timeout: Default::default(),
                 peer_handshake_timeout: Default::default(),
+                max_pending_inbound_connections: Default::default(),
             },
 
             bind_addresses: Default::default(),