fix(dcutr): use listener role for QUIC hole punching

Author: richard-ramos

libp2p/builders.nim

@@ -243,7 +243,7 @@ proc withWsTransport*(
 proc withQuicTransport*(b: SwitchBuilder): SwitchBuilder =
   b.withTransport(
     proc(config: TransportConfig): Transport =
-      QuicTransport.new(config.upgr, config.privateKey, config.connManager)
+      QuicTransport.new(config.upgr, config.privateKey, config.rng, config.connManager)
   )
 
 proc withMemoryTransport*(b: SwitchBuilder): SwitchBuilder =

libp2p/dialer.nim

@@ -56,7 +56,7 @@ proc dialAndUpgrade*(
       let dialed =
         try:
           libp2p_total_dial_attempts.inc()
-          await transport.dial(hostname, addrs, peerId)
+          await transport.dial(hostname, addrs, peerId, dir)
         except CancelledError as exc:
           trace "Dialing canceled", description = exc.msg, peerId
           raise exc

libp2p/protocols/connectivity/relay/rtransport.nim

@@ -107,6 +107,7 @@ method dial*(
     hostname: string,
     ma: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   peerId.withValue(pid):
     try:

libp2p/transports/memorytransport.nim

@@ -97,6 +97,7 @@ method dial*(
     hostname: string,
     ma: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   try:
     let listener = getInstance().dial($ma)

libp2p/transports/quictransport.nim

@@ -5,6 +5,7 @@ import std/[hashes, sets, sequtils]
 import chronos, chronicles, metrics, results
 import lsquic
 import
+  ../crypto/rng,
   ../wire,
   ../connmanager,
   ../multiaddress,
@@ -350,11 +351,15 @@ proc new*(
     _: type QuicTransport,
     u: Upgrade,
     privateKey: PrivateKey,
+    rng: Rng,
     connManager: ConnManager = nil,
 ): QuicTransport =
+  doAssert not rng.isNil, "Rng is nil"
+
   let self = QuicTransport(
     upgrader: QuicUpgrade(ms: u.ms, connManager: connManager.toOpt()),
     privateKey: privateKey,
+    rng: rng,
     certGenerator: defaultCertGenerator,
   )
   procCall Transport(self).initialize()
@@ -364,12 +369,16 @@ proc new*(
     _: type QuicTransport,
     u: Upgrade,
     privateKey: PrivateKey,
+    rng: Rng,
     certGenerator: CertGenerator,
     connManager: ConnManager = nil,
 ): QuicTransport =
+  doAssert not rng.isNil, "Rng is nil"
+
   let self = QuicTransport(
     upgrader: QuicUpgrade(ms: u.ms, connManager: connManager.toOpt()),
     privateKey: privateKey,
+    rng: rng,
     certGenerator: certGenerator,
   )
   procCall Transport(self).initialize()
@@ -576,6 +585,7 @@ method dial*(
     hostname: string,
     address: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   let taAddress =
     try:
@@ -586,6 +596,23 @@ method dial*(
       )
 
   try:
+    if dir == Direction.In:
+      let endpoint = self.listenerEndpointFor(taAddress)
+      if endpoint.isNone():
+        raise newException(
+          QuicTransportDialError,
+          "error in QUIC hole punch: no unique listener for address family",
+        )
+
+      # The Sync sender is the QUIC server. Open its NAT mapping with random
+      # UDP packets from the listener socket and let the expected inbound
+      # connection complete the DCUtR attempt.
+      while true:
+        let payload = self.rng.generateBytes(64)
+        await endpoint.get().datagramTransport().sendTo(taAddress, payload)
+        let delay = 10 + int(self.rng.generate(uint32) mod 191'u32)
+        await sleepAsync(delay.milliseconds)
+
     let endpoint = self.dialEndpointFor(taAddress)
     let quicConnection = await endpoint.dial(taAddress)
     peerId.withValue(expectedPeerId):
@@ -606,6 +633,8 @@ method dial*(
     )
   except TransportOsError as e:
     raise newException(QuicTransportDialError, "error in quic dial:" & e.msg, e)
+  except chronos.TransportError as e:
+    raise newException(QuicTransportDialError, "error in quic dial: " & e.msg, e)
   except DialError as e:
     raise newException(QuicTransportDialError, "error in quic dial:" & e.msg, e)
   except QuicError as e:

libp2p/transports/tcptransport.nim

@@ -286,6 +286,7 @@ method dial*(
     hostname: string,
     address: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   ## dial a peer
   if self.stopping:

libp2p/transports/tortransport.nim

@@ -221,6 +221,7 @@ method dial*(
     hostname: string,
     address: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   ## dial a peer
   ##

libp2p/transports/transport.nim

@@ -75,6 +75,7 @@ method dial*(
     hostname: string,
     address: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.base, gcsafe, async: (raises: [TransportError, CancelledError]).} =
   ## dial a peer
   ##

libp2p/transports/wstransport.nim

@@ -506,6 +506,7 @@ method dial*(
     hostname: string,
     address: MultiAddress,
     peerId: Opt[PeerId] = Opt.none(PeerId),
+    dir: Direction = Direction.Out,
 ): Future[RawConn] {.async: (raises: [transport.TransportError, CancelledError]).} =
   ## dial a peer
   ##

tests/libp2p/protocols/test_dcutr.nim

@@ -108,7 +108,7 @@ suite "Dcutr":
     # inbound identify stream
     check await directConnIdentified.withTimeout(30.seconds)
 
-  asyncTest "DCUtR establishes a new QUIC connection":
+  asyncTest "DCUtR QUIC initiator uses listener role":
     let behindNATSwitch = makeSwitch(QuicAutoAddress)
     let publicSwitch = makeSwitch(QuicAutoAddress)
 
@@ -126,35 +126,18 @@ suite "Dcutr":
       behindNATSwitch.connManager.connCount(publicSwitch.peerInfo.peerId)
     check initialConnCount == 1
 
-    let directConnSeen =
-      Future[void].Raising([CancelledError]).init("dcutr direct connection seen")
-
-    proc onDirectConn(
-        peerId: PeerId, event: ConnEvent
-    ): Future[void] {.async: (raises: [CancelledError]).} =
-      if peerId == publicSwitch.peerInfo.peerId and
-          behindNATSwitch.connManager.connCount(publicSwitch.peerInfo.peerId) >
-          initialConnCount:
-        directConnSeen.completeOnce()
-
-    # use event handler to wait-and-assert exact moment when condition 
-    # (publicSwitch has connected to behindNATSwitch) is satisfied 
-    # instead of polling, as polling can miss a short-lived true state
-    behindNATSwitch.connManager.addConnEventHandler(
-      onDirectConn, ConnEventKind.Connected
-    )
-
     for t in behindNATSwitch.transports:
       t.networkReachability = NetworkReachability.NotReachable
 
-    await DcutrClient
-      .new(connectTimeout = 5.seconds)
-      .startSync(
+    # A direct QUIC connection stands in for the relay, so the listener-role
+    # hole punch must time out.
+    try:
+      await DcutrClient.new(connectTimeout = 300.millis).startSync(
         behindNATSwitch, publicSwitch.peerInfo.peerId, behindNATSwitch.peerInfo.addrs
       )
-      .wait(10.seconds)
-
-    check await directConnSeen.withTimeout(10.seconds)
+      check false
+    except DcutrError as err:
+      check err.parent of AsyncTimeoutError
 
   template ductrClientTest(
       behindNATSwitch: Switch, publicSwitch: Switch, body: untyped