feat(service): add wildcard address resolver

libp2p/builders.nim

@@ -19,7 +19,8 @@ runnableExamples:
 {.push raises: [].}
 
 import
-  options, tables, chronos, chronicles, sequtils,
+  options, tables, chronos, chronicles, sequtils
+import
   switch, peerid, peerinfo, stream/connection, multiaddress,
   crypto/crypto, transports/[transport, tcptransport],
   muxers/[muxer, mplex/mplex, yamux/yamux],
@@ -28,6 +29,7 @@ import
   connmanager, upgrademngrs/muxedupgrade, observedaddrmanager,
   nameresolving/nameresolver,
   errors, utility
+import services/wildcardresolverservice
 
 export
   switch, peerid, peerinfo, connection, multiaddress, crypto, errors
@@ -59,6 +61,7 @@ type
     rdv: RendezVous
     services: seq[Service]
     observedAddrManager: ObservedAddrManager
+    enableWildcardResolver: bool
 
 proc new*(T: type[SwitchBuilder]): T {.public.} =
   ## Creates a SwitchBuilder
@@ -76,7 +79,8 @@ proc new*(T: type[SwitchBuilder]): T {.public.} =
     maxOut: -1,
     maxConnsPerPeer: MaxConnectionsPerPeer,
     protoVersion: ProtoVersion,
-    agentVersion: AgentVersion)
+    agentVersion: AgentVersion,
+    enableWildcardResolver: true)
 
 proc withPrivateKey*(b: SwitchBuilder, privateKey: PrivateKey): SwitchBuilder {.public.} =
   ## Set the private key of the switch. Will be used to
