Create an ObservedAddrManager and add an AddressMapper in AutonatService and AutoRelayService

.pinned

@@ -1,6 +1,6 @@
 bearssl;https://github.com/status-im/nim-bearssl@#acf9645e328bdcab481cfda1c158e07ecd46bd7b
 chronicles;https://github.com/status-im/nim-chronicles@#32ac8679680ea699f7dbc046e8e0131cac97d41a
-chronos;https://github.com/status-im/nim-chronos@#5d3da66e563d21277b57a9b601744273c083a01b
+chronos;https://github.com/status-im/nim-chronos@#f7835a192b45c37e97614d865141f21eea8c156e
 dnsclient;https://github.com/ba0f3/dnsclient.nim@#fcd7443634b950eaea574e5eaa00a628ae029823
 faststreams;https://github.com/status-im/nim-faststreams@#814f8927e1f356f39219f37f069b83066bcc893a
 httputils;https://github.com/status-im/nim-http-utils@#a85bd52ae0a956983ca6b3267c72961d2ec0245f

libp2p/observedaddrmanager.nim

@@ -0,0 +1,80 @@
+# 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.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import
+  std/[sequtils, tables],
+  chronos,
+  multiaddress
+
+type
+  ## Manages observed MultiAddresses by reomte peers. It keeps track of the most observed IP and IP/Port.
+  ObservedAddrManager* = ref object of RootObj
+    observedIPsAndPorts: seq[MultiAddress]
+    maxSize: int
+    minCount: int
+
+  IPVersion* = enum
+    IPv4, IPv6
+
+proc addObservation*(self:ObservedAddrManager, observedAddr: MultiAddress) =
+  ## Adds a new observed MultiAddress. If the number of observations exceeds maxSize, the oldest one is removed.
+  ## Both IP and IP/Port are tracked.
+  if self.observedIPsAndPorts.len >= self.maxSize:
+      self.observedIPsAndPorts.del(0)
+  self.observedIPsAndPorts.add(observedAddr)
+
+proc getIP(self: ObservedAddrManager, observations: seq[MultiAddress], ipVersion: MaPattern): Opt[MultiAddress] =
+  var countTable = toCountTable(observations)
+  countTable.sort()
+  var orderedPairs = toSeq(countTable.pairs)
+  for (ma, count) in orderedPairs:
+    let ip = ma[0].get()
+    if ipVersion.match(ip) and count >= self.minCount:
+      return Opt.some(ma)
+  return Opt.none(MultiAddress)
+
+proc getMostObservedIP*(self: ObservedAddrManager, ipVersion: IPVersion): Opt[MultiAddress] =
+  ## Returns the most observed IP address or none if the number of observations are less than minCount.
+  let observedIPs = self.observedIPsAndPorts.mapIt(it[0].get())
+  return self.getIP(observedIPs, if ipVersion == IPv4: IP4 else: IP6)
+
+proc getMostObservedIPAndPort*(self: ObservedAddrManager, ipVersion: IPVersion): Opt[MultiAddress] =
+  ## Returns the most observed IP/Port address or none if the number of observations are less than minCount.
+  return self.getIP(self.observedIPsAndPorts, if ipVersion == IPv4: IP4 else: IP6)
+
+proc getMostObservedIPsAndPorts*(self: ObservedAddrManager): seq[MultiAddress] =
+  ## Returns the most observed IP4/Port and IP6/Port address or an empty seq if the number of observations
+  ## are less than minCount.
+  var res: seq[MultiAddress]
+  let ip4 = self.getMostObservedIPAndPort(IPv4)
+  if ip4.isSome():
+    res.add(ip4.get())
+  let ip6 = self.getMostObservedIPAndPort(IPv6)
+  if ip6.isSome():
+    res.add(ip6.get())
+  return res
+
+proc `$`*(self: ObservedAddrManager): string =
+  ## Returns a string representation of the ObservedAddrManager.
+  return "IPs and Ports: " & $self.observedIPsAndPorts
+
+proc new*(
+  T: typedesc[ObservedAddrManager],
+  maxSize = 10,
+  minCount = 3): T =
+  ## Creates a new ObservedAddrManager.
+  return T(
+    observedIPsAndPorts: newSeq[MultiAddress](),
+    maxSize: maxSize,
+    minCount: minCount)

libp2p/peerstore.nim

