feat(quic): implement hole punching

Implement `Transport::dial_as_listener` for QUIC as specified by the [DCUtR spec](https://github.com/libp2p/specs/blob/master/relay/DCUtR.md).

To facilitate hole punching in QUIC, one side needs to send random UDP packets to establish a mapping in the routing table of the NAT device. If successful, our listener will emit a new inbound connection. This connection needs to then be sent to the dialing task. We achieve this by storing a `HashMap` of hole punch attempts indexed by the remote's `SocketAddr`. A matching incoming connection is then sent via a oneshot channel to the dialing task which continues with upgrading the connection.

Related #2883.

Pull-Request: #3964.

Author: arpankapoor

Cargo.lock

@@ -11,4 +11,5 @@ env_logger = "0.10.0"
 futures = "0.3.28"
 futures-timer = "3.0"
 libp2p = { path = "../../libp2p", features = ["async-std", "dns", "dcutr", "identify", "macros", "noise", "ping", "relay", "rendezvous", "tcp", "tokio", "yamux"] }
+libp2p-quic = { path = "../../transports/quic", features = ["async-std"] }
 log = "0.4"

examples/dcutr/Cargo.toml

@@ -23,13 +23,14 @@
 use clap::Parser;
 use futures::{
     executor::{block_on, ThreadPool},
-    future::FutureExt,
+    future::{Either, FutureExt},
     stream::StreamExt,
 };
 use libp2p::{
     core::{
         multiaddr::{Multiaddr, Protocol},
-        transport::{OrTransport, Transport},
+        muxing::StreamMuxerBox,
+        transport::Transport,
         upgrade,
     },
     dcutr,
@@ -38,9 +39,9 @@ use libp2p::{
     swarm::{NetworkBehaviour, SwarmBuilder, SwarmEvent},
     tcp, yamux, PeerId,
 };
+use libp2p_quic as quic;
 use log::info;
 use std::error::Error;
-use std::net::Ipv4Addr;
 use std::str::FromStr;
 
 #[derive(Debug, Parser)]
@@ -91,19 +92,26 @@ fn main() -> Result<(), Box<dyn Error>> {
 
     let (relay_transport, client) = relay::client::new(local_peer_id);
 
-    let transport = OrTransport::new(
-        relay_transport,
-        block_on(DnsConfig::system(tcp::async_io::Transport::new(
-            tcp::Config::default().port_reuse(true),
-        )))
-        .unwrap(),
-    )
-    .upgrade(upgrade::Version::V1Lazy)
-    .authenticate(
-        noise::Config::new(&local_key).expect("Signing libp2p-noise static DH keypair failed."),
-    )
-    .multiplex(yamux::Config::default())
-    .boxed();
+    let transport = {
+        let relay_tcp_quic_transport = relay_transport
+            .or_transport(tcp::async_io::Transport::new(
+                tcp::Config::default().port_reuse(true),
+            ))
+            .upgrade(upgrade::Version::V1)
+            .authenticate(noise::Config::new(&local_key).unwrap())
+            .multiplex(yamux::Config::default())
+            .or_transport(quic::async_std::Transport::new(quic::Config::new(
+                &local_key,
+            )));
+
+        block_on(DnsConfig::system(relay_tcp_quic_transport))
+            .unwrap()
+            .map(|either_output, _| match either_output {
+                Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+                Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+            })
+            .boxed()
+    };
 
     #[derive(NetworkBehaviour)]
     #[behaviour(to_swarm = "Event")]
@@ -164,11 +172,10 @@ fn main() -> Result<(), Box<dyn Error>> {
     .build();
 
     swarm
-        .listen_on(
-            Multiaddr::empty()
-                .with("0.0.0.0".parse::<Ipv4Addr>().unwrap().into())
-                .with(Protocol::Tcp(0)),
-        )
+        .listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap())
+        .unwrap();
+    swarm
+        .listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap())
         .unwrap();
 
     // Wait to listen on all interfaces.

examples/dcutr/src/main.rs

@@ -12,3 +12,4 @@ async-trait = "0.1"
 env_logger = "0.10.0"
 futures = "0.3.28"
 libp2p = { path = "../../libp2p", features = ["async-std", "noise", "macros", "ping", "tcp", "identify", "yamux", "relay"] }
+libp2p-quic = { path = "../../transports/quic", features = ["async-std"] }

examples/relay-server/Cargo.toml

@@ -22,10 +22,11 @@
 #![doc = include_str!("../README.md")]
 
 use clap::Parser;
-use futures::executor::block_on;
 use futures::stream::StreamExt;
+use futures::{executor::block_on, future::Either};
 use libp2p::{
     core::multiaddr::Protocol,
+    core::muxing::StreamMuxerBox,
     core::upgrade,
     core::{Multiaddr, Transport},
     identify, identity,
@@ -34,6 +35,7 @@ use libp2p::{
     swarm::{NetworkBehaviour, SwarmBuilder, SwarmEvent},
     tcp,
 };
+use libp2p_quic as quic;
 use std::error::Error;
 use std::net::{Ipv4Addr, Ipv6Addr};
 
@@ -50,12 +52,21 @@ fn main() -> Result<(), Box<dyn Error>> {
 
     let tcp_transport = tcp::async_io::Transport::default();
 
-    let transport = tcp_transport
+    let tcp_transport = tcp_transport
         .upgrade(upgrade::Version::V1Lazy)
         .authenticate(
             noise::Config::new(&local_key).expect("Signing libp2p-noise static DH keypair failed."),
         )
-        .multiplex(libp2p::yamux::Config::default())
+        .multiplex(libp2p::yamux::Config::default());
+
+    let quic_transport = quic::async_std::Transport::new(quic::Config::new(&local_key));
+
+    let transport = quic_transport
+        .or_transport(tcp_transport)
+        .map(|either_output, _| match either_output {
+            Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+            Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+        })
         .boxed();
 
     let behaviour = Behaviour {
@@ -70,13 +81,22 @@ fn main() -> Result<(), Box<dyn Error>> {
     let mut swarm = SwarmBuilder::without_executor(transport, behaviour, local_peer_id).build();
 
     // Listen on all interfaces
-    let listen_addr = Multiaddr::empty()
+    let listen_addr_tcp = Multiaddr::empty()
         .with(match opt.use_ipv6 {
             Some(true) => Protocol::from(Ipv6Addr::UNSPECIFIED),
             _ => Protocol::from(Ipv4Addr::UNSPECIFIED),
         })
         .with(Protocol::Tcp(opt.port));
-    swarm.listen_on(listen_addr)?;
+    swarm.listen_on(listen_addr_tcp)?;
+
+    let listen_addr_quic = Multiaddr::empty()
+        .with(match opt.use_ipv6 {
+            Some(true) => Protocol::from(Ipv6Addr::UNSPECIFIED),
+            _ => Protocol::from(Ipv4Addr::UNSPECIFIED),
+        })
+        .with(Protocol::Udp(opt.port))
+        .with(Protocol::QuicV1);
+    swarm.listen_on(listen_addr_quic)?;
 
     block_on(async {
         loop {

examples/relay-server/src/main.rs

@@ -3,7 +3,10 @@
 - Raise MSRV to 1.65.
   See [PR 3715].
 
+- Add hole punching support by implementing `Transport::dial_as_listener`. See [PR 3964].
+
 [PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715
+[PR 3964]: https://github.com/libp2p/rust-libp2p/pull/3964
 
 ## 0.7.0-alpha.3
 

transports/quic/CHANGELOG.md

@@ -23,7 +23,7 @@ quinn-proto = { version = "0.10.1", default-features = false, features = ["tls-r
 rand = "0.8.5"
 rustls = { version = "0.21.1", default-features = false }
 thiserror = "1.0.40"
-tokio = { version = "1.28.2", default-features = false, features = ["net", "rt"], optional = true }
+tokio = { version = "1.28.2", default-features = false, features = ["net", "rt", "time"], optional = true }
 
 [features]
 tokio = ["dep:tokio", "if-watch/tokio"]

transports/quic/Cargo.toml

@@ -279,6 +279,13 @@ impl Channel {
         Ok(Ok(()))
     }
 
+    pub(crate) async fn send(&mut self, to_endpoint: ToEndpoint) -> Result<(), Disconnected> {
+        self.to_endpoint
+            .send(to_endpoint)
+            .await
+            .map_err(|_| Disconnected {})
+    }
+
     /// Send a message to inform the [`Driver`] about an
     /// event caused by the owner of this [`Channel`] dropping.
     /// This clones the sender to the endpoint to guarantee delivery.

transports/quic/src/endpoint.rs

@@ -0,0 +1,47 @@
+use std::{net::SocketAddr, time::Duration};
+
+use futures::future::Either;
+use rand::{distributions, Rng};
+
+use crate::{
+    endpoint::{self, ToEndpoint},
+    Error, Provider,
+};
+
+pub(crate) async fn hole_puncher<P: Provider>(
+    endpoint_channel: endpoint::Channel,
+    remote_addr: SocketAddr,
+    timeout_duration: Duration,
+) -> Error {
+    let punch_holes_future = punch_holes::<P>(endpoint_channel, remote_addr);
+    futures::pin_mut!(punch_holes_future);
+    match futures::future::select(P::sleep(timeout_duration), punch_holes_future).await {
+        Either::Left(_) => Error::HandshakeTimedOut,
+        Either::Right((hole_punch_err, _)) => hole_punch_err,
+    }
+}
+
+async fn punch_holes<P: Provider>(
+    mut endpoint_channel: endpoint::Channel,
+    remote_addr: SocketAddr,
+) -> Error {
+    loop {
+        let sleep_duration = Duration::from_millis(rand::thread_rng().gen_range(10..=200));
+        P::sleep(sleep_duration).await;
+
+        let random_udp_packet = ToEndpoint::SendUdpPacket(quinn_proto::Transmit {
+            destination: remote_addr,
+            ecn: None,
+            contents: rand::thread_rng()
+                .sample_iter(distributions::Standard)
+                .take(64)
+                .collect(),
+            segment_size: None,
+            src_ip: None,
+        });
+
+        if endpoint_channel.send(random_udp_packet).await.is_err() {
+            return Error::EndpointDriverCrashed;
+        }
+    }
+}

transports/quic/src/hole_punching.rs

@@ -59,9 +59,12 @@
 
 mod connection;
 mod endpoint;
+mod hole_punching;
 mod provider;
 mod transport;
 
+use std::net::SocketAddr;
+
 pub use connection::{Connecting, Connection, Substream};
 pub use endpoint::Config;
 #[cfg(feature = "async-std")]
@@ -94,6 +97,14 @@ pub enum Error {
     /// The [`Connecting`] future timed out.
     #[error("Handshake with the remote timed out.")]
     HandshakeTimedOut,
+
+    /// Error when `Transport::dial_as_listener` is called without an active listener.
+    #[error("Tried to dial as listener without an active listener.")]
+    NoActiveListenerForDialAsListener,
+
+    /// Error when holepunching for a remote is already in progress
+    #[error("Already punching hole for {0}).")]
+    HolePunchInProgress(SocketAddr),
 }
 
 /// Dialing a remote peer failed.

transports/quic/src/lib.rs

@@ -18,12 +18,13 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use futures::Future;
+use futures::{future::BoxFuture, Future};
 use if_watch::IfEvent;
 use std::{
     io,
     net::SocketAddr,
     task::{Context, Poll},
+    time::Duration,
 };
 
 #[cfg(feature = "async-std")]
@@ -74,4 +75,7 @@ pub trait Provider: Unpin + Send + Sized + 'static {
         watcher: &mut Self::IfWatcher,
         cx: &mut Context<'_>,
     ) -> Poll<io::Result<IfEvent>>;
+
+    /// Sleep for specified amount of time.
+    fn sleep(duration: Duration) -> BoxFuture<'static, ()>;
 }

transports/quic/src/provider.rs

@@ -26,6 +26,7 @@ use std::{
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
+    time::Duration,
 };
 
 use crate::GenTransport;
@@ -104,6 +105,10 @@ impl super::Provider for Provider {
     ) -> Poll<io::Result<if_watch::IfEvent>> {
         watcher.poll_if_event(cx)
     }
+
+    fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
+        async_std::task::sleep(duration).boxed()
+    }
 }
 
 type ReceiveStreamItem = (

transports/quic/src/provider/async_std.rs

@@ -18,11 +18,12 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use futures::{ready, Future};
+use futures::{future::BoxFuture, ready, Future, FutureExt};
 use std::{
     io,
     net::SocketAddr,
     task::{Context, Poll},
+    time::Duration,
 };
 use tokio::{io::ReadBuf, net::UdpSocket};
 
@@ -95,4 +96,8 @@ impl super::Provider for Provider {
     ) -> Poll<io::Result<if_watch::IfEvent>> {
         watcher.poll_if_event(cx)
     }
+
+    fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
+        tokio::time::sleep(duration).boxed()
+    }
 }