feat(swarm)!: allow NetworkBehaviours to manage connections

Previously, a `ConnectionHandler` was immediately requested from the `NetworkBehaviour` as soon as a new dial was initiated or a new incoming connection accepted.

With this patch, we delay the creation of the handler until the connection is actually established and fully upgraded, i.e authenticated and multiplexed.

As a consequence, `NetworkBehaviour::new_handler` is now deprecated in favor of a new set of callbacks:

- `NetworkBehaviour::handle_pending_inbound_connection`
- `NetworkBehaviour::handle_pending_outbound_connection`
- `NetworkBehaviour::handle_established_inbound_connection`
- `NetworkBehaviour::handle_established_outbound_connection`

All callbacks are fallible, allowing the `NetworkBehaviour` to abort the connection either immediately or after it is fully established. All callbacks also receive a `ConnectionId` parameter which uniquely identifies the connection. For example, in case a `NetworkBehaviour` issues a dial via `NetworkBehaviourAction::Dial`, it can unambiguously detect this dial in these lifecycle callbacks via the `ConnectionId`.

Finally, `NetworkBehaviour::handle_pending_outbound_connection` also replaces `NetworkBehaviour::addresses_of_peer` by allowing the behaviour to return more addresses to be used for the dial.

Resolves #2824.

Pull-Request: #3254.

Author: thomaseizinger

CHANGELOG.md

@@ -47,6 +47,11 @@
 
 # 0.51.0 [unreleased]
 
+- Enable `NetworkBehaviour`s to manage connections.
+  This deprecates `NetworkBehaviour::new_handler` and `NetworkBehaviour::addresses_of_peer`.
+  Due to limitations in the Rust compiler, these deprecations may not show up for you, nevertheless they will be removed in a future release.
+  See [`libp2p-swarm`'s CHANGELOG](swarm/CHANGELOG.md#0420) for details.
+
 - Count bandwidth at the application level. Previously `BandwidthLogging` would implement `Transport` and now implements `StreamMuxer` ([PR 3180](https://github.com/libp2p/rust-libp2p/pull/3180)).
     - `BandwidthLogging::new` now requires a 2nd argument: `Arc<BandwidthSinks>`
     - Remove `BandwidthFuture`

misc/metrics/src/swarm.rs

@@ -244,6 +244,9 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
                     libp2p_swarm::DialError::WrongPeerId { .. } => {
                         record(OutgoingConnectionError::WrongPeerId)
                     }
+                    libp2p_swarm::DialError::Denied { .. } => {
+                        record(OutgoingConnectionError::Denied)
+                    }
                 };
             }
             libp2p_swarm::SwarmEvent::BannedPeer { endpoint, .. } => {
@@ -344,6 +347,7 @@ enum OutgoingConnectionError {
     WrongPeerId,
     TransportMultiaddrNotSupported,
     TransportOther,
+    Denied,
 }
 
 #[derive(EncodeLabelSet, Hash, Clone, Eq, PartialEq, Debug)]
@@ -360,6 +364,7 @@ enum IncomingConnectionError {
     TransportErrorOther,
     Aborted,
     ConnectionLimit,
+    Denied,
 }
 
 impl From<&libp2p_swarm::ListenError> for IncomingConnectionError {
@@ -377,6 +382,7 @@ impl From<&libp2p_swarm::ListenError> for IncomingConnectionError {
                 libp2p_core::transport::TransportError::Other(_),
             ) => IncomingConnectionError::TransportErrorOther,
             libp2p_swarm::ListenError::Aborted => IncomingConnectionError::Aborted,
+            libp2p_swarm::ListenError::Denied { .. } => IncomingConnectionError::Denied,
         }
     }
 }

protocols/autonat/src/behaviour.rs

@@ -37,8 +37,8 @@ use libp2p_swarm::{
         AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredExternalAddr,
         ExpiredListenAddr, FromSwarm,
     },
-    ConnectionId, ExternalAddresses, ListenAddresses, NetworkBehaviour, NetworkBehaviourAction,
-    PollParameters, THandlerInEvent, THandlerOutEvent,
+    ConnectionDenied, ConnectionId, ExternalAddresses, ListenAddresses, NetworkBehaviour,
+    NetworkBehaviourAction, PollParameters, THandler, THandlerInEvent, THandlerOutEvent,
 };
 use std::{
     collections::{HashMap, VecDeque},
@@ -485,12 +485,55 @@ impl NetworkBehaviour for Behaviour {
         }
     }
 
-    fn new_handler(&mut self) -> Self::ConnectionHandler {
-        self.inner.new_handler()
+    fn handle_pending_inbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        local_addr: &Multiaddr,
+        remote_addr: &Multiaddr,
+    ) -> Result<(), ConnectionDenied> {
+        self.inner
+            .handle_pending_inbound_connection(connection_id, local_addr, remote_addr)
+    }
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _connection_id: ConnectionId,
+        peer: PeerId,
+        local_addr: &Multiaddr,
+        remote_addr: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        self.inner.handle_established_inbound_connection(
+            _connection_id,
+            peer,
+            local_addr,
+            remote_addr,
+        )
     }
 
