fix(dcutr): preserve QUIC server role during hole punching

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/crypto/rng.nim

@@ -113,6 +113,19 @@ proc randBelow(rng: Rng, max: uint32): int =
     if r >= threshold:
       return (r mod max).int
 
+proc rand*(rng: Rng, low, high: int): int =
+  ## Return a uniformly random integer in the inclusive range [`low`, `high`].
+  doAssert low <= high, "random range must not be empty"
+
+  let width =
+    if low >= 0 or high < 0:
+      uint64(high - low) + 1
+    else:
+      uint64(-(low + 1)) + uint64(high) + 2
+  doAssert width <= uint64(uint32.high), "random range is too large"
+
+  low + rng.randBelow(width.uint32)
+
 proc pick*[T](rng: Rng, x: openArray[T], n: int): Opt[seq[T]] =
   doAssert n >= 0, "n must be non-negative"
   if x.len == 0:

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 = self.rng.rand(10, 200)
+        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

tests/libp2p/transports/test_quic.nim

@@ -9,6 +9,7 @@ import
     transports/transport,
     transports/quictransport,
     upgrademngrs/upgrade,
+    utils/future,
     muxers/muxer,
     multiaddress,
   ]
@@ -19,7 +20,7 @@ import ./utils
 
 proc quicTransProvider(): Transport {.gcsafe, raises: [].} =
   try:
-    return QuicTransport.new(Upgrade(), PrivateKey.random(ECDSA, rng()).tryGet())
+    return QuicTransport.new(Upgrade(), PrivateKey.random(ECDSA, rng()).tryGet(), rng())
   except ResultError[crypto.CryptoError]:
     raiseAssert "should not happen"
 
@@ -55,6 +56,36 @@ suite "Quic transport":
     streamProvider,
   )
 
+  asyncTest "listener-role dial sends UDP hole-punch packets":
+    let packetReceived =
+      Future[int].Raising([CancelledError]).init("QUIC hole-punch packet received")
+
+    proc receivePacket(
+        transp: DatagramTransport, remote: TransportAddress
+    ): Future[void] {.async: (raises: []).} =
+      try:
+        packetReceived.completeOnce(transp.getMessage().len)
+      except chronos.TransportError:
+        discard
+
+    let receiver =
+      newDatagramTransport(receivePacket, local = initTAddress("127.0.0.1:0"))
+    defer:
+      receiver.close()
+
+    let puncher = await createQuicTransport(isServer = true)
+    defer:
+      await puncher.stop()
+
+    let remoteAddr = MultiAddress
+      .init("/ip4/127.0.0.1/udp/" & $receiver.localAddress().port & "/quic-v1")
+      .tryGet()
+    let punchFut = puncher.dial("", remoteAddr, Opt.none(PeerId), Direction.In)
+    defer:
+      await punchFut.cancelAndWait()
+
+    check (await packetReceived.wait(1.seconds)) == 64
+
   asyncTest "transport e2e - invalid cert - server":
     let server = await createQuicTransport(isServer = true, withInvalidCert = true)
     asyncSpawn createServerAcceptConn(server)()

tests/libp2p/transports/utils.nim

@@ -90,9 +90,9 @@ proc createQuicTransport*(
 
   let trans =
     if withInvalidCert:
-      QuicTransport.new(Upgrade(), key, invalidCertGenerator)
+      QuicTransport.new(Upgrade(), key, rng(), invalidCertGenerator)
     else:
-      QuicTransport.new(Upgrade(), key)
+      QuicTransport.new(Upgrade(), key, rng())
 
   if isServer: # servers are started because they need to listen
     await trans.start(addresses)