@@ -85,20 +89,18 @@ proc withPrivateKey*(b: SwitchBuilder, privateKey: PrivateKey): SwitchBuilder {.
   b.privKey = some(privateKey)
   b
 
-proc withAddress*(b: SwitchBuilder, address: MultiAddress): SwitchBuilder {.public.} =
-  ## | Set the listening address of the switch
-  ## | Calling it multiple time will override the value
-
-  b.addresses = @[address]
-  b
-
-proc withAddresses*(b: SwitchBuilder, addresses: seq[MultiAddress]): SwitchBuilder {.public.} =
+proc withAddresses*(b: SwitchBuilder, addresses: seq[MultiAddress], enableWildcardResolver: bool = true): SwitchBuilder {.public.} =
   ## | Set the listening addresses of the switch
   ## | Calling it multiple time will override the value
-
   b.addresses = addresses
+  b.enableWildcardResolver = enableWildcardResolver
   b
 
+proc withAddress*(b: SwitchBuilder, address: MultiAddress, enableWildcardResolver: bool = true): SwitchBuilder {.public.} =
+  ## | Set the listening address of the switch
+  ## | Calling it multiple time will override the value
+  b.withAddresses(@[address], enableWildcardResolver)
+
 proc withSignedPeerRecord*(b: SwitchBuilder, sendIt = true): SwitchBuilder {.public.} =
   b.sendSignedPeerRecord = sendIt
   b
@@ -261,6 +263,9 @@ proc build*(b: SwitchBuilder): Switch
     else:
       PeerStore.new(identify)
 
+  if b.enableWildcardResolver:
+    b.services.insert(WildcardAddressResolverService.new(), 0)
+
   let switch = newSwitch(
     peerInfo = peerInfo,
     transports = transports,

libp2p/multiaddress.nim

@@ -18,7 +18,7 @@ import tables, strutils, sets
 import multicodec, multihash, multibase, transcoder, vbuffer, peerid,
        protobuf/minprotobuf, errors, utility
 import stew/[base58, base32, endians2, results]
-export results, minprotobuf, vbuffer, utility
+export results, minprotobuf, vbuffer, utility, multicodec
 
 logScope:
   topics = "libp2p multiaddress"

libp2p/muxers/yamux/yamux.nim

@@ -164,6 +164,7 @@ type
     closedRemotely: Future[void].Raising([])
     closedLocally: bool
     receivedData: AsyncEvent
+    returnedEof: bool
 
 proc `$`(channel: YamuxChannel): string =
   result = if channel.conn.dir == Out: "=> " else: "<= "
@@ -203,8 +204,8 @@ proc remoteClosed(channel: YamuxChannel) {.async: (raises: []).} =
 
 method closeImpl*(channel: YamuxChannel) {.async: (raises: []).} =
   if not channel.closedLocally:
-    trace "Closing yamux channel locally", streamId = channel.id, conn = channel.conn
     channel.closedLocally = true
+    channel.isEof = true
 
     if not channel.isReset and channel.sendQueue.len == 0:
       try: await channel.conn.write(YamuxHeader.data(channel.id, 0, {Fin}))
@@ -272,16 +273,17 @@ method readOnce*(
         newLPStreamClosedError()
       else:
         newLPStreamConnDownError()
-  if channel.isEof:
+  if channel.returnedEof:
     raise newLPStreamRemoteClosedError()
   if channel.recvQueue.len == 0:
     channel.receivedData.clear()
     try:  # https://github.com/status-im/nim-chronos/issues/516
       discard await race(channel.closedRemotely, channel.receivedData.wait())
     except ValueError: raiseAssert("Futures list is not empty")
     if channel.closedRemotely.completed() and channel.recvQueue.len == 0:
+      channel.returnedEof = true
       channel.isEof = true
-      return 0 # we return 0 to indicate that the channel is closed for reading from now on
+      return 0
 
   let toRead = min(channel.recvQueue.len, nbytes)
 

libp2p/peerinfo.nim

@@ -24,12 +24,16 @@ type
   AddressMapper* =
     proc(listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]]
       {.gcsafe, raises: [].}
+      ## A proc that expected to resolve the listen addresses into dialable addresses
 
   PeerInfo* {.public.} = ref object
     peerId*: PeerId
     listenAddrs*: seq[MultiAddress]
-    addrs: seq[MultiAddress]
+    ## contains addresses the node listens on, which may include wildcard and private addresses (not directly reachable).
+    addrs*: seq[MultiAddress]
+    ## contains resolved addresses that other peers can use to connect, including public-facing NAT and port-forwarded addresses.
     addressMappers*: seq[AddressMapper]
+    ## contains a list of procs that can be used to resolve the listen addresses into dialable addresses.
     protocols*: seq[string]
     protoVersion*: string
     agentVersion*: string
@@ -49,6 +53,7 @@ func shortLog*(p: PeerInfo): auto =
 chronicles.formatIt(PeerInfo): shortLog(it)
 
 proc update*(p: PeerInfo) {.async.} =
+  # if p.addrs.len == 0:
   p.addrs = p.listenAddrs
   for mapper in p.addressMappers:
     p.addrs = await mapper(p.addrs)

libp2p/protocols/ping.nim

@@ -56,7 +56,7 @@ method init*(p: Ping) =
       trace "handling ping", conn
       var buf: array[PingSize, byte]
       await conn.readExactly(addr buf[0], PingSize)
-      trace "echoing ping", conn, pingData = @buf
+      trace "echoing ping", conn
       await conn.write(@buf)
       if not isNil(p.pingHandler):
         await p.pingHandler(conn.peerId)

libp2p/services/autorelayservice.nim

@@ -38,7 +38,7 @@ proc isRunning*(self: AutoRelayService): bool =
 proc addressMapper(
   self: AutoRelayService,
   listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]] {.async.} =
-  return concat(toSeq(self.relayAddresses.values))
+  return concat(toSeq(self.relayAddresses.values)) & listenAddrs
 
 proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, switch: Switch) {.async.} =
   while self.running:

libp2p/services/wildcardresolverservice.nim

