feat(transport): Implement dial_with_new_port to prevent port reuse

core/src/either.rs

@@ -206,6 +206,25 @@ where
         }
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        use TransportError::*;
+        match self {
+            Either::Left(a) => match a.dial_with_new_port(addr) {
+                Ok(connec) => Ok(EitherFuture::First(connec)),
+                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
+                Err(Other(err)) => Err(Other(Either::Left(err))),
+            },
+            Either::Right(b) => match b.dial_with_new_port(addr) {
+                Ok(connec) => Ok(EitherFuture::Second(connec)),
+                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
+                Err(Other(err)) => Err(Other(Either::Right(err))),
+            },
+        }
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         match self {
             Either::Left(a) => a.address_translation(server, observed),

core/src/transport.rs

@@ -134,6 +134,14 @@ pub trait Transport {
         addr: Multiaddr,
     ) -> Result<Self::Dial, TransportError<Self::Error>>;
 
+    ///
+    fn dial_with_new_port(
+        &mut self,
+        _addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        unimplemented!()
+    }
+
     /// Poll for [`TransportEvent`]s.
     ///
     /// A [`TransportEvent::Incoming`] should be produced whenever a connection is received at the lowest

core/src/transport/and_then.rs

@@ -105,6 +105,28 @@ where
         Ok(future)
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let dialed_fut = self
+            .transport
+            .dial_with_new_port(addr.clone())
+            .map_err(|err| err.map(Either::Left))?;
+        let future = AndThenFuture {
+            inner: Either::Left(Box::pin(dialed_fut)),
+            args: Some((
+                self.fun.clone(),
+                ConnectedPoint::Dialer {
+                    address: addr,
+                    role_override: Endpoint::Dialer,
+                },
+            )),
+            _marker: PhantomPinned,
+        };
+        Ok(future)
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.transport.address_translation(server, observed)
     }

core/src/transport/boxed.rs

@@ -56,6 +56,8 @@ trait Abstract<O> {
     fn remove_listener(&mut self, id: ListenerId) -> bool;
     fn dial(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
     fn dial_as_listener(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
+    fn dial_with_new_port(&mut self, addr: Multiaddr)
+        -> Result<Dial<O>, TransportError<io::Error>>;
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr>;
     fn poll(
         self: Pin<&mut Self>,
@@ -92,6 +94,16 @@ where
         Ok(Box::pin(fut) as Dial<_>)
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Dial<O>, TransportError<io::Error>> {
+        let fut = Transport::dial_with_new_port(self, addr)
+            .map(|r| r.map_err(box_err))
+            .map_err(|e| e.map(box_err))?;
+        Ok(Box::pin(fut) as Dial<_>)
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         Transport::address_translation(self, server, observed)
     }
@@ -142,6 +154,13 @@ impl<O> Transport for Boxed<O> {
         self.inner.dial_as_listener(addr)
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        self.inner.dial_with_new_port(addr)
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.inner.address_translation(server, observed)
     }

core/src/transport/choice.rs

@@ -107,6 +107,29 @@ where
         Err(TransportError::MultiaddrNotSupported(addr))
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let addr = match self.0.dial_with_new_port(addr) {
+            Ok(connec) => return Ok(EitherFuture::First(connec)),
+            Err(TransportError::MultiaddrNotSupported(addr)) => addr,
+            Err(TransportError::Other(err)) => {
+                return Err(TransportError::Other(Either::Left(err)))
+            }
+        };
+
+        let addr = match self.1.dial_with_new_port(addr) {
+            Ok(connec) => return Ok(EitherFuture::Second(connec)),
+            Err(TransportError::MultiaddrNotSupported(addr)) => addr,
+            Err(TransportError::Other(err)) => {
+                return Err(TransportError::Other(Either::Right(err)))
+            }
+        };
+
+        Err(TransportError::MultiaddrNotSupported(addr))
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         if let Some(addr) = self.0.address_translation(server, observed) {
             Some(addr)

core/src/transport/global_only.rs

@@ -333,6 +333,32 @@ impl<T: crate::Transport + Unpin> crate::Transport for Transport<T> {
         }
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        match addr.iter().next() {
+            Some(Protocol::Ip4(a)) => {
+                if !ipv4_global::is_global(a) {
+                    debug!("Not dialing non global IP address {:?}.", a);
+                    return Err(TransportError::MultiaddrNotSupported(addr));
+                }
+                self.inner.dial_with_new_port(addr)
+            }
+            Some(Protocol::Ip6(a)) => {
+                if !ipv6_global::is_global(a) {
+                    debug!("Not dialing non global IP address {:?}.", a);
+                    return Err(TransportError::MultiaddrNotSupported(addr));
+                }
+                self.inner.dial_with_new_port(addr)
+            }
+            _ => {
+                debug!("Not dialing unsupported Multiaddress {:?}.", addr);
+                Err(TransportError::MultiaddrNotSupported(addr))
+            }
+        }
+    }
+
     fn address_translation(&self, listen: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.inner.address_translation(listen, observed)
     }

core/src/transport/map.rs

@@ -96,6 +96,21 @@ where
         })
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let future = self.transport.dial_with_new_port(addr.clone())?;
+        let p = ConnectedPoint::Dialer {
+            address: addr,
+            role_override: Endpoint::Dialer,
+        };
+        Ok(MapFuture {
+            inner: future,
+            args: Some((self.fun.clone(), p)),
+        })
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.transport.address_translation(server, observed)
     }

core/src/transport/map_err.rs

@@ -84,6 +84,20 @@ where
         }
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let map = self.map.clone();
+        match self.transport.dial_with_new_port(addr) {
+            Ok(future) => Ok(MapErrDial {
+                inner: future,
+                map: Some(map),
+            }),
+            Err(err) => Err(err.map(map)),
+        }
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.transport.address_translation(server, observed)
     }