@@ -28,7 +28,7 @@ else:
 
 import
   std/[tables, sets, options, macros],
-  chronos,
+  chronos, chronicles,
   ./crypto/crypto,
   ./protocols/identify,
   ./protocols/protocol,
@@ -220,3 +220,40 @@ proc identify*(
       peerStore.updatePeerInfo(info)
   finally:
     await stream.closeWithEOF()
+
+proc getMostObservedIP*(self: PeerStore, ipVersion: IPVersion): Opt[MultiAddress] =
+  ## Returns the most observed IP address or none if the number of observations are less than minCount.
+  return self.identify.getMostObservedIP(ipVersion)
+
+proc replaceMAIpByMostObserved*(
+  self: PeerStore,
+  ma: MultiAddress): Opt[MultiAddress] =
+  try:
+    let maIP = ma[0]
+    let maWithoutIP = ma[1..^1]
+
+    if maWithoutIP.isErr():
+      return Opt.none(MultiAddress)
+
+    let observedIP =
+      if IP4.match(maIP.get()):
+        self.getMostObservedIP(IPv4)
+      else:
+        self.getMostObservedIP(IPv6)
+
+    let newMA =
+      if observedIP.isNone() or maIP.get() == observedIP.get():
+        ma
+      else:
+        observedIP.get() & maWithoutIP.get()
+
+    return Opt.some(newMA)
+  except CatchableError as error:
+    debug "Error while handling manual port forwarding", msg = error.msg
+    return Opt.none(MultiAddress)
+
+proc guessDialableAddrs*(self: PeerStore, listenAddrs: seq[MultiAddress]): seq[MultiAddress] =
+  for l in listenAddrs:
+    let guess = self.replaceMAIpByMostObserved(l)
+    if guess.isSome():
+      result.add(guess.get())

libp2p/protocols/connectivity/autonat/service.nim

@@ -15,6 +15,7 @@ else:
 import std/[options, deques, sequtils]
 import chronos, metrics
 import ../../../switch
+import ../../../wire
 import client
 import ../../../utils/heartbeat
 import ../../../crypto/crypto
@@ -27,6 +28,7 @@ declarePublicGauge(libp2p_autonat_reachability_confidence, "autonat reachability
 type
   AutonatService* = ref object of Service
     newConnectedPeerHandler: PeerEventHandler
+    addressMapper: AddressMapper
     scheduleHandle: Future[void]
     networkReachability: NetworkReachability
     confidence: Option[float]
@@ -133,6 +135,7 @@ proc askPeer(self: AutonatService, switch: Switch, peerId: PeerId): Future[Netwo
   await self.handleAnswer(ans)
   if not isNil(self.statusAndConfidenceHandler):
     await self.statusAndConfidenceHandler(self.networkReachability, self.confidence)
+  await switch.peerInfo.update()
   return ans
 
 proc askConnectedPeers(self: AutonatService, switch: Switch) {.async.} =
@@ -153,7 +156,30 @@ proc schedule(service: AutonatService, switch: Switch, interval: Duration) {.asy
   heartbeat "Scheduling AutonatService run", interval:
     await service.run(switch)
 
+proc addressMapper(
+  self: AutonatService,
+  peerStore: PeerStore,
+  listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]] {.gcsafe, async.} =
+
+  var addrs = newSeq[MultiAddress]()
+  for listenAddr in listenAddrs:
+    var processedMA = listenAddr
+    try:
+      let hostIP = initTAddress(listenAddr).get()
+      if not hostIP.isGlobal():
+        if self.networkReachability == NetworkReachability.Reachable:
+          let maOpt = peerStore.replaceMAIpByMostObserved(listenAddr) # handle manual port forwarding
+          if maOpt.isSome():
+            processedMA = maOpt.get()
+    except CatchableError as exc:
+      debug "Error while handling address mapper", msg = exc.msg
+    addrs.add(processedMA)
+  return addrs
+
 method setup*(self: AutonatService, switch: Switch): Future[bool] {.async.} =
+  self.addressMapper = proc (listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]] {.gcsafe, async.} =
+    return await addressMapper(self, switch.peerStore, listenAddrs)
+
   info "Setting up AutonatService"
   let hasBeenSetup = await procCall Service(self).setup(switch)
   if hasBeenSetup:
@@ -163,14 +189,14 @@ method setup*(self: AutonatService, switch: Switch): Future[bool] {.async.} =
       switch.connManager.addPeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
     if self.scheduleInterval.isSome():
       self.scheduleHandle = schedule(self, switch, self.scheduleInterval.get())
+    switch.peerInfo.addressMappers.add(self.addressMapper)
   return hasBeenSetup
 
 method run*(self: AutonatService, switch: Switch) {.async, public.} =
   trace "Running AutonatService"
   await askConnectedPeers(self, switch)
   await self.callHandler()
 
-
 method stop*(self: AutonatService, switch: Switch): Future[bool] {.async, public.} =
   info "Stopping AutonatService"
   let hasBeenStopped = await procCall Service(self).stop(switch)
@@ -180,6 +206,8 @@ method stop*(self: AutonatService, switch: Switch): Future[bool] {.async, public
       self.scheduleHandle = nil
     if not isNil(self.newConnectedPeerHandler):
       switch.connManager.removePeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+    switch.peerInfo.addressMappers.keepItIf(it != self.addressMapper)
+    await switch.peerInfo.update()
   return hasBeenStopped
 
 proc statusAndConfidenceHandler*(self: AutonatService, statusAndConfidenceHandler: StatusAndConfidenceHandler) =

libp2p/protocols/identify.nim

@@ -26,7 +26,10 @@ import ../protobuf/minprotobuf,
        ../multiaddress,
        ../protocols/protocol,
        ../utility,
-       ../errors
+       ../errors,
+       ../observedaddrmanager
+
+export observedaddrmanager
 
 logScope:
   topics = "libp2p identify"
@@ -56,6 +59,7 @@ type
   Identify* = ref object of LPProtocol
     peerInfo*: PeerInfo
     sendSignedPeerRecord*: bool
+    observedAddrManager*: ObservedAddrManager
 
   IdentifyPushHandler* = proc (
     peer: PeerId,
@@ -160,7 +164,8 @@ proc new*(
   ): T =
   let identify = T(
     peerInfo: peerInfo,
-    sendSignedPeerRecord: sendSignedPeerRecord
+    sendSignedPeerRecord: sendSignedPeerRecord,
+    observedAddrManager: ObservedAddrManager.new(),
   )
   identify.init()
   identify
@@ -182,7 +187,7 @@ method init*(p: Identify) =
   p.handler = handle
   p.codec = IdentifyCodec
 
-proc identify*(p: Identify,
+proc identify*(self: Identify,
                conn: Connection,
                remotePeerId: PeerId): Future[IdentifyInfo] {.async, gcsafe.} =
   trace "initiating identify", conn
@@ -194,23 +199,24 @@ proc identify*(p: Identify,
   let infoOpt = decodeMsg(message)
   if infoOpt.isNone():
     raise newException(IdentityInvalidMsgError, "Incorrect message received!")
-  result = infoOpt.get()
-
-  if result.pubkey.isSome:
-    let peer = PeerId.init(result.pubkey.get())
-    if peer.isErr:
-      raise newException(IdentityInvalidMsgError, $peer.error)
-    else:
-      result.peerId = peer.get()
-      if peer.get() != remotePeerId:
-        trace "Peer ids don't match",
-              remote = peer,
-              local = remotePeerId
-
-        raise newException(IdentityNoMatchError, "Peer ids don't match")
-  else:
+
+  var info = infoOpt.get()
+  if info.pubkey.isNone():
     raise newException(IdentityInvalidMsgError, "No pubkey in identify")
 
+  let peer = PeerId.init(info.pubkey.get())
+  if peer.isErr:
+    raise newException(IdentityInvalidMsgError, $peer.error)
+
+  if peer.get() != remotePeerId:
+    trace "Peer ids don't match", remote = peer, local = remotePeerId
+    raise newException(IdentityNoMatchError, "Peer ids don't match")
+  info.peerId = peer.get()
+
+  if info.observedAddr.isSome:
+    self.observedAddrManager.addObservation(info.observedAddr.get())
+  return info
+
 proc new*(T: typedesc[IdentifyPush], handler: IdentifyPushHandler = nil): T {.public.} =
   ## Create a IdentifyPush protocol. `handler` will be called every time
   ## a peer sends us new `PeerInfo`
@@ -254,3 +260,7 @@ proc push*(p: IdentifyPush, peerInfo: PeerInfo, conn: Connection) {.async, publi
   ## Send new `peerInfo`s to a connection
   var pb = encodeMsg(peerInfo, conn.observedAddr, true)
   await conn.writeLp(pb.buffer)
+
+proc getMostObservedIP*(self: Identify, ipVersion: IPVersion): Opt[MultiAddress] =
+  ## Returns the most observed IP address or none if the number of observations are less than minCount.
+  return self.observedAddrManager.getMostObservedIP(ipVersion)

libp2p/services/autorelayservice.nim

@@ -32,9 +32,15 @@ type
     backingOff: seq[PeerId]
     peerAvailable: AsyncEvent
     onReservation: OnReservationHandler
+    addressMapper: AddressMapper
     rng: ref HmacDrbgContext
 
-proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, selfPid: PeerId) {.async.} =
+proc addressMapper(
+  self: AutoRelayService,
+  listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]] {.gcsafe, async.} =
+  return concat(toSeq(self.relayAddresses.values))
+
+proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, switch: Switch) {.async.} =
   while self.running:
     let
       rsvp = await self.client.reserve(relayPid).wait(chronos.seconds(5))
@@ -46,11 +52,15 @@ proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, selfPid: PeerId)
       break
     if relayPid notin self.relayAddresses or self.relayAddresses[relayPid] != relayedAddr:
       self.relayAddresses[relayPid] = relayedAddr
+      await switch.peerInfo.update()
       if not self.onReservation.isNil():
         self.onReservation(concat(toSeq(self.relayAddresses.values)))
     await sleepAsync chronos.seconds(ttl - 30)
 
 method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsafe.} =
+  self.addressMapper = proc (listenAddrs: seq[MultiAddress]): Future[seq[MultiAddress]] {.gcsafe, async.} =
+    return await addressMapper(self, listenAddrs)
+
   let hasBeenSetUp = await procCall Service(self).setup(switch)
   if hasBeenSetUp:
     proc handlePeerJoined(peerId: PeerId, event: PeerEvent) {.async.} =
@@ -63,6 +73,7 @@ method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcs
         future[].cancel()
     switch.addPeerEventHandler(handlePeerJoined, Joined)
     switch.addPeerEventHandler(handlePeerLeft, Left)
+    switch.peerInfo.addressMappers.add(self.addressMapper)
     await self.run(switch)
   return hasBeenSetUp
 
@@ -96,7 +107,7 @@ proc innerRun(self: AutoRelayService, switch: Switch) {.async, gcsafe.} =
     for relayPid in connectedPeers:
       if self.relayPeers.len() >= self.numRelays:
         break
-      self.relayPeers[relayPid] = self.reserveAndUpdate(relayPid, switch.peerInfo.peerId)
+      self.relayPeers[relayPid] = self.reserveAndUpdate(relayPid, switch)
 
     if self.relayPeers.len() > 0:
       await one(toSeq(self.relayPeers.values())) or self.peerAvailable.wait()
@@ -116,6 +127,8 @@ method stop*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsa
   if hasBeenStopped:
     self.running = false
     self.runner.cancel()
+    switch.peerInfo.addressMappers.keepItIf(it != self.addressMapper)
+    await switch.peerInfo.update()
   return hasBeenStopped
 
 proc getAddresses*(self: AutoRelayService): seq[MultiAddress] =

tests/asyncunit.nim

@@ -1,6 +1,6 @@
-import unittest2
+import unittest2, chronos
 
-export unittest2
+export unittest2, chronos
 
 template asyncTeardown*(body: untyped): untyped =
   teardown:

tests/testautonatservice.nim

@@ -104,9 +104,13 @@ suite "Autonat Service":
     check autonatService.networkReachability() == NetworkReachability.Reachable
     check libp2p_autonat_reachability_confidence.value(["Reachable"]) == 0.3
 
+    check switch1.peerInfo.addrs == switch1.peerStore.guessDialableAddrs(switch1.peerInfo.listenAddrs)
+
     await allFuturesThrowing(
       switch1.stop(), switch2.stop(), switch3.stop(), switch4.stop())
 
+    check switch1.peerInfo.addrs == switch1.peerInfo.listenAddrs
+
   asyncTest "Peer must be not reachable and then reachable":
 
     let autonatClientStub = AutonatClientStub.new(expectedDials = 6)