Merge branch 'master' into fix-ads-pruning

Author: SionoiS

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/protocols/service_discovery/registrar.nim

@@ -107,7 +107,8 @@ proc pruneExpiredEntries[K](
     bounds.del(k)
 
 proc pruneExpiredAds*(registrar: Registrar, advertExpiry: Duration) =
-  let now = Moment.now()
+  #Always use seconds granularity
+  let now = Moment.init(Moment.now().epochSeconds, Second)
 
   var expiredCount = 0
 
@@ -162,8 +163,7 @@ proc waitingTime*(
     (serviceSim + discoConfig.ipSimCoefficient * ipSim + discoConfig.safetyParam)
 
   # Bound & Quantize W
-  w = max(0.0, w)
-  w = min(w, float64(uint32.high))
+  w = w.clamp(0.0, float64(uint32.high))
   w = round(w)
 
   var waitDuration = w.int64.secs
@@ -470,7 +470,8 @@ proc registration*(disco: ServiceDiscovery, peerId: PeerId, inMsg: Message): Mes
 
     return msg
 
-  let now = Moment.now()
+  #Always use seconds granularity
+  let now = Moment.init(Moment.now().epochSeconds, Second)
 
   let ticketOpt = disco.isValidTicket(regMsg, now).valueOr:
     error "invalid ticket", error
@@ -498,6 +499,8 @@ proc registration*(disco: ServiceDiscovery, peerId: PeerId, inMsg: Message): Mes
 
     return msg
 
+  tWait = min(disco.discoConfig.advertExpiry, tWait)
+
   disco.registrar.updateLowerBounds(serviceId, ad, tWait, now)
 
   var ticket = Ticket(

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,
@@ -25,6 +26,8 @@ export transport
 logScope:
   topics = "libp2p quictransport"
 
+const QuicHolePunchPacketSize* = 64
+
 type
   P2PConnection = connection.Connection
   QuicConnection = lsquic.Connection
@@ -353,11 +356,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()
@@ -367,12 +374,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()
@@ -583,6 +594,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:
@@ -593,6 +605,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(QuicHolePunchPacketSize)
+        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):
@@ -613,6 +642,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
   ##