core/src/transport: Add Transport::dial_as_listener

core/src/connection.rs

@@ -96,7 +96,10 @@ pub enum PendingPoint {
     /// There is no single address associated with the Dialer of a pending
     /// connection. Addresses are dialed in parallel. Only once the first dial
     /// is successful is the address of the connection known.
-    Dialer,
+    Dialer {
+        /// Same as [`ConnectedPoint::Dialer`] `role_override`.
+        role_override: Endpoint,
+    },
     /// The socket comes from a listener.
     Listener {
         /// Local connection address.
@@ -109,7 +112,7 @@ pub enum PendingPoint {
 impl From<ConnectedPoint> for PendingPoint {
     fn from(endpoint: ConnectedPoint) -> Self {
         match endpoint {
-            ConnectedPoint::Dialer { .. } => PendingPoint::Dialer,
+            ConnectedPoint::Dialer { role_override, .. } => PendingPoint::Dialer { role_override },
             ConnectedPoint::Listener {
                 local_addr,
                 send_back_addr,
@@ -128,6 +131,27 @@ pub enum ConnectedPoint {
     Dialer {
         /// Multiaddress that was successfully dialed.
         address: Multiaddr,
+        /// Whether the role of the local node on the connection should be
+        /// overriden. I.e. whether the local node should act as a listener on
+        /// the outgoing connection.
+        ///
+        /// This option is needed for NAT and firewall hole punching.
+        ///
+        /// - [`Endpoint::Dialer`] represents the default non-overriding option.
+        ///
+        /// - [`Endpoint::Listener`] represents the overriding option.
+        ///   Realization depends on the transport protocol. E.g. in the case of
+        ///   TCP, both endpoints dial each other, resulting in a _simultaneous
+        ///   open_ TCP connection. On this new connection both endpoints assume
+        ///   to be the dialer of the connection. This is problematic during the
+        ///   connection upgrade process where an upgrade assumes one side to be
+        ///   the listener. With the help of this option, both peers can
+        ///   negotiate the roles (dialer and listener) for the new connection
+        ///   ahead of time, through some external channel, e.g. the DCUtR
+        ///   protocol, and thus have one peer dial the other and upgrade the
+        ///   connection as a dialer and one peer dial the other and upgrade the
+        ///   connection _as a listener_ overriding its role.
+        role_override: Endpoint,
     },
     /// We received the node.
     Listener {
@@ -183,7 +207,7 @@ impl ConnectedPoint {
     /// not be usable to establish new connections.
     pub fn get_remote_address(&self) -> &Multiaddr {
         match self {
-            ConnectedPoint::Dialer { address } => address,
+            ConnectedPoint::Dialer { address, .. } => address,
             ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
         }
     }
@@ -193,7 +217,7 @@ impl ConnectedPoint {
     /// For `Dialer`, this modifies `address`. For `Listener`, this modifies `send_back_addr`.
     pub fn set_remote_address(&mut self, new_address: Multiaddr) {
         match self {
-            ConnectedPoint::Dialer { address } => *address = new_address,
+            ConnectedPoint::Dialer { address, .. } => *address = new_address,
             ConnectedPoint::Listener { send_back_addr, .. } => *send_back_addr = new_address,
         }
     }

core/src/connection/listeners.rs

@@ -491,6 +491,13 @@ mod tests {
                 panic!()
             }
 
+            fn dial_with_role_override(
+                self,
+                _: Multiaddr,
+            ) -> Result<Self::Dial, transport::TransportError<Self::Error>> {
+                panic!()
+            }
+
             fn address_translation(&self, _: &Multiaddr, _: &Multiaddr) -> Option<Multiaddr> {
                 None
             }
@@ -542,6 +549,13 @@ mod tests {
                 panic!()
             }
 
+            fn dial_with_role_override(
+                self,
+                _: Multiaddr,
+            ) -> Result<Self::Dial, transport::TransportError<Self::Error>> {
+                panic!()
+            }
+
             fn address_translation(&self, _: &Multiaddr, _: &Multiaddr) -> Option<Multiaddr> {
                 None
             }

core/src/connection/pool.rs

@@ -22,8 +22,8 @@
 use crate::{
     connection::{
         handler::{THandlerError, THandlerInEvent, THandlerOutEvent},
-        Connected, ConnectionError, ConnectionHandler, ConnectionId, ConnectionLimit, IncomingInfo,
-        IntoConnectionHandler, PendingConnectionError, PendingInboundConnectionError,
+        Connected, ConnectionError, ConnectionHandler, ConnectionId, ConnectionLimit, Endpoint,
+        IncomingInfo, IntoConnectionHandler, PendingConnectionError, PendingInboundConnectionError,
         PendingOutboundConnectionError, PendingPoint, Substream,
     },
     muxing::StreamMuxer,
@@ -460,7 +460,7 @@ where
                     local_addr,
                     send_back_addr,
                 }),
-                PendingPoint::Dialer => None,
+                PendingPoint::Dialer { .. } => None,
             })
     }
 
@@ -535,6 +535,7 @@ where
         addresses: impl Iterator<Item = Multiaddr> + Send + 'static,
         peer: Option<PeerId>,
         handler: THandler,
+        role_override: Endpoint,
     ) -> Result<ConnectionId, DialError<THandler>>
     where
         TTrans: Clone + Send,
@@ -544,7 +545,13 @@ where
             return Err(DialError::ConnectionLimit { limit, handler });
         };
 
-        let dial = ConcurrentDial::new(transport, peer, addresses, self.dial_concurrency_factor);
+        let dial = ConcurrentDial::new(
+            transport,
+            peer,
+            addresses,
+            self.dial_concurrency_factor,
+            role_override,
+        );
 
         let connection_id = self.next_connection_id();
 
@@ -560,13 +567,15 @@ where
             .boxed(),
         );
 
-        self.counters.inc_pending(&PendingPoint::Dialer);
+        let endpoint = PendingPoint::Dialer { role_override };
+
+        self.counters.inc_pending(&endpoint);
         self.pending.insert(
             connection_id,
             PendingConnectionInfo {
                 peer_id: peer,
                 handler,
-                endpoint: PendingPoint::Dialer,
+                endpoint: endpoint,
                 _drop_notifier: drop_notifier,
             },
         );
@@ -739,9 +748,13 @@ where
                     self.counters.dec_pending(&endpoint);
 
                     let (endpoint, concurrent_dial_errors) = match (endpoint, outgoing) {
-                        (PendingPoint::Dialer, Some((address, errors))) => {
-                            (ConnectedPoint::Dialer { address }, Some(errors))
-                        }
+                        (PendingPoint::Dialer { role_override }, Some((address, errors))) => (
+                            ConnectedPoint::Dialer {
+                                address,
+                                role_override,
+                            },
+                            Some(errors),
+                        ),
                         (
                             PendingPoint::Listener {
                                 local_addr,
@@ -755,7 +768,7 @@ where
                             },
                             None,
                         ),
-                        (PendingPoint::Dialer, None) => unreachable!(
+                        (PendingPoint::Dialer { .. }, None) => unreachable!(
                             "Established incoming connection via pending outgoing connection."
                         ),
                         (PendingPoint::Listener { .. }, Some(_)) => unreachable!(
@@ -904,7 +917,7 @@ where
                         self.counters.dec_pending(&endpoint);
 
                         match (endpoint, error) {
-                            (PendingPoint::Dialer, Either::Left(error)) => {
+                            (PendingPoint::Dialer { .. }, Either::Left(error)) => {
                                 return Poll::Ready(PoolEvent::PendingOutboundConnectionError {
                                     id,
                                     error,
@@ -927,7 +940,7 @@ where
                                     local_addr,
                                 });
                             }
-                            (PendingPoint::Dialer, Either::Right(_)) => {
+                            (PendingPoint::Dialer { .. }, Either::Right(_)) => {
                                 unreachable!("Inbound error for outbound connection.")
                             }
                             (PendingPoint::Listener { .. }, Either::Left(_)) => {
@@ -1170,7 +1183,7 @@ impl ConnectionCounters {
 
     fn inc_pending(&mut self, endpoint: &PendingPoint) {
         match endpoint {
-            PendingPoint::Dialer => {
+            PendingPoint::Dialer { .. } => {
                 self.pending_outgoing += 1;
             }
             PendingPoint::Listener { .. } => {
@@ -1185,7 +1198,7 @@ impl ConnectionCounters {
 
     fn dec_pending(&mut self, endpoint: &PendingPoint) {
         match endpoint {
-            PendingPoint::Dialer => {
+            PendingPoint::Dialer { .. } => {
                 self.pending_outgoing -= 1;
             }
             PendingPoint::Listener { .. } => {

core/src/connection/pool/concurrent_dial.rs

@@ -18,9 +18,8 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-pub use crate::connection::{ConnectionCounters, ConnectionLimits};
-
 use crate::{
+    connection::Endpoint,
     transport::{Transport, TransportError},
     Multiaddr, PeerId,
 };
@@ -63,14 +62,23 @@ where
         peer: Option<PeerId>,
         addresses: impl Iterator<Item = Multiaddr> + Send + 'static,
         concurrency_factor: NonZeroU8,
+        role_override: Endpoint,
     ) -> Self {
         let mut pending_dials = addresses.map(move |address| match p2p_addr(peer, address) {
-            Ok(address) => match transport.clone().dial(address.clone()) {
-                Ok(fut) => fut
-                    .map(|r| (address, r.map_err(|e| TransportError::Other(e))))
-                    .boxed(),
-                Err(err) => futures::future::ready((address, Err(err))).boxed(),
-            },
+            Ok(address) => {
+                let dial = match role_override {
+                    Endpoint::Dialer => transport.clone().dial(address.clone()),
+                    Endpoint::Listener => {
+                        transport.clone().dial_with_role_override(address.clone())
+                    }
+                };
+                match dial {
+                    Ok(fut) => fut
+                        .map(|r| (address, r.map_err(|e| TransportError::Other(e))))
+                        .boxed(),
+                    Err(err) => futures::future::ready((address, Err(err))).boxed(),
+                }
+            }
             Err(address) => futures::future::ready((
                 address.clone(),
                 Err(TransportError::MultiaddrNotSupported(address)),

core/src/either.rs

@@ -529,6 +529,28 @@ where
         }
     }
 
+    fn dial_with_role_override(
+        self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>>
+    where
+        Self: Sized,
+    {
+        use TransportError::*;
+        match self {
+            EitherTransport::Left(a) => match a.dial_with_role_override(addr) {
+                Ok(connec) => Ok(EitherFuture::First(connec)),
+                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
+                Err(Other(err)) => Err(Other(EitherError::A(err))),
+            },
+            EitherTransport::Right(b) => match b.dial_with_role_override(addr) {
+                Ok(connec) => Ok(EitherFuture::Second(connec)),
+                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
+                Err(Other(err)) => Err(Other(EitherError::B(err))),
+            },
+        }
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         match self {
             EitherTransport::Left(a) => a.address_translation(server, observed),

core/src/network.rs

@@ -21,7 +21,7 @@
 mod event;
 pub mod peer;
 
-pub use crate::connection::{ConnectionCounters, ConnectionLimits};
+pub use crate::connection::{ConnectionCounters, ConnectionLimits, Endpoint};
 pub use event::{IncomingConnection, NetworkEvent};
 pub use peer::Peer;
 
@@ -95,7 +95,7 @@ where
         self.pool
             .iter_pending_info()
             .filter(move |(_, endpoint, peer_id)| {
-                matches!(endpoint, PendingPoint::Dialer) && peer_id.as_ref() == Some(&peer)
+                matches!(endpoint, PendingPoint::Dialer { .. }) && peer_id.as_ref() == Some(&peer)
             })
             .map(|(connection_id, _, _)| connection_id)
     }
@@ -195,6 +195,7 @@ where
         &mut self,
         address: &Multiaddr,
         handler: THandler,
+        role_override: Endpoint,
     ) -> Result<ConnectionId, DialError<THandler>>
     where
         TTrans: Transport + Send,
@@ -213,6 +214,7 @@ where
                     peer,
                     addresses: std::iter::once(address.clone()),
                     handler,
+                    role_override: Endpoint::Dialer,
                 });
             }
         }
@@ -222,6 +224,7 @@ where
             std::iter::once(address.clone()),
             None,
             handler,
+            role_override,
         )
     }
 
@@ -242,6 +245,7 @@ where
             opts.addresses,
             Some(opts.peer),
             opts.handler,
+            opts.role_override,
         )?;
 
         Ok(id)
@@ -279,7 +283,7 @@ where
     pub fn dialing_peers(&self) -> impl Iterator<Item = &PeerId> {
         self.pool
             .iter_pending_info()
-            .filter(|(_, endpoint, _)| matches!(endpoint, PendingPoint::Dialer))
+            .filter(|(_, endpoint, _)| matches!(endpoint, PendingPoint::Dialer { .. }))
             .filter_map(|(_, _, peer)| peer.as_ref())
     }
 
@@ -469,6 +473,7 @@ struct DialingOpts<THandler, I> {
     peer: PeerId,
     handler: THandler,
     addresses: I,
+    role_override: Endpoint,
 }
 
 /// Information about the network obtained by [`Network::info()`].

core/src/network/peer.rs

@@ -21,7 +21,7 @@
 use super::{DialError, DialingOpts, Network};
 use crate::{
     connection::{
-        handler::THandlerInEvent, pool::Pool, ConnectionHandler, ConnectionId,
+        handler::THandlerInEvent, pool::Pool, ConnectionHandler, ConnectionId, Endpoint,
         EstablishedConnection, EstablishedConnectionIter, IntoConnectionHandler, PendingConnection,
     },
     Multiaddr, PeerId, Transport,
@@ -151,15 +151,13 @@ where
 
     /// Initiates a new dialing attempt to this peer using the given addresses.
     ///
-    /// The connection ID of the first connection attempt, i.e. to `address`,
-    /// is returned, together with a [`DialingPeer`] for further use. The
-    /// `remaining` addresses are tried in order in subsequent connection
-    /// attempts in the context of the same dialing attempt, if the connection
-    /// attempt to the first address fails.
+    /// The [`ConnectionId`] of the connection attempt is returned together with
+    /// a [`DialingPeer`] for further use.
     pub fn dial<I>(
         self,
         addresses: I,
         handler: THandler,
+        role_override: Endpoint,
     ) -> Result<(ConnectionId, DialingPeer<'a, TTrans, THandler>), DialError<THandler>>
     where
         I: IntoIterator<Item = Multiaddr>,
@@ -176,6 +174,7 @@ where
             peer: peer_id,
             handler,
             addresses: addresses.into_iter(),
+            role_override,
         })?;
 
         Ok((id, DialingPeer { network, peer_id }))