2024-03-14 16:48:09 +00:00
|
|
|
|
{.used.}
|
|
|
|
|
|
|
|
|
|
import
|
|
|
|
|
std/[options, sequtils, tables],
|
|
|
|
|
testutils/unittests,
|
|
|
|
|
chronos,
|
|
|
|
|
chronicles,
|
|
|
|
|
stew/shims/net,
|
|
|
|
|
libp2p/[switch, peerId, crypto/crypto, multistream, muxers/muxer],
|
|
|
|
|
eth/[keys, p2p/discoveryv5/enr]
|
|
|
|
|
|
|
|
|
|
import
|
|
|
|
|
../../../waku/[
|
|
|
|
|
waku_node,
|
|
|
|
|
node/peer_manager,
|
|
|
|
|
waku_discv5,
|
|
|
|
|
waku_peer_exchange,
|
|
|
|
|
waku_peer_exchange/rpc,
|
|
|
|
|
waku_peer_exchange/rpc_codec,
|
|
|
|
|
waku_peer_exchange/protocol,
|
|
|
|
|
node/peer_manager,
|
|
|
|
|
waku_relay/protocol,
|
|
|
|
|
waku_relay,
|
|
|
|
|
waku_core,
|
2024-03-15 23:08:47 +00:00
|
|
|
|
waku_core/message/codec,
|
2024-03-14 16:48:09 +00:00
|
|
|
|
],
|
|
|
|
|
../testlib/[wakucore, wakunode, simple_mock, assertions],
|
|
|
|
|
./utils.nim
|
|
|
|
|
|
|
|
|
|
suite "Waku Peer Exchange":
|
|
|
|
|
# Some of this tests use node.wakuPeerExchange instead of just a standalone PeerExchange.
|
|
|
|
|
# This is because attempts to connect the switches for two standalones PeerExchanges failed.
|
|
|
|
|
# TODO: Try to make the tests work with standalone PeerExchanges
|
|
|
|
|
|
|
|
|
|
suite "request":
|
|
|
|
|
asyncTest "Retrieve and provide peer exchange peers from discv5":
|
|
|
|
|
## Given (copied from test_waku_discv5.nim)
|
|
|
|
|
let
|
|
|
|
|
# todo: px flag
|
2024-03-15 23:08:47 +00:00
|
|
|
|
flags = CapabilitiesBitfield.init(
|
|
|
|
|
lightpush = false, filter = false, store = false, relay = true
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
bindIp = parseIpAddress("0.0.0.0")
|
|
|
|
|
extIp = parseIpAddress("127.0.0.1")
|
|
|
|
|
|
|
|
|
|
nodeKey1 = generateSecp256k1Key()
|
|
|
|
|
nodeTcpPort1 = Port(64010)
|
|
|
|
|
nodeUdpPort1 = Port(9000)
|
2024-03-15 23:08:47 +00:00
|
|
|
|
node1 = newTestWakuNode(
|
|
|
|
|
nodeKey1,
|
|
|
|
|
bindIp,
|
|
|
|
|
nodeTcpPort1,
|
|
|
|
|
some(extIp),
|
|
|
|
|
wakuFlags = some(flags),
|
|
|
|
|
discv5UdpPort = some(nodeUdpPort1),
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
|
|
|
|
nodeKey2 = generateSecp256k1Key()
|
|
|
|
|
nodeTcpPort2 = Port(64012)
|
|
|
|
|
nodeUdpPort2 = Port(9002)
|
2024-03-15 23:08:47 +00:00
|
|
|
|
node2 = newTestWakuNode(
|
|
|
|
|
nodeKey2,
|
|
|
|
|
bindIp,
|
|
|
|
|
nodeTcpPort2,
|
|
|
|
|
some(extIp),
|
|
|
|
|
wakuFlags = some(flags),
|
|
|
|
|
discv5UdpPort = some(nodeUdpPort2),
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
|
|
|
|
nodeKey3 = generateSecp256k1Key()
|
|
|
|
|
nodeTcpPort3 = Port(64014)
|
|
|
|
|
nodeUdpPort3 = Port(9004)
|
2024-03-15 23:08:47 +00:00
|
|
|
|
node3 = newTestWakuNode(
|
|
|
|
|
nodeKey3,
|
|
|
|
|
bindIp,
|
|
|
|
|
nodeTcpPort3,
|
|
|
|
|
some(extIp),
|
|
|
|
|
wakuFlags = some(flags),
|
|
|
|
|
discv5UdpPort = some(nodeUdpPort3),
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
|
|
|
|
# discv5
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let conf1 = WakuDiscoveryV5Config(
|
|
|
|
|
discv5Config: none(DiscoveryConfig),
|
|
|
|
|
address: bindIp,
|
|
|
|
|
port: nodeUdpPort1,
|
|
|
|
|
privateKey: keys.PrivateKey(nodeKey1.skkey),
|
|
|
|
|
bootstrapRecords: @[],
|
|
|
|
|
autoupdateRecord: true,
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let disc1 =
|
|
|
|
|
WakuDiscoveryV5.new(node1.rng, conf1, some(node1.enr), some(node1.peerManager))
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let conf2 = WakuDiscoveryV5Config(
|
|
|
|
|
discv5Config: none(DiscoveryConfig),
|
|
|
|
|
address: bindIp,
|
|
|
|
|
port: nodeUdpPort2,
|
|
|
|
|
privateKey: keys.PrivateKey(nodeKey2.skkey),
|
|
|
|
|
bootstrapRecords: @[disc1.protocol.getRecord()],
|
|
|
|
|
autoupdateRecord: true,
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let disc2 =
|
|
|
|
|
WakuDiscoveryV5.new(node2.rng, conf2, some(node2.enr), some(node2.peerManager))
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
|
|
|
|
await allFutures(node1.start(), node2.start(), node3.start())
|
|
|
|
|
let resultDisc1StartRes = await disc1.start()
|
|
|
|
|
assert resultDisc1StartRes.isOk(), resultDisc1StartRes.error
|
|
|
|
|
let resultDisc2StartRes = await disc2.start()
|
|
|
|
|
assert resultDisc2StartRes.isOk(), resultDisc2StartRes.error
|
|
|
|
|
|
|
|
|
|
## When
|
|
|
|
|
var attempts = 10
|
|
|
|
|
while (disc1.protocol.nodesDiscovered < 1 or disc2.protocol.nodesDiscovered < 1) and
|
|
|
|
|
attempts > 0:
|
|
|
|
|
await sleepAsync(1.seconds)
|
|
|
|
|
attempts -= 1
|
|
|
|
|
|
|
|
|
|
# node2 can be connected, so will be returned by peer exchange
|
|
|
|
|
require (
|
|
|
|
|
await node1.peerManager.connectRelay(node2.switch.peerInfo.toRemotePeerInfo())
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
# Mount peer exchange
|
|
|
|
|
await node1.mountPeerExchange()
|
|
|
|
|
await node3.mountPeerExchange()
|
|
|
|
|
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let dialResponse =
|
|
|
|
|
await node3.dialForPeerExchange(node1.switch.peerInfo.toRemotePeerInfo())
|
2024-03-14 16:48:09 +00:00
|
|
|
|
let response = dialResponse.get()
|
|
|
|
|
|
|
|
|
|
## Then
|
|
|
|
|
check:
|
|
|
|
|
response.get().peerInfos.len == 1
|
|
|
|
|
response.get().peerInfos[0].enr == disc2.protocol.localNode.record.raw
|
|
|
|
|
|
|
|
|
|
await allFutures(
|
|
|
|
|
[node1.stop(), node2.stop(), node3.stop(), disc1.stop(), disc2.stop()]
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
asyncTest "Request returns some discovered peers":
|
|
|
|
|
let
|
|
|
|
|
node1 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
node2 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
|
|
|
|
|
# Start and mount peer exchange
|
|
|
|
|
await allFutures([node1.start(), node2.start()])
|
|
|
|
|
await allFutures([node1.mountPeerExchange(), node2.mountPeerExchange()])
|
|
|
|
|
|
|
|
|
|
# Create connection
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let connOpt = await node2.peerManager.dialPeer(
|
|
|
|
|
node1.switch.peerInfo.toRemotePeerInfo(), WakuPeerExchangeCodec
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
require:
|
|
|
|
|
connOpt.isSome
|
|
|
|
|
|
|
|
|
|
# Create some enr and add to peer exchange (simulating disv5)
|
|
|
|
|
var enr1, enr2 = enr.Record()
|
|
|
|
|
check enr1.fromUri(
|
|
|
|
|
"enr:-Iu4QGNuTvNRulF3A4Kb9YHiIXLr0z_CpvWkWjWKU-o95zUPR_In02AWek4nsSk7G_-YDcaT4bDRPzt5JIWvFqkXSNcBgmlkgnY0gmlwhE0WsGeJc2VjcDI1NmsxoQKp9VzU2FAh7fwOwSpg1M_Ekz4zzl0Fpbg6po2ZwgVwQYN0Y3CC6mCFd2FrdTIB"
|
|
|
|
|
)
|
|
|
|
|
check enr2.fromUri(
|
|
|
|
|
"enr:-Iu4QGJllOWlviPIh_SGR-VVm55nhnBIU5L-s3ran7ARz_4oDdtJPtUs3Bc5aqZHCiPQX6qzNYF2ARHER0JPX97TFbEBgmlkgnY0gmlwhE0WsGeJc2VjcDI1NmsxoQP3ULycvday4EkvtVu0VqbBdmOkbfVLJx8fPe0lE_dRkIN0Y3CC6mCFd2FrdTIB"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
# Mock that we have discovered these enrs
|
|
|
|
|
node1.wakuPeerExchange.enrCache.add(enr1)
|
|
|
|
|
node1.wakuPeerExchange.enrCache.add(enr2)
|
|
|
|
|
|
|
|
|
|
# Request 2 peer from px. Test all request variants
|
|
|
|
|
let response1 = await node2.wakuPeerExchange.request(2)
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let response2 =
|
|
|
|
|
await node2.wakuPeerExchange.request(2, node1.peerInfo.toRemotePeerInfo())
|
2024-03-14 16:48:09 +00:00
|
|
|
|
let response3 = await node2.wakuPeerExchange.request(2, connOpt.get())
|
|
|
|
|
|
|
|
|
|
# Check the response or dont even continue
|
|
|
|
|
require:
|
|
|
|
|
response1.isOk
|
|
|
|
|
response2.isOk
|
|
|
|
|
response3.isOk
|
|
|
|
|
|
|
|
|
|
check:
|
|
|
|
|
response1.get().peerInfos.len == 2
|
|
|
|
|
response2.get().peerInfos.len == 2
|
|
|
|
|
response3.get().peerInfos.len == 2
|
|
|
|
|
|
|
|
|
|
# Since it can return duplicates test that at least one of the enrs is in the response
|
|
|
|
|
response1.get().peerInfos.anyIt(it.enr == enr1.raw) or
|
|
|
|
|
response1.get().peerInfos.anyIt(it.enr == enr2.raw)
|
|
|
|
|
response2.get().peerInfos.anyIt(it.enr == enr1.raw) or
|
|
|
|
|
response2.get().peerInfos.anyIt(it.enr == enr2.raw)
|
|
|
|
|
response3.get().peerInfos.anyIt(it.enr == enr1.raw) or
|
|
|
|
|
response3.get().peerInfos.anyIt(it.enr == enr2.raw)
|
|
|
|
|
|
|
|
|
|
asyncTest "Request fails gracefully":
|
|
|
|
|
let
|
|
|
|
|
node1 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
node2 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
|
|
|
|
|
# Start and mount peer exchange
|
|
|
|
|
await allFutures([node1.start(), node2.start()])
|
|
|
|
|
await allFutures([node1.mountPeerExchange(), node2.mountPeerExchange()])
|
|
|
|
|
|
|
|
|
|
# Create connection
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let connOpt = await node2.peerManager.dialPeer(
|
|
|
|
|
node1.switch.peerInfo.toRemotePeerInfo(), WakuPeerExchangeCodec
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
require connOpt.isSome
|
|
|
|
|
|
|
|
|
|
# Force closing the connection to simulate a failed peer
|
|
|
|
|
await connOpt.get().close()
|
|
|
|
|
|
|
|
|
|
# Request 2 peer from px
|
|
|
|
|
let response = await node1.wakuPeerExchange.request(2, connOpt.get())
|
|
|
|
|
|
|
|
|
|
# Check that it failed gracefully
|
|
|
|
|
check:
|
|
|
|
|
response.isErr
|
|
|
|
|
|
|
|
|
|
asyncTest "Request 0 peers, with 0 peers in PeerExchange":
|
|
|
|
|
# Given a disconnected PeerExchange
|
|
|
|
|
let
|
|
|
|
|
switch = newTestSwitch()
|
|
|
|
|
peerManager = PeerManager.new(switch)
|
|
|
|
|
peerExchange = WakuPeerExchange.new(peerManager)
|
|
|
|
|
|
|
|
|
|
# When requesting 0 peers
|
|
|
|
|
let response = await peerExchange.request(0)
|
|
|
|
|
|
|
|
|
|
# Then the response should be an error
|
|
|
|
|
check:
|
|
|
|
|
response.isErr
|
|
|
|
|
response.error == "peer_not_found_failure"
|
|
|
|
|
|
|
|
|
|
asyncTest "Request 0 peers, with 1 peer in PeerExchange":
|
|
|
|
|
# Given two valid nodes with PeerExchange
|
|
|
|
|
let
|
|
|
|
|
node1 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
node2 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
|
|
|
|
|
# Start and mount peer exchange
|
|
|
|
|
await allFutures([node1.start(), node2.start()])
|
|
|
|
|
await allFutures([node1.mountPeerExchange(), node2.mountPeerExchange()])
|
|
|
|
|
|
|
|
|
|
# Connect the nodes
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let dialResponse = await node2.peerManager.dialPeer(
|
|
|
|
|
node1.switch.peerInfo.toRemotePeerInfo(), WakuPeerExchangeCodec
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
assert dialResponse.isSome
|
|
|
|
|
|
|
|
|
|
# Mock that we have discovered one enr
|
|
|
|
|
var record = enr.Record()
|
|
|
|
|
check record.fromUri(
|
|
|
|
|
"enr:-Iu4QGNuTvNRulF3A4Kb9YHiIXLr0z_CpvWkWjWKU-o95zUPR_In02AWek4nsSk7G_-YDcaT4bDRPzt5JIWvFqkXSNcBgmlkgnY0gmlwhE0WsGeJc2VjcDI1NmsxoQKp9VzU2FAh7fwOwSpg1M_Ekz4zzl0Fpbg6po2ZwgVwQYN0Y3CC6mCFd2FrdTIB"
|
|
|
|
|
)
|
|
|
|
|
node1.wakuPeerExchange.enrCache.add(record)
|
|
|
|
|
|
|
|
|
|
# When requesting 0 peers
|
|
|
|
|
let response = await node1.wakuPeerExchange.request(0)
|
|
|
|
|
|
|
|
|
|
# Then the response should be empty
|
|
|
|
|
assertResultOk(response)
|
|
|
|
|
check response.get().peerInfos.len == 0
|
|
|
|
|
|
|
|
|
|
asyncTest "Request with invalid peer info":
|
|
|
|
|
# Given two valid nodes with PeerExchange
|
|
|
|
|
let
|
|
|
|
|
node1 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
node2 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
|
|
|
|
|
# Start and mount peer exchange
|
|
|
|
|
await allFutures([node1.start(), node2.start()])
|
|
|
|
|
await allFutures([node1.mountPeerExchange(), node2.mountPeerExchange()])
|
|
|
|
|
|
|
|
|
|
# Mock that we have discovered one enr
|
|
|
|
|
var record = enr.Record()
|
|
|
|
|
check record.fromUri(
|
|
|
|
|
"enr:-Iu4QGNuTvNRulF3A4Kb9YHiIXLr0z_CpvWkWjWKU-o95zUPR_In02AWek4nsSk7G_-YDcaT4bDRPzt5JIWvFqkXSNcBgmlkgnY0gmlwhE0WsGeJc2VjcDI1NmsxoQKp9VzU2FAh7fwOwSpg1M_Ekz4zzl0Fpbg6po2ZwgVwQYN0Y3CC6mCFd2FrdTIB"
|
|
|
|
|
)
|
|
|
|
|
node1.wakuPeerExchange.enrCache.add(record)
|
|
|
|
|
|
|
|
|
|
# When making any request with an invalid peer info
|
|
|
|
|
var remotePeerInfo2 = node2.peerInfo.toRemotePeerInfo()
|
|
|
|
|
remotePeerInfo2.peerId.data.add(255.byte)
|
|
|
|
|
let response = await node1.wakuPeerExchange.request(1, remotePeerInfo2)
|
|
|
|
|
|
|
|
|
|
# Then the response should be an error
|
|
|
|
|
check:
|
|
|
|
|
response.isErr
|
|
|
|
|
response.error == "dial_failure"
|
|
|
|
|
|
|
|
|
|
asyncTest "Connections are closed after response is sent":
|
|
|
|
|
# Create 3 nodes
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let nodes = toSeq(0 ..< 3).mapIt(
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
|
|
|
|
|
await allFutures(nodes.mapIt(it.start()))
|
|
|
|
|
await allFutures(nodes.mapIt(it.mountPeerExchange()))
|
|
|
|
|
|
|
|
|
|
# Multiple nodes request to node 0
|
2024-03-15 23:08:47 +00:00
|
|
|
|
for i in 1 ..< 3:
|
|
|
|
|
let resp = await nodes[i].wakuPeerExchange.request(
|
|
|
|
|
2, nodes[0].switch.peerInfo.toRemotePeerInfo()
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
require resp.isOk
|
|
|
|
|
|
|
|
|
|
# Wait for streams to be closed
|
|
|
|
|
await sleepAsync(1.seconds)
|
|
|
|
|
|
|
|
|
|
# Check that all streams are closed for px
|
|
|
|
|
check:
|
|
|
|
|
nodes[0].peerManager.getNumStreams(WakuPeerExchangeCodec) == (0, 0)
|
|
|
|
|
nodes[1].peerManager.getNumStreams(WakuPeerExchangeCodec) == (0, 0)
|
|
|
|
|
nodes[2].peerManager.getNumStreams(WakuPeerExchangeCodec) == (0, 0)
|
|
|
|
|
|
|
|
|
|
suite "Protocol Handler":
|
|
|
|
|
asyncTest "Works as expected":
|
|
|
|
|
let
|
|
|
|
|
node1 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
node2 =
|
|
|
|
|
newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
|
|
|
|
|
|
|
|
# Start and mount peer exchange
|
|
|
|
|
await allFutures([node1.start(), node2.start()])
|
|
|
|
|
await allFutures([node1.mountPeerExchange(), node2.mountPeerExchange()])
|
|
|
|
|
|
|
|
|
|
# Mock that we have discovered these enrs
|
|
|
|
|
var enr1 = enr.Record()
|
|
|
|
|
check enr1.fromUri(
|
|
|
|
|
"enr:-Iu4QGNuTvNRulF3A4Kb9YHiIXLr0z_CpvWkWjWKU-o95zUPR_In02AWek4nsSk7G_-YDcaT4bDRPzt5JIWvFqkXSNcBgmlkgnY0gmlwhE0WsGeJc2VjcDI1NmsxoQKp9VzU2FAh7fwOwSpg1M_Ekz4zzl0Fpbg6po2ZwgVwQYN0Y3CC6mCFd2FrdTIB"
|
|
|
|
|
)
|
|
|
|
|
node1.wakuPeerExchange.enrCache.add(enr1)
|
|
|
|
|
|
|
|
|
|
# Create connection
|
2024-03-15 23:08:47 +00:00
|
|
|
|
let connOpt = await node2.peerManager.dialPeer(
|
|
|
|
|
node1.switch.peerInfo.toRemotePeerInfo(), WakuPeerExchangeCodec
|
|
|
|
|
)
|
2024-03-14 16:48:09 +00:00
|
|
|
|
require connOpt.isSome
|
|
|
|
|
let conn = connOpt.get()
|
|
|
|
|
|
|
|
|
|
# Send bytes so that they directly hit the handler
|
|
|
|
|
let rpc = PeerExchangeRpc(request: PeerExchangeRequest(numPeers: 1))
|
|
|
|
|
|
|
|
|
|
var buffer: seq[byte]
|
|
|
|
|
await conn.writeLP(rpc.encode().buffer)
|
|
|
|
|
buffer = await conn.readLp(MaxRpcSize.int)
|
|
|
|
|
|
|
|
|
|
# Decode the response
|
|
|
|
|
let decodedBuff = PeerExchangeRpc.decode(buffer)
|
|
|
|
|
require decodedBuff.isOk
|
|
|
|
|
|
|
|
|
|
# Check we got back the enr we mocked
|
|
|
|
|
check:
|
|
|
|
|
decodedBuff.get().response.peerInfos.len == 1
|
|
|
|
|
decodedBuff.get().response.peerInfos[0].enr == enr1.raw
|