core/src/transport/memory.rs

@@ -237,6 +237,13 @@ impl Transport for MemoryTransport {
         self.dial(addr)
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<DialFuture, TransportError<Self::Error>> {
+        self.dial(addr)
+    }
+
     fn address_translation(&self, _server: &Multiaddr, _observed: &Multiaddr) -> Option<Multiaddr> {
         None
     }

core/src/transport/optional.rs

@@ -95,6 +95,17 @@ where
         }
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        if let Some(inner) = self.0.as_mut() {
+            inner.dial_with_new_port(addr)
+        } else {
+            Err(TransportError::MultiaddrNotSupported(addr))
+        }
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         if let Some(inner) = &self.0 {
             inner.address_translation(server, observed)

core/src/transport/timeout.rs

@@ -120,6 +120,20 @@ where
         })
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let dial = self
+            .inner
+            .dial_with_new_port(addr)
+            .map_err(|err| err.map(TransportTimeoutError::Other))?;
+        Ok(Timeout {
+            inner: dial,
+            timer: Delay::new(self.outgoing_timeout),
+        })
+    }
+
     fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
         self.inner.address_translation(server, observed)
     }

core/src/transport/upgrade.rs

@@ -350,6 +350,13 @@ where
         self.0.dial_as_listener(addr)
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        self.0.dial_with_new_port(addr)
+    }
+
     fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
         self.0.listen_on(addr)
     }
@@ -429,6 +436,20 @@ where
         })
     }
 
+    fn dial_with_new_port(
+        &mut self,
+        addr: Multiaddr,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
+        let future = self
+            .inner
+            .dial_with_new_port(addr)
+            .map_err(|err| err.map(TransportUpgradeError::Transport))?;
+        Ok(DialUpgradeFuture {
+            future: Box::pin(future),
+            upgrade: future::Either::Left(Some(self.upgrade.clone())),
+        })
+    }
+
     fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
         self.inner
             .listen_on(addr)

protocols/autonat/src/behaviour/as_client.rs

@@ -300,7 +300,7 @@ impl<'a> AsClient<'a> {
                 return Err(OutboundProbeError::NoServer);
             }
         };
-        let request_id = self.inner.send_request(
+        let request_id = self.inner.send_request_with_new_port(
             &server,
             DialRequest {
                 peer_id: self.local_peer_id,

protocols/autonat/src/behaviour/as_server.rs

@@ -135,6 +135,7 @@ impl<'a> HandleInnerEvent for AsServer<'a> {
                                     .override_dial_concurrency_factor(
                                         NonZeroU8::new(1).expect("1 > 0"),
                                     )
+                                    .use_new_port()
                                     .addresses(addrs)
                                     .build(),
                             },

protocols/request-response/src/lib.rs

@@ -396,6 +396,33 @@ where
         request_id
     }
 
+    /// TODO: Document
+    pub fn send_request_with_new_port(
+        &mut self,
+        peer: &PeerId,
+        request: TCodec::Request,
+    ) -> RequestId {
+        let request_id = self.next_request_id();
+        let request = RequestProtocol {
+            request_id,
+            codec: self.codec.clone(),
+            protocols: self.outbound_protocols.clone(),
+            request,
+        };
+
+        if let Some(request) = self.try_send_request(peer, request) {
+            self.pending_events.push_back(ToSwarm::Dial {
+                opts: DialOpts::peer_id(*peer).use_new_port().build(),
+            });
+            self.pending_outbound_requests
+                .entry(*peer)
+                .or_default()
+                .push(request);
+        }
+
+        request_id
+    }
+
     /// Initiates sending a response to an inbound request.
     ///
     /// If the [`ResponseChannel`] is already closed due to a timeout or the