-    fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec<Multiaddr> {
-        self.inner.addresses_of_peer(peer)
+    fn handle_pending_outbound_connection(
+        &mut self,
+        _connection_id: ConnectionId,
+        maybe_peer: Option<PeerId>,
+        _addresses: &[Multiaddr],
+        _effective_role: Endpoint,
+    ) -> Result<Vec<Multiaddr>, ConnectionDenied> {
+        self.inner.handle_pending_outbound_connection(
+            _connection_id,
+            maybe_peer,
+            _addresses,
+            _effective_role,
+        )
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _connection_id: ConnectionId,
+        peer: PeerId,
+        addr: &Multiaddr,
+        role_override: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        self.inner
+            .handle_established_outbound_connection(_connection_id, peer, addr, role_override)
     }
 
     fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {

protocols/dcutr/src/behaviour_impl.rs

@@ -24,14 +24,14 @@ use crate::handler;
 use either::Either;
 use libp2p_core::connection::ConnectedPoint;
 use libp2p_core::multiaddr::Protocol;
-use libp2p_core::{Multiaddr, PeerId};
+use libp2p_core::{Endpoint, Multiaddr, PeerId};
 use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
 use libp2p_swarm::dial_opts::{self, DialOpts};
+use libp2p_swarm::{dummy, ConnectionDenied, ConnectionId, THandler, THandlerOutEvent};
 use libp2p_swarm::{
     ConnectionHandlerUpgrErr, ExternalAddresses, NetworkBehaviour, NetworkBehaviourAction,
     NotifyHandler, PollParameters, THandlerInEvent,
 };
-use libp2p_swarm::{ConnectionId, THandlerOutEvent};
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::task::{Context, Poll};
 use thiserror::Error;
@@ -64,12 +64,14 @@ pub enum Error {
     #[error("Failed to dial peer.")]
     Dial,
     #[error("Failed to establish substream: {0}.")]
-    Handler(ConnectionHandlerUpgrErr<void::Void>),
+    Handler(ConnectionHandlerUpgrErr<Void>),
 }
 
 pub struct Behaviour {
     /// Queue of actions to return when polled.
-    queued_events: VecDeque<NetworkBehaviourAction<Event, Either<handler::relayed::Command, Void>>>,
+    queued_events: VecDeque<
+        NetworkBehaviourAction<Event, Either<handler::relayed::Command, Either<Void, Void>>>,
+    >,
 
     /// All direct (non-relayed) connections.
     direct_connections: HashMap<PeerId, HashSet<ConnectionId>>,
@@ -237,11 +239,82 @@ impl Behaviour {
 }
 
 impl NetworkBehaviour for Behaviour {
-    type ConnectionHandler = handler::Prototype;
+    type ConnectionHandler = Either<
+        handler::relayed::Handler,
+        Either<handler::direct::Handler, dummy::ConnectionHandler>,
+    >;
     type OutEvent = Event;
 
-    fn new_handler(&mut self) -> Self::ConnectionHandler {
-        handler::Prototype
+    fn handle_established_inbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        local_addr: &Multiaddr,
+        remote_addr: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        match self
+            .outgoing_direct_connection_attempts
+            .remove(&(connection_id, peer))
+        {
+            None => {
+                let handler = if is_relayed(local_addr) {
+                    Either::Left(handler::relayed::Handler::new(ConnectedPoint::Listener {
+                        local_addr: local_addr.clone(),
+                        send_back_addr: remote_addr.clone(),
+                    })) // TODO: We could make two `handler::relayed::Handler` here, one inbound one outbound.
+                } else {
+                    Either::Right(Either::Right(dummy::ConnectionHandler))
+                };
+
+                Ok(handler)
+            }
+            Some(_) => {
+                assert!(
+                    !is_relayed(local_addr),
+                    "`Prototype::DirectConnection` is never created for relayed connection."
+                );
+
+                Ok(Either::Right(Either::Left(
+                    handler::direct::Handler::default(),
+                )))
+            }
+        }
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        addr: &Multiaddr,
+        role_override: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        match self
+            .outgoing_direct_connection_attempts
+            .remove(&(connection_id, peer))
+        {
+            None => {
+                let handler = if is_relayed(addr) {
+                    Either::Left(handler::relayed::Handler::new(ConnectedPoint::Dialer {
+                        address: addr.clone(),
+                        role_override,
+                    })) // TODO: We could make two `handler::relayed::Handler` here, one inbound one outbound.
+                } else {
+                    Either::Right(Either::Right(dummy::ConnectionHandler))
+                };
+
+                Ok(handler)
+            }
+            Some(_) => {
+                assert!(
+                    !is_relayed(addr),
+                    "`Prototype::DirectConnection` is never created for relayed connection."
+                );
+
+                Ok(Either::Right(Either::Left(
+                    handler::direct::Handler::default(),
+                )))
+            }
+        }
     }
 
     fn on_connection_handler_event(
@@ -332,7 +405,7 @@ impl NetworkBehaviour for Behaviour {
                 self.queued_events
                     .push_back(NetworkBehaviourAction::Dial { opts });
             }
-            Either::Right(handler::direct::Event::DirectConnectionEstablished) => {
+            Either::Right(Either::Left(handler::direct::Event::DirectConnectionEstablished)) => {
                 self.queued_events.extend([
                     NetworkBehaviourAction::NotifyHandler {
                         peer_id: event_source,
@@ -348,6 +421,7 @@ impl NetworkBehaviour for Behaviour {
                     ),
                 ]);
             }
+            Either::Right(Either::Right(never)) => void::unreachable(never),
         };
     }
 
@@ -386,3 +460,7 @@ impl NetworkBehaviour for Behaviour {
         }
     }
 }
+
+fn is_relayed(addr: &Multiaddr) -> bool {
+    addr.iter().any(|p| p == Protocol::P2pCircuit)
+}

protocols/dcutr/src/handler.rs

@@ -18,29 +18,5 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::protocol;
-use either::Either;
-use libp2p_core::{ConnectedPoint, PeerId};
-use libp2p_swarm::handler::SendWrapper;
-use libp2p_swarm::{ConnectionHandler, IntoConnectionHandler};
-
 pub mod direct;
 pub mod relayed;
-
-pub struct Prototype;
-
-impl IntoConnectionHandler for Prototype {
-    type Handler = Either<relayed::Handler, direct::Handler>;
-
-    fn into_handler(self, _remote_peer_id: &PeerId, endpoint: &ConnectedPoint) -> Self::Handler {
-        if endpoint.is_relayed() {
-            Either::Left(relayed::Handler::new(endpoint.clone()))
-        } else {
-            Either::Right(direct::Handler::default()) // This is a direct connection. What we don't know is whether it is the one we created or another one that happened accidentally.
-        }
-    }
-
-    fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
-        Either::Left(SendWrapper(Either::Left(protocol::inbound::Upgrade {})))
-    }
-}

protocols/floodsub/src/layer.rs

@@ -26,11 +26,11 @@ use crate::topic::Topic;
 use crate::FloodsubConfig;
 use cuckoofilter::{CuckooError, CuckooFilter};
 use fnv::FnvHashSet;
-use libp2p_core::PeerId;
+use libp2p_core::{Endpoint, Multiaddr, PeerId};
 use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, FromSwarm};
 use libp2p_swarm::{
-    dial_opts::DialOpts, ConnectionId, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
-    OneShotHandler, PollParameters, THandlerInEvent, THandlerOutEvent,
+    dial_opts::DialOpts, ConnectionDenied, ConnectionId, NetworkBehaviour, NetworkBehaviourAction,
+    NotifyHandler, OneShotHandler, PollParameters, THandler, THandlerInEvent, THandlerOutEvent,
 };
 use log::warn;
 use smallvec::SmallVec;
@@ -334,8 +334,24 @@ impl NetworkBehaviour for Floodsub {
     type ConnectionHandler = OneShotHandler<FloodsubProtocol, FloodsubRpc, InnerMessage>;
     type OutEvent = FloodsubEvent;
 
-    fn new_handler(&mut self) -> Self::ConnectionHandler {
-        Default::default()
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Default::default())
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Default::default())
     }
 
     fn on_connection_handler_event(

protocols/gossipsub/src/behaviour.rs

@@ -36,13 +36,14 @@ use prost::Message as _;
 use rand::{seq::SliceRandom, thread_rng};
 
 use libp2p_core::{
-    identity::Keypair, multiaddr::Protocol::Ip4, multiaddr::Protocol::Ip6, Multiaddr, PeerId,
+    identity::Keypair, multiaddr::Protocol::Ip4, multiaddr::Protocol::Ip6, Endpoint, Multiaddr,
+    PeerId,
 };
 use libp2p_swarm::{
     behaviour::{AddressChange, ConnectionClosed, ConnectionEstablished, FromSwarm},
     dial_opts::DialOpts,
-    ConnectionId, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
-    THandlerInEvent, THandlerOutEvent,
+    ConnectionDenied, ConnectionId, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
+    PollParameters, THandler, THandlerInEvent, THandlerOutEvent,
 };
 use wasm_timer::Instant;
 
@@ -3289,11 +3290,30 @@ where
     type ConnectionHandler = Handler;
     type OutEvent = Event;
 
-    fn new_handler(&mut self) -> Self::ConnectionHandler {
-        Handler::new(
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(
             ProtocolConfig::new(&self.config),
             self.config.idle_timeout(),
-        )
+        ))
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(
+            ProtocolConfig::new(&self.config),
+            self.config.idle_timeout(),
+        ))
     }
 
     fn on_connection_handler_event(