mirror of
https://github.com/logos-storage/logos-storage-nim.git
synced 2026-05-12 06:19:33 +00:00
Add relay integration and tests
This commit is contained in:
Arnaud
parent
bc05f8154d
commit
e48ca913a9
@ -18,6 +18,7 @@ import pkg/chronos
|
||||
import pkg/chronicles
|
||||
import pkg/libp2p
|
||||
import pkg/libp2p/protocols/connectivity/autonatv2/service
|
||||
import pkg/libp2p/services/autorelayservice
|
||||
|
||||
import ./utils
|
||||
import ./utils/natutils
|
||||
@ -66,12 +67,18 @@ type PrefSrcStatus = enum
|
||||
type NatMapper* = ref object of RootObj
|
||||
|
||||
method mapNatAddresses*(
|
||||
m: NatMapper, addrs: seq[MultiAddress], discoveryPort: Port
|
||||
m: NatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.base, gcsafe, raises: [].} =
|
||||
raiseAssert "mapNatAddresses not implemented"
|
||||
|
||||
method getReachableAddresses*(
|
||||
m: NatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.base, gcsafe, raises: [].} =
|
||||
raiseAssert "getReachableAddresses not implemented"
|
||||
|
||||
type DefaultNatMapper* = ref object of NatMapper
|
||||
natConfig*: NatConfig
|
||||
discoveryPort*: Port
|
||||
|
||||
## Also does threadvar initialisation.
|
||||
## Must be called before redirectPorts() in each thread.
|
||||
@ -451,29 +458,70 @@ proc nattedAddress*(
|
||||
(newAddrs, discoveryAddrs)
|
||||
|
||||
method mapNatAddresses*(
|
||||
m: DefaultNatMapper, addrs: seq[MultiAddress], discoveryPort: Port
|
||||
m: DefaultNatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.gcsafe, raises: [].} =
|
||||
nattedAddress(m.natConfig, addrs, discoveryPort)
|
||||
nattedAddress(m.natConfig, addrs, m.discoveryPort)
|
||||
|
||||
method getReachableAddresses*(
|
||||
m: DefaultNatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.gcsafe, raises: [].} =
|
||||
let ip =
|
||||
if m.natConfig.hasExtIp:
|
||||
some(m.natConfig.extIp)
|
||||
else:
|
||||
let (routeIp, _) = getRoutePrefSrc(static parseIpAddress("0.0.0.0"))
|
||||
routeIp
|
||||
if ip.isNone:
|
||||
return (@[], @[])
|
||||
let announceAddrs =
|
||||
addrs.mapIt(it.remapAddr(ip = ip, port = none(Port))).deduplicate()
|
||||
(announceAddrs, @[getMultiAddrWithIPAndUDPPort(ip.get, m.discoveryPort)])
|
||||
|
||||
proc hasPublicIp*(addrs: seq[MultiAddress]): bool =
|
||||
for addr in addrs:
|
||||
let (ip, _) = getAddressAndPort(addr)
|
||||
if ip.isSome and isGlobalUnicast(ip.get):
|
||||
return true
|
||||
|
||||
proc handleNatStatus*(
|
||||
networkReachability: NetworkReachability,
|
||||
confidence: Opt[float],
|
||||
mapper: NatMapper,
|
||||
listenAddrs: seq[MultiAddress],
|
||||
discoveryPort: Port,
|
||||
discovery: Discovery,
|
||||
switch: Switch,
|
||||
autoRelayService: AutoRelayService,
|
||||
) {.async: (raises: [CancelledError]).} =
|
||||
debug "AutoNAT status", reachability = networkReachability, confidence
|
||||
|
||||
case networkReachability
|
||||
of Reachable:
|
||||
# TODO: switch DHT to server mode, stop relay if running
|
||||
discard
|
||||
of NotReachable:
|
||||
let (announceAddrs, discoveryAddrs) =
|
||||
mapper.mapNatAddresses(listenAddrs, discoveryPort)
|
||||
discovery.updateAnnounceRecord(announceAddrs)
|
||||
discovery.updateDhtRecord(announceAddrs & discoveryAddrs)
|
||||
of Unknown:
|
||||
# Nothing to do here, not enough confidence score result
|
||||
discard
|
||||
of Reachable:
|
||||
# For UPnP, it the mapping was a success,
|
||||
# the autorelay service has been stopped
|
||||
# and the address was already announced
|
||||
if autoRelayService.isRunning:
|
||||
if not await autoRelayService.stop(switch):
|
||||
debug "AutoRelayService stop method returned false"
|
||||
|
||||
let (announceAddrs, discoveryAddrs) =
|
||||
mapper.getReachableAddresses(switch.peerInfo.addrs)
|
||||
discovery.updateAnnounceRecord(announceAddrs)
|
||||
discovery.updateDhtRecord(announceAddrs & discoveryAddrs)
|
||||
# TODO: switch DHT to server mode
|
||||
of NotReachable:
|
||||
let (announceAddrs, discoveryAddrs) = mapper.mapNatAddresses(switch.peerInfo.addrs)
|
||||
|
||||
# With a UPnP / NatPmP successful mapping,
|
||||
# we suppose that having a public IP make it Reachable.
|
||||
# If not, the state will be updated in the next Autonat iteration.
|
||||
# TODO: Do we need to manually call dialMe to make sure we are Reachable ?
|
||||
if hasPublicIp(announceAddrs):
|
||||
discovery.updateAnnounceRecord(announceAddrs)
|
||||
discovery.updateDhtRecord(announceAddrs & discoveryAddrs)
|
||||
|
||||
if autoRelayService.isRunning:
|
||||
if not await autoRelayService.stop(switch):
|
||||
debug "AutoRelayService stop method returned false"
|
||||
else:
|
||||
if not autoRelayService.isRunning:
|
||||
if not await autoRelayService.setup(switch):
|
||||
debug "AutoRelayService setup method returned false"
|
||||
|
||||
@ -18,6 +18,8 @@ import pkg/taskpools
|
||||
import pkg/presto
|
||||
import pkg/libp2p
|
||||
import pkg/libp2p/protocols/connectivity/autonatv2/[service, client]
|
||||
import pkg/libp2p/protocols/connectivity/relay/client as relayClientModule
|
||||
import pkg/libp2p/services/autorelayservice
|
||||
import pkg/confutils
|
||||
import pkg/confutils/defs
|
||||
import pkg/stew/io2
|
||||
@ -54,6 +56,7 @@ type
|
||||
maintenance: BlockMaintainer
|
||||
taskpool: Taskpool
|
||||
autonatService*: AutonatV2Service
|
||||
autoRelayService: AutoRelayService
|
||||
isStarted: bool
|
||||
|
||||
StoragePrivateKey* = libp2p.PrivateKey # alias
|
||||
@ -195,6 +198,8 @@ proc new*(
|
||||
## create StorageServer including setting up datastore, repostore, etc
|
||||
let listenMultiAddr = getMultiAddrWithIpAndTcpPort(config.listenIp, config.listenPort)
|
||||
|
||||
let relayClient = relayClientModule.RelayClient.new()
|
||||
|
||||
let autonatClient = AutonatV2Client.new(random.Rng.instance())
|
||||
let autonatService = AutonatV2Service.new(
|
||||
rng = random.Rng.instance(),
|
||||
@ -220,6 +225,7 @@ proc new*(
|
||||
.withSignedPeerRecord(true)
|
||||
.withTcpTransport({ServerFlags.ReuseAddr, ServerFlags.TcpNoDelay})
|
||||
.withAutonatV2Server()
|
||||
.withCircuitRelay(relayClient)
|
||||
.withServices(@[Service(autonatService)])
|
||||
.build()
|
||||
|
||||
@ -327,6 +333,16 @@ proc new*(
|
||||
taskPool = taskPool,
|
||||
)
|
||||
|
||||
autoRelayService = AutoRelayService.new(
|
||||
maxNumRelays = config.natMaxRelays,
|
||||
client = relayClient,
|
||||
onReservation = proc(addresses: seq[MultiAddress]) {.gcsafe, raises: [].} =
|
||||
debug "Relay reservation updated", addresses
|
||||
discovery.updateAnnounceRecord(addresses)
|
||||
discovery.updateDhtRecord(addresses),
|
||||
rng = random.Rng.instance(),
|
||||
)
|
||||
|
||||
var restServer: RestServerRef = nil
|
||||
|
||||
if config.apiBindAddress.isSome:
|
||||
@ -344,14 +360,15 @@ proc new*(
|
||||
switch.mount(network)
|
||||
switch.mount(manifestProto)
|
||||
|
||||
let natMapper = DefaultNatMapper(natConfig: config.nat)
|
||||
let natMapper =
|
||||
DefaultNatMapper(natConfig: config.nat, discoveryPort: config.discoveryPort)
|
||||
autonatService.setStatusAndConfidenceHandler(
|
||||
proc(
|
||||
networkReachability: NetworkReachability, confidence: Opt[float]
|
||||
) {.async: (raises: [CancelledError]).} =
|
||||
debug "AutoNAT status", reachability = networkReachability, confidence
|
||||
await handleNatStatus(
|
||||
networkReachability, confidence, natMapper, switch.peerInfo.addrs,
|
||||
config.discoveryPort, discovery,
|
||||
networkReachability, natMapper, discovery, switch, autoRelayService
|
||||
)
|
||||
)
|
||||
|
||||
@ -364,4 +381,5 @@ proc new*(
|
||||
taskPool: taskPool,
|
||||
logFile: logFile,
|
||||
autonatService: autonatService,
|
||||
autoRelayService: autoRelayService,
|
||||
)
|
||||
|
||||
@ -1,10 +1,15 @@
|
||||
import std/[unittest, net]
|
||||
import std/net
|
||||
import pkg/chronos
|
||||
import pkg/libp2p
|
||||
import pkg/libp2p/[multiaddress, multihash, multicodec]
|
||||
import pkg/libp2p/protocols/connectivity/autonatv2/service
|
||||
import pkg/libp2p/protocols/connectivity/autonatv2/service except setup
|
||||
import pkg/libp2p/protocols/connectivity/autonatv2/types
|
||||
import pkg/libp2p/protocols/connectivity/relay/client as relayClientModule
|
||||
import pkg/libp2p/services/autorelayservice except setup
|
||||
|
||||
import pkg/results
|
||||
|
||||
import ./helpers
|
||||
import ../asynctest
|
||||
import ../../storage/nat
|
||||
import ../../storage/discovery
|
||||
import ../../storage/rng
|
||||
@ -15,7 +20,12 @@ type MockNatMapper = ref object of NatMapper
|
||||
mapped: tuple[libp2p, discovery: seq[MultiAddress]]
|
||||
|
||||
method mapNatAddresses*(
|
||||
m: MockNatMapper, addrs: seq[MultiAddress], discoveryPort: Port
|
||||
m: MockNatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.raises: [].} =
|
||||
m.mapped
|
||||
|
||||
method getReachableAddresses*(
|
||||
m: MockNatMapper, addrs: seq[MultiAddress]
|
||||
): tuple[libp2p, discovery: seq[MultiAddress]] {.raises: [].} =
|
||||
m.mapped
|
||||
|
||||
@ -44,7 +54,6 @@ suite "NAT Address Tests":
|
||||
MultiAddress.init("/ip4/8.8.8.8/tcp/5000").expect("valid multiaddr"),
|
||||
MultiAddress.init("/ip4/8.8.8.8/tcp/5000").expect("valid multiaddr"),
|
||||
]
|
||||
|
||||
#ipv6Addr = MultiAddress.init("/ip6/::1/tcp/5000").expect("valid multiaddr")
|
||||
addrs = @[localAddr, anyAddr, publicAddr]
|
||||
|
||||
@ -78,17 +87,94 @@ suite "setupAddress":
|
||||
check tcpPort == some(Port(5000))
|
||||
check udpPort == some(Port(5001))
|
||||
|
||||
suite "handleNatStatus":
|
||||
let key = PrivateKey.random(Rng.instance[]).get()
|
||||
suite "getReachableAddresses":
|
||||
test "returns remapped addresses when extIp is configured":
|
||||
let
|
||||
natConfig = NatConfig(hasExtIp: true, extIp: parseIpAddress("1.2.3.4"))
|
||||
mapper = DefaultNatMapper(natConfig: natConfig, discoveryPort: Port(8090))
|
||||
listenAddr = MultiAddress.init("/ip4/0.0.0.0/tcp/5000").expect("valid")
|
||||
|
||||
test "NotReachable updates announce addresses":
|
||||
let disc = Discovery.new(key, announceAddrs = @[])
|
||||
let (libp2pAddrs, discAddrs) = mapper.getReachableAddresses(@[listenAddr])
|
||||
|
||||
check libp2pAddrs == @[MultiAddress.init("/ip4/1.2.3.4/tcp/5000").expect("valid")]
|
||||
check discAddrs == @[MultiAddress.init("/ip4/1.2.3.4/udp/8090").expect("valid")]
|
||||
|
||||
suite "hasPublicIp":
|
||||
test "hasPublicIp returns true when the address is public":
|
||||
let ma = MultiAddress.init("/ip4/8.8.8.8/tcp/8080").expect("valid")
|
||||
check hasPublicIp(@[ma])
|
||||
|
||||
test "hasPublicIp returns false when the address is private":
|
||||
let ma = MultiAddress.init("/ip4/192.168.1.1/tcp/8080").expect("valid")
|
||||
check not hasPublicIp(@[ma])
|
||||
|
||||
test "hasPublicIp returns false when the address is empty":
|
||||
check not hasPublicIp(@[])
|
||||
|
||||
asyncchecksuite "handleNatStatus":
|
||||
var sw: Switch
|
||||
var key: PrivateKey
|
||||
var disc: Discovery
|
||||
let autoRelay =
|
||||
AutoRelayService.new(1, relayClientModule.RelayClient.new(), nil, Rng.instance())
|
||||
|
||||
setup:
|
||||
key = PrivateKey.random(Rng.instance[]).get()
|
||||
disc = Discovery.new(key, announceAddrs = @[])
|
||||
sw = newStandardSwitch()
|
||||
await sw.start()
|
||||
|
||||
teardown:
|
||||
await sw.stop()
|
||||
|
||||
if autoRelay.isRunning:
|
||||
discard await autoRelay.stop(sw)
|
||||
|
||||
test "handleNatStatus announces address when the node is not Reachable and the UPnP succeed with public ip":
|
||||
let announceAddr = MultiAddress.init("/ip4/1.2.3.4/tcp/8080").expect("valid")
|
||||
let discAddr = MultiAddress.init("/ip4/1.2.3.4/udp/8090").expect("valid")
|
||||
let mapper = MockNatMapper(mapped: (@[announceAddr], @[discAddr]))
|
||||
|
||||
waitFor handleNatStatus(
|
||||
NotReachable, Opt.none(float), mapper, @[], Port(8090), disc
|
||||
)
|
||||
await handleNatStatus(NotReachable, mapper, disc, sw, autoRelay)
|
||||
|
||||
check disc.announceAddrs == @[announceAddr]
|
||||
check not autoRelay.isRunning
|
||||
|
||||
# test "handleNatStatus does not announce address when the node is not Reachable and the UPnP succeed with private ip":
|
||||
# let privateAddr = MultiAddress.init("/ip4/192.168.1.1/tcp/8080").expect("valid")
|
||||
# let mapper = MockNatMapper(mapped: (@[privateAddr], @[]))
|
||||
|
||||
# await handleNatStatus(
|
||||
# NotReachable, mapper, disc, sw, autoRelay
|
||||
# )
|
||||
|
||||
# check disc.announceAddrs == @[]
|
||||
# check not autoRelay.isRunning
|
||||
|
||||
test "handleNatStatus starts autoRelay when node is not Reachable and UPnP failed":
|
||||
let mapper = MockNatMapper(mapped: (@[], @[]))
|
||||
|
||||
await handleNatStatus(NotReachable, mapper, disc, sw, autoRelay)
|
||||
|
||||
check autoRelay.isRunning
|
||||
# The addresses will be announced in the onReservation callback
|
||||
# after a node accepted a Relay reservation.
|
||||
|
||||
test "handleNatStatus does not announce address when node is Reachable and relay is not running":
|
||||
let mapper = MockNatMapper(mapped: (@[], @[]))
|
||||
|
||||
await handleNatStatus(Reachable, mapper, disc, sw, autoRelay)
|
||||
|
||||
check disc.announceAddrs == newSeq[MultiAddress]()
|
||||
check not autoRelay.isRunning
|
||||
|
||||
test "handleNatStatus stops relay and announces address when node is Reachable and relay is running":
|
||||
let announceAddr = MultiAddress.init("/ip4/1.2.3.4/tcp/8080").expect("valid")
|
||||
let discAddr = MultiAddress.init("/ip4/1.2.3.4/udp/8090").expect("valid")
|
||||
let mapper = MockNatMapper(mapped: (@[announceAddr], @[discAddr]))
|
||||
|
||||
discard await autorelayservice.setup(autoRelay, sw)
|
||||
await handleNatStatus(Reachable, mapper, disc, sw, autoRelay)
|
||||
|
||||
check not autoRelay.isRunning
|
||||
check disc.announceAddrs == @[announceAddr]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user