@@ -0,0 +1,205 @@
+# Nim-LibP2P
+# Copyright (c) 2024 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.
+
+{.push raises: [].}
+
+import std/sequtils
+import stew/[byteutils, results, endians2]
+import chronos, chronos/transports/[osnet, ipnet], chronicles
+import ../[multiaddress, multicodec]
+import ../switch
+
+logScope:
+  topics = "libp2p wildcardresolverservice"
+
+type
+  WildcardAddressResolverService* = ref object of Service
+    ## Service used to resolve wildcard addresses of the type "0.0.0.0" for IPv4 or "::" for IPv6.
+    ## When used with a `Switch`, this service will be automatically set up and stopped
+    ## when the switch starts and stops. This is facilitated by adding the service to the switch's
+    ## list of services using the `.withServices(@[svc])` method in the `SwitchBuilder`.
+    networkInterfaceProvider: NetworkInterfaceProvider
+    ## Provides a list of network addresses.
+    addressMapper: AddressMapper
+    ## An implementation of an address mapper that takes a list of listen addresses and expands each wildcard address
+    ## to the respective list of interface addresses. As an example, if the listen address is 0.0.0.0:4001
+    ## and the machine has 2 interfaces with IPs 172.217.11.174 and 64.233.177.113, the address mapper will
+    ## expand the wildcard address to 172.217.11.174:4001 and 64.233.177.113:4001.
+
+  NetworkInterfaceProvider* =
+    proc(addrFamily: AddressFamily): seq[InterfaceAddress] {.gcsafe, raises: [].}
+
+proc isLoopbackOrUp(networkInterface: NetworkInterface): bool =
+  if (networkInterface.ifType == IfSoftwareLoopback) or
+      (networkInterface.state == StatusUp): true else: false
+
+proc getAddresses(addrFamily: AddressFamily): seq[InterfaceAddress] =
+  ## This method retrieves the addresses of network interfaces based on the specified address family.
+  ##
+  ## The `getAddresses` method filters the available network interfaces to include only
+  ## those that are either loopback or up. It then collects all the addresses from these
+  ## interfaces and filters them to match the provided address family.
+  ##
+  ## Parameters:
+  ## - `addrFamily`: The address family to filter the network addresses (e.g., `AddressFamily.IPv4` or `AddressFamily.IPv6`).
+  ##
+  ## Returns:
+  ## - A sequence of `InterfaceAddress` objects that match the specified address family.
+  let
+    interfaces = getInterfaces().filterIt(it.isLoopbackOrUp())
+    flatInterfaceAddresses = concat(interfaces.mapIt(it.addresses))
+    filteredInterfaceAddresses =
+      flatInterfaceAddresses.filterIt(it.host.family == addrFamily)
+  return filteredInterfaceAddresses
+
+proc new*(
+    T: typedesc[WildcardAddressResolverService],
+    networkInterfaceProvider: NetworkInterfaceProvider = getAddresses,
+): T =
+  ## This procedure initializes a new `WildcardAddressResolverService` with the provided network interface provider.
+  ##
+  ## Parameters:
+  ## - `T`: The type descriptor for `WildcardAddressResolverService`.
+  ## - `networkInterfaceProvider`: A provider that offers access to network interfaces. Defaults to a new instance of `NetworkInterfaceProvider`.
+  ##
+  ## Returns:
+  ## - A new instance of `WildcardAddressResolverService`.
+  return T(networkInterfaceProvider: networkInterfaceProvider)
+
+proc getProtocolArgument*(ma: MultiAddress, codec: MultiCodec): MaResult[seq[byte]] =
+  var buffer: seq[byte]
+  for item in ma:
+    let
+      ritem = ?item
+      code = ?ritem.protoCode()
+    if code == codec:
+      let arg = ?ritem.protoAddress()
+      return ok(arg)
+
+  err("Multiaddress codec has not been found")
+
+proc getWildcardMultiAddresses(
+    interfaceAddresses: seq[InterfaceAddress], protocol: Protocol, port: Port
+): seq[MultiAddress] =
+  var addresses: seq[MultiAddress]
+  for ifaddr in interfaceAddresses:
+    var address = ifaddr.host
+    address.port = port
+    MultiAddress.init(address, protocol).withValue(maddress):
+      addresses.add(maddress)
+  addresses
+
+proc getWildcardAddress(
+    maddress: MultiAddress,
+    multiCodec: MultiCodec,
+    anyAddr: openArray[uint8],
+    addrFamily: AddressFamily,
+    port: Port,
+    networkInterfaceProvider: NetworkInterfaceProvider,
+): seq[MultiAddress] =
+  var addresses: seq[MultiAddress]
+  maddress.getProtocolArgument(multiCodec).withValue(address):
+    if address == anyAddr:
+      let filteredInterfaceAddresses = networkInterfaceProvider(addrFamily)
+      addresses.add(
+        getWildcardMultiAddresses(filteredInterfaceAddresses, IPPROTO_TCP, port)
+      )
+    else:
+      addresses.add(maddress)
+  return addresses
+
+proc expandWildcardAddresses(
+    networkInterfaceProvider: NetworkInterfaceProvider, listenAddrs: seq[MultiAddress]
+): seq[MultiAddress] =
+  var addresses: seq[MultiAddress]
+  # In this loop we expand bound addresses like `0.0.0.0` and `::` to list of interface addresses.
+  for listenAddr in listenAddrs:
+    if TCP_IP.matchPartial(listenAddr):
+      listenAddr.getProtocolArgument(multiCodec("tcp")).withValue(portArg):
+        let port = Port(uint16.fromBytesBE(portArg))
+        if IP4.matchPartial(listenAddr):
+          let wildcardAddresses = getWildcardAddress(
+            listenAddr,
+            multiCodec("ip4"),
+            AnyAddress.address_v4,
+            AddressFamily.IPv4,
+            port,
+            networkInterfaceProvider,
+          )
+          addresses.add(wildcardAddresses)
+        elif IP6.matchPartial(listenAddr):
+          let wildcardAddresses = getWildcardAddress(
+            listenAddr,
+            multiCodec("ip6"),
+            AnyAddress6.address_v6,
+            AddressFamily.IPv6,
+            port,
+            networkInterfaceProvider,
+          )
+          addresses.add(wildcardAddresses)
+        else:
+          addresses.add(listenAddr)
+    else:
+      addresses.add(listenAddr)
+  addresses
+
+method setup*(
+    self: WildcardAddressResolverService, switch: Switch
+): Future[bool] {.async.} =
+  ## Sets up the `WildcardAddressResolverService`.
+  ##
+  ## This method adds the address mapper to the peer's list of address mappers.
+  ##
+  ## Parameters:
+  ## - `self`: The instance of `WildcardAddressResolverService` being set up.
+  ## - `switch`: The switch context in which the service operates.
+  ##
+  ## Returns:
+  ## - A `Future[bool]` that resolves to `true` if the setup was successful, otherwise `false`.
+  self.addressMapper = proc(
+      listenAddrs: seq[MultiAddress]
+  ): Future[seq[MultiAddress]] {.async.} =
+    return expandWildcardAddresses(self.networkInterfaceProvider, listenAddrs)
+
+  debug "Setting up WildcardAddressResolverService"
+  let hasBeenSetup = await procCall Service(self).setup(switch)
+  if hasBeenSetup:
+    switch.peerInfo.addressMappers.add(self.addressMapper)
+  return hasBeenSetup
+
+method run*(self: WildcardAddressResolverService, switch: Switch) {.async, public.} =
+  ## Runs the WildcardAddressResolverService for a given switch.
+  ##
+  ## It updates the peer information for the provided switch by running the registered address mapper. Any other
+  ## address mappers that are registered with the switch will also be run.
+  ##
+  trace "Running WildcardAddressResolverService"
+  await switch.peerInfo.update()
+
+method stop*(
+    self: WildcardAddressResolverService, switch: Switch
+): Future[bool] {.async, public.} =
+  ## Stops the WildcardAddressResolverService.
+  ##
+  ## Handles the shutdown process of the WildcardAddressResolverService for a given switch.
+  ## It removes the address mapper from the switch's list of address mappers.
+  ## It then updates the peer information for the provided switch. Any wildcard address wont be resolved anymore.
+  ##
+  ## Parameters:
+  ## - `self`: The instance of the WildcardAddressResolverService.
+  ## - `switch`: The Switch object associated with the service.
+  ##
+  ## Returns:
+  ## - A future that resolves to `true` if the service was successfully stopped, otherwise `false`.
+  debug "Stopping WildcardAddressResolverService"
+  let hasBeenStopped = await procCall Service(self).stop(switch)
+  if hasBeenStopped:
+    switch.peerInfo.addressMappers.keepItIf(it != self.addressMapper)
+    await switch.peerInfo.update()
+  return hasBeenStopped

libp2p/utility.nim

@@ -112,6 +112,9 @@ template withValue*[T](self: Opt[T] | Option[T], value, body: untyped): untyped
     let value {.inject.} = temp.get()
     body
 
+template withValue*[T, E](self: Result[T, E], value, body: untyped): untyped =
+  self.toOpt().withValue(value, body)
+
 macro withValue*[T](self: Opt[T] | Option[T], value, body, elseStmt: untyped): untyped =
   let elseBody = elseStmt[0]
   quote do: