mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-06-04 13:09:32 +00:00
549834203d
* bump libp2p pin to release/v2.0.0 (c43199378) * pin nimble.lock: lsquic/websock/boringssl/protobuf_serialization/npeg/jwt * add libp2p_mix dep and point libp2p/protocols/mix -> libp2p_mix * migrate rng to libp2p Rng type (prod, channels, noise, tests) * noise: take Rng, extract bearSslDrbg internally * waku_switch: TransportConfig factory; withMaxInOut; local MaxConnections * waku_relay/rendezvous/discv5/kademlia: v2.0.0 API (rng, config, ServiceDiscovery) * tests: newStandardSwitch shim; PeerId.random(rng); common.rng()/crypto.newRng() * drop libp2p/utils/semaphore (use chronos AsyncSemaphore) * add waku/compat/option_valueor shim where needed * add std/options where transitive re-export dropped
107 lines
3.3 KiB
Nim
107 lines
3.3 KiB
Nim
{.used.}
|
|
|
|
import
|
|
std/[strutils, sequtils, tempfiles, options],
|
|
stew/byteutils,
|
|
chronos,
|
|
chronicles,
|
|
libp2p/switch,
|
|
libp2p/protocols/pubsub/pubsub
|
|
|
|
import brokers/broker_context
|
|
|
|
from std/times import epochTime
|
|
|
|
import
|
|
waku/[
|
|
waku_relay, node/waku_node, node/peer_manager, waku_core, waku_node, waku_rln_relay
|
|
],
|
|
../waku_store/store_utils,
|
|
../waku_archive/archive_utils,
|
|
../testlib/[wakucore, futures]
|
|
|
|
proc noopRawHandler*(): WakuRelayHandler =
|
|
var handler: WakuRelayHandler
|
|
handler = proc(topic: PubsubTopic, msg: WakuMessage): Future[void] {.async, gcsafe.} =
|
|
discard
|
|
handler
|
|
|
|
proc newTestWakuRelay*(switch = newTestSwitch()): Future[WakuRelay] {.async.} =
|
|
let proto = WakuRelay.new(switch).tryGet()
|
|
|
|
let protocolMatcher = proc(proto: string): bool {.gcsafe.} =
|
|
return proto.startsWith(WakuRelayCodec)
|
|
|
|
switch.mount(proto, protocolMatcher)
|
|
|
|
return proto
|
|
|
|
proc setupRln*(node: WakuNode, identifier: uint) {.async.} =
|
|
await node.mountRlnRelay(
|
|
WakuRlnConfig(dynamic: false, credIndex: some(identifier), epochSizeSec: 1)
|
|
)
|
|
|
|
proc subscribeToContentTopicWithHandler*(
|
|
node: WakuNode, contentTopic: string
|
|
): Future[bool] =
|
|
var completionFut = newFuture[bool]()
|
|
proc relayHandler(
|
|
topic: PubsubTopic, msg: WakuMessage
|
|
): Future[void] {.async, gcsafe.} =
|
|
if topic == topic:
|
|
completionFut.complete(true)
|
|
|
|
(node.subscribe((kind: ContentSub, topic: contentTopic), relayHandler)).isOkOr:
|
|
error "Failed to subscribe to content topic", error
|
|
completionFut.complete(true)
|
|
return completionFut
|
|
|
|
proc subscribeCompletionHandler*(node: WakuNode, pubsubTopic: string): Future[bool] =
|
|
var completionFut = newFuture[bool]()
|
|
proc relayHandler(
|
|
topic: PubsubTopic, msg: WakuMessage
|
|
): Future[void] {.async, gcsafe.} =
|
|
if topic == pubsubTopic:
|
|
completionFut.complete(true)
|
|
|
|
(node.subscribe((kind: PubsubSub, topic: pubsubTopic), relayHandler)).isOkOr:
|
|
error "Failed to subscribe to pubsub topic", error
|
|
completionFut.complete(false)
|
|
return completionFut
|
|
|
|
proc sendRlnMessage*(
|
|
client: WakuNode,
|
|
pubsubTopic: string,
|
|
contentTopic: string,
|
|
completionFuture: Future[bool],
|
|
payload: seq[byte] = "Hello".toBytes(),
|
|
): Future[bool] {.async.} =
|
|
var message = WakuMessage(payload: payload, contentTopic: contentTopic)
|
|
doAssert(client.wakuRlnRelay.appendRLNProof(message, epochTime()).isOk())
|
|
discard await client.publish(some(pubsubTopic), message)
|
|
let isCompleted = await completionFuture.withTimeout(FUTURE_TIMEOUT)
|
|
return isCompleted
|
|
|
|
proc sendRlnMessageWithInvalidProof*(
|
|
client: WakuNode,
|
|
pubsubTopic: string,
|
|
contentTopic: string,
|
|
completionFuture: Future[bool],
|
|
payload: seq[byte] = "Hello".toBytes(),
|
|
): Future[bool] {.async.} =
|
|
let
|
|
extraBytes: seq[byte] = @[byte(1), 2, 3]
|
|
rateLimitProofRes = client.wakuRlnRelay.groupManager.generateProof(
|
|
concat(payload, extraBytes),
|
|
# we add extra bytes to invalidate proof verification against original payload
|
|
client.wakuRlnRelay.getCurrentEpoch(),
|
|
messageId = MessageId(0),
|
|
)
|
|
rateLimitProof = rateLimitProofRes.get().encode().buffer
|
|
message =
|
|
WakuMessage(payload: @payload, contentTopic: contentTopic, proof: rateLimitProof)
|
|
|
|
discard await client.publish(some(pubsubTopic), message)
|
|
let isCompleted = await completionFuture.withTimeout(FUTURE_TIMEOUT)
|
|
return isCompleted
|