Hole Punching (#806)

Co-authored-by: Tanguy <tanguy@status.im>

libp2p/protocols/connectivity/autonat/client.nim

@@ -20,8 +20,6 @@ import ../../../switch,
        ../../../peerid
 import core
 
-export core
-
 logScope:
   topics = "libp2p autonat"
 

libp2p/protocols/connectivity/autonat/service.nim

@@ -17,9 +17,12 @@ import chronos, metrics
 import ../../../switch
 import ../../../wire
 import client
+from core import NetworkReachability, AutonatUnreachableError
 import ../../../utils/heartbeat
 import ../../../crypto/crypto
 
+export options, core.NetworkReachability
+
 logScope:
   topics = "libp2p autonatservice"
 
@@ -30,7 +33,7 @@ type
     newConnectedPeerHandler: PeerEventHandler
     addressMapper: AddressMapper
     scheduleHandle: Future[void]
-    networkReachability: NetworkReachability
+    networkReachability*: NetworkReachability
     confidence: Option[float]
     answers: Deque[NetworkReachability]
     autonatClient: AutonatClient
@@ -71,9 +74,6 @@ proc new*(
     dialTimeout: dialTimeout,
     enableAddressMapper: enableAddressMapper)
 
-proc networkReachability*(self: AutonatService): NetworkReachability {.inline.} =
-  return self.networkReachability
-
 proc callHandler(self: AutonatService) {.async.} =
   if not isNil(self.statusAndConfidenceHandler):
     await self.statusAndConfidenceHandler(self.networkReachability, self.confidence)

libp2p/protocols/connectivity/dcutr/client.nim

@@ -23,6 +23,8 @@ import ../../protocol,
        ../../../switch,
        ../../../utils/future
 
+export DcutrError
+
 type
   DcutrClient* = ref object
     connectTimeout: Duration

libp2p/protocols/connectivity/dcutr/server.nim

@@ -23,6 +23,7 @@ import ../../protocol,
        ../../../switch,
        ../../../utils/future
 
+export DcutrError
 export chronicles
 
 type Dcutr* = ref object of LPProtocol

libp2p/protocols/connectivity/relay/relay.nim

@@ -101,7 +101,7 @@ proc createReserveResponse(
                      status: some(Ok))
   return ok(msg)
 
-proc isRelayed(conn: Connection): bool =
+proc isRelayed*(conn: Connection): bool =
   var wrappedConn = conn
   while not isNil(wrappedConn):
     if wrappedConn of RelayConnection:

libp2p/services/hpservice.nim

@@ -0,0 +1,105 @@
+# Nim-LibP2P
+# Copyright (c) 2022 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[tables, sequtils]
+
+import chronos, chronicles
+
+import ../switch, ../wire
+import ../protocols/rendezvous
+import ../services/autorelayservice
+import ../discovery/[rendezvousinterface, discoverymngr]
+import ../protocols/connectivity/relay/relay
+import ../protocols/connectivity/autonat/service
+import ../protocols/connectivity/dcutr/[client, server]
+
+
+logScope:
+  topics = "libp2p hpservice"
+
+type
+  HPService* = ref object of Service
+    newConnectedPeerHandler: PeerEventHandler
+    onNewStatusHandler: StatusAndConfidenceHandler
+    autoRelayService: AutoRelayService
+    autonatService: AutonatService
+    isPublicIPAddrProc: IsPublicIPAddrProc
+
+  IsPublicIPAddrProc* = proc(ta: TransportAddress): bool {.gcsafe, raises: [Defect].}
+
+proc new*(T: typedesc[HPService], autonatService: AutonatService, autoRelayService: AutoRelayService,
+          isPublicIPAddrProc: IsPublicIPAddrProc = isGlobal): T =
+  return T(autonatService: autonatService, autoRelayService: autoRelayService, isPublicIPAddrProc: isPublicIPAddrProc)
+
+proc tryStartingDirectConn(self: HPService, switch: Switch, peerId: PeerId): Future[bool] {.async.} =
+  await sleepAsync(500.milliseconds) # wait for AddressBook to be populated
+  for address in switch.peerStore[AddressBook][peerId]:
+    try:
+      let ta = initTAddress(address)
+      if ta.isOk() and self.isPublicIPAddrProc(ta.get()):
+        await switch.connect(peerId, @[address], true, false)
+        debug "Direct connection created."
+        return true
+    except CatchableError as err:
+      debug "Failed to create direct connection.", err = err.msg
+      continue
+  return false
+
+method setup*(self: HPService, switch: Switch): Future[bool] {.async.} =
+  var hasBeenSetup = await procCall Service(self).setup(switch)
+  hasBeenSetup = hasBeenSetup and await self.autonatService.setup(switch)
+
+  if hasBeenSetup:
+    let dcutrProto = Dcutr.new(switch)
+    switch.mount(dcutrProto)
+
+    self.newConnectedPeerHandler = proc (peerId: PeerId, event: PeerEvent): Future[void] {.async.} =
+      try:
+        let conn = switch.connManager.selectMuxer(peerId).connection
+        if isRelayed(conn) and conn.transportDir == Direction.In:
+          if await self.tryStartingDirectConn(switch, peerId):
+            await conn.close()
+            return
+          let dcutrClient = DcutrClient.new()
+          var natAddrs = switch.peerStore.getMostObservedProtosAndPorts()
+          if natAddrs.len == 0:
+            natAddrs =  switch.peerInfo.listenAddrs.mapIt(switch.peerStore.guessDialableAddr(it))
+          await dcutrClient.startSync(switch, peerId, natAddrs)
+          await sleepAsync(2000.milliseconds) # grace period before closing relayed connection
+          await conn.close()
+      except CatchableError as err:
+        debug "Hole punching failed during dcutr", err = err.msg
+
+    switch.connManager.addPeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+
+    self.onNewStatusHandler = proc (networkReachability: NetworkReachability, confidence: Option[float]) {.gcsafe, async.} =
+      if networkReachability == NetworkReachability.NotReachable:
+        discard await self.autoRelayService.setup(switch)
+      elif networkReachability == NetworkReachability.Reachable:
+        discard await self.autoRelayService.stop(switch)
+
+      # We do it here instead of in the AutonatService because this is useful only when hole punching.
+      for t in switch.transports:
+        t.networkReachability = networkReachability
+
+    self.autonatService.statusAndConfidenceHandler(self.onNewStatusHandler)
+  return hasBeenSetup
+
+method run*(self: HPService, switch: Switch) {.async, public.} =
+  await self.autonatService.run(switch)
+
+method stop*(self: HPService, switch: Switch): Future[bool] {.async, public.} =
+  discard await self.autonatService.stop(switch)
+  if not isNil(self.newConnectedPeerHandler):
+    switch.connManager.removePeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)

