1818 metrics,
1919 metrics/ chronos_httpserver
2020import
21+ libp2p/ crypto/ rng as libp2p_rng,
2122 libp2p/ [
2223 switch, # manage transports, a single entry point for dialing and listening
2324 crypto/ crypto, # cryptographic functions
@@ -29,12 +30,13 @@ import
2930 peerid, # Implement how peers interact
3031 protobuf/ minprotobuf, # message serialisation/deserialisation from and to protobufs
3132 nameresolving/ dnsresolver,
32- protocols/ mix/ curve25519,
33- protocols/ mix/ mix_protocol,
3433 ] # define DNS resolution
34+ # libp2p_mix has been extracted into its own package; import from there.
35+ import libp2p_mix/ curve25519, libp2p_mix/ mix_protocol
3536import
3637 logos_delivery/ waku/ [
3738 waku_core,
39+ waku_core/ topics/ sharding,
3840 waku_lightpush/ common,
3941 waku_lightpush/ rpc,
4042 waku_enr,
4749 common/ utils/ nat,
4850 waku_store/ common,
4951 waku_filter_v2/ client,
52+ waku_filter_v2/ common as filter_common,
53+ waku_mix/ protocol,
5054 common/ logging,
5155 ],
5256 ./ config_chat2mix
@@ -57,6 +61,105 @@ import ../../logos_delivery/waku/waku_rln_relay
5761logScope:
5862 topics = " chat2 mix"
5963
64+ # ########################
65+ # # Mix Spam Protection ##
66+ # ########################
67+
68+ # Forward declaration
69+ proc maintainSpamProtectionSubscription (
70+ node: WakuNode , contentTopics: seq [ContentTopic ]
71+ ) {.async .}
72+
73+ proc setupMixSpamProtectionViaFilter (node: WakuNode ) {.async .} =
74+ # # Setup filter-based spam protection coordination for mix protocol.
75+ # # Since chat2mix doesn't use relay, we subscribe via filter to receive
76+ # # spam protection coordination messages.
77+
78+ # Register message handler for spam protection coordination
79+ let spamTopics = node.wakuMix.getSpamProtectionContentTopics ()
80+
81+ proc handleSpamMessage (
82+ pubsubTopic: PubsubTopic , message: WakuMessage
83+ ): Future [void ] {.async , gcsafe .} =
84+ await node.wakuMix.handleMessage (pubsubTopic, message)
85+
86+ node.wakuFilterClient.registerPushHandler (handleSpamMessage)
87+
88+ # Wait for filter peer and maintain subscription
89+ asyncSpawn maintainSpamProtectionSubscription (node, spamTopics)
90+
91+ proc maintainSpamProtectionSubscription (
92+ node: WakuNode , contentTopics: seq [ContentTopic ]
93+ ) {.async .} =
94+ # # Maintain filter subscription for spam protection topics.
95+ # # Monitors subscription health with periodic pings and re-subscribes on failure.
96+ const RetryInterval = chronos.seconds (5 )
97+ const SubscriptionMaintenance = chronos.seconds (30 )
98+ const MaxFailedSubscribes = 3
99+ var currentFilterPeer: Option [RemotePeerInfo ] = none (RemotePeerInfo )
100+ var noFailedSubscribes = 0
101+
102+ while true :
103+ # Select or reuse filter peer
104+ if currentFilterPeer.isNone ():
105+ let filterPeerOpt = node.peerManager.selectPeer (WakuFilterSubscribeCodec )
106+ if filterPeerOpt.isNone ():
107+ debug " No filter peer available yet for spam protection, retrying..."
108+ await sleepAsync (RetryInterval )
109+ continue
110+ currentFilterPeer = some (filterPeerOpt.get ())
111+ info " Selected filter peer for spam protection"
112+ peer = currentFilterPeer.get ().peerId
113+
114+ # Check if subscription is still alive with ping
115+ let pingErr = (await node.wakuFilterClient.ping (currentFilterPeer.get ())).errorOr:
116+ # Subscription is alive, wait before next check
117+ await sleepAsync (SubscriptionMaintenance )
118+ if noFailedSubscribes > 0 :
119+ noFailedSubscribes = 0
120+ continue
121+
122+ # Subscription lost, need to re-subscribe
123+ warn " Spam protection filter subscription ping failed, re-subscribing"
124+ error = pingErr, peer = currentFilterPeer.get ().peerId
125+
126+ # Determine pubsub topic from content topics (using auto-sharding)
127+ if node.wakuAutoSharding.isNone ():
128+ error " Auto-sharding not configured, cannot determine pubsub topic for spam protection"
129+ await sleepAsync (RetryInterval )
130+ continue
131+
132+ let shardRes = node.wakuAutoSharding.get ().getShard (contentTopics[0 ])
133+ if shardRes.isErr ():
134+ error " Failed to determine shard for spam protection" = shardRes.error
135+ await sleepAsync (RetryInterval )
136+ continue
137+
138+ let shard = shardRes.get ()
139+ let pubsubTopic: PubsubTopic = shard # converter toPubsubTopic
140+
141+ # Subscribe to spam protection topics
142+ let res = await node.wakuFilterClient.subscribe (
143+ currentFilterPeer.get (), pubsubTopic, contentTopics
144+ )
145+ if res.isErr ():
146+ noFailedSubscribes += 1
147+ warn " Failed to subscribe to spam protection topics via filter"
148+ error = res.error, topics = contentTopics, failCount = noFailedSubscribes
149+
150+ if noFailedSubscribes >= MaxFailedSubscribes :
151+ # Try with a different peer
152+ warn " Max subscription failures reached, selecting new filter peer"
153+ currentFilterPeer = none (RemotePeerInfo )
154+ noFailedSubscribes = 0
155+
156+ await sleepAsync (RetryInterval )
157+ else :
158+ info " Successfully subscribed to spam protection topics via filter"
159+ topics = contentTopics, peer = currentFilterPeer.get ().peerId
160+ noFailedSubscribes = 0
161+ await sleepAsync (SubscriptionMaintenance )
162+
60163const Help = """
61164 Commands: /[?|help|connect|nick|exit]
62165 help: Prints this help
@@ -209,20 +312,21 @@ proc publish(c: Chat, line: string) {.async.} =
209312 try :
210313 if not c.node.wakuLightpushClient.isNil ():
211314 # Attempt lightpush with mix
212-
213- (
214- waitFor c.node.lightpushPublish (
215- some (c.conf.getPubsubTopic (c.node, c.contentTopic)),
216- message,
217- none (RemotePeerInfo ),
218- true ,
219- )
220- ).isOkOr:
221- error " failed to publish lightpush message" = error
315+ let res = await c.node.lightpushPublish (
316+ some (c.conf.getPubsubTopic (c.node, c.contentTopic)),
317+ message,
318+ none (RemotePeerInfo ),
319+ true ,
320+ )
321+ if res.isErr ():
322+ error " failed to publish lightpush message" = res.error
323+ echo " Error: " & res.error.desc.get (" unknown error"
222324 else :
223325 error " failed to publish message as lightpush client is not initialized"
326+ echo " Error: lightpush client is not initialized"
224327 except CatchableError :
225328 error " caught error publishing message: " = getCurrentExceptionMsg ()
329+ echo " Error: " & getCurrentExceptionMsg ()
226330
227331# TODO This should read or be subscribe handler subscribe
228332proc readAndPrint (c: Chat ) {.async .} =
@@ -396,7 +500,7 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
396500 if conf.nodekey.isSome ():
397501 conf.nodekey.get ()
398502 else :
399- PrivateKey .random (Secp256k1 , rng[] ).tryGet ()
503+ PrivateKey .random (Secp256k1 , libp2p_rng. newBearSslRng ( rng) ).tryGet ()
400504
401505 # set log level
402506 if conf.logLevel != LogLevel .NONE :
@@ -451,7 +555,11 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
451555 error " failed to generate mix key pair" = error
452556 return
453557
454- (await node.mountMix (conf.clusterId, mixPrivKey, conf.mixnodes)).isOkOr:
558+ (
559+ await node.mountMix (
560+ conf.clusterId, mixPrivKey, conf.mixnodes, some (conf.rlnUserMessageLimit)
561+ )
562+ ).isOkOr:
455563 error " failed to mount waku mix protocol: " = $ error
456564 quit (QuitFailure )
457565
@@ -467,12 +575,13 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
467575 if kadBootstrapPeers.len > 0 :
468576 node.wakuKademlia = WakuKademlia .new (
469577 node.switch,
470- ExtendedKademliaDiscoveryParams
578+ ExtendedServiceDiscoveryParams
471579 bootstrapNodes: kadBootstrapPeers,
472580 mixPubKey: some (mixPubKey),
473581 advertiseMix: false ,
474582 ),
475583 node.peerManager,
584+ rng = libp2p_rng.newBearSslRng (node.rng),
476585 getMixNodePoolSize = proc (): int {.gcsafe , raises : [].} =
477586 if node.wakuMix.isNil ():
478587 0
@@ -486,6 +595,10 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
486595
487596 # await node.mountRendezvousClient(conf.clusterId)
488597
598+ # Subscribe to spam protection coordination topics via filter since chat2mix doesn't use relay
599+ if not node.wakuFilterClient.isNil ():
600+ asyncSpawn setupMixSpamProtectionViaFilter (node)
601+
489602 await node.start ()
490603
491604 node.peerManager.start ()
@@ -675,7 +788,7 @@ proc main(rng: ref HmacDrbgContext) {.async.} =
675788 raise e
676789
677790when isMainModule : # isMainModule = true when the module is compiled as the main file
678- let rng = crypto. newRng ()
791+ let rng = HmacDrbgContext . new ()
679792 try :
680793 waitFor (main (rng))
681794 except CatchableError as e:
0 commit comments