libp2p/wire.nim

@@ -71,7 +71,7 @@ proc initTAddress*(ma: MultiAddress): MaResult[TransportAddress] =
           res.port = Port(fromBytesBE(uint16, pbuf))
           ok(res)
   else:
-    err("MultiAddress must be wire address (tcp, udp or unix)")
+    err("MultiAddress must be wire address (tcp, udp or unix): " & $ma)
 
 proc connect*(
   ma: MultiAddress,

tests/stubs/autonatclientstub.nim

@@ -19,6 +19,7 @@ import ../../libp2p/[protocols/connectivity/autonat/client,
                      peerid,
                      multiaddress,
                      switch]
+from ../../libp2p/protocols/connectivity/autonat/core import NetworkReachability, AutonatUnreachableError, AutonatError
 
 type
   AutonatClientStub* = ref object of AutonatClient

tests/stubs/switchstub.nim

@@ -0,0 +1,48 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+{.used.}
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import chronos
+import ../../libp2p/[peerid, multiaddress, switch]
+
+type
+  SwitchStub* = ref object of Switch
+    switch*: Switch
+    connectStub*: proc(): Future[void] {.async.}
+
+method connect*(
+ self: SwitchStub,
+ peerId: PeerId,
+ addrs: seq[MultiAddress],
+ forceDial = false,
+ reuseConnection = true,
+ upgradeDir = Direction.Out) {.async.} =
+  if (self.connectStub != nil):
+    await self.connectStub()
+  else:
+    await self.switch.connect(peerId, addrs, forceDial, reuseConnection, upgradeDir)
+
+proc new*(T: typedesc[SwitchStub], switch: Switch, connectStub: proc (): Future[void] {.async.} = nil): T =
+  return SwitchStub(
+    switch: switch,
+    peerInfo: switch.peerInfo,
+    ms: switch.ms,
+    transports: switch.transports,
+    connManager: switch.connManager,
+    peerStore: switch.peerStore,
+    dialer: switch.dialer,
+    nameResolver: switch.nameResolver,
+    services: switch.services,
+    connectStub: connectStub)

tests/stubs/torstub.nim

@@ -1,3 +1,12 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
 {.used.}
 
 when (NimMajor, NimMinor) < (1, 4):