feat(mix): DoS protection integration

.gitignore

@@ -3,6 +3,11 @@
 # Executables shall be put in an ignored build/ directory
 /build
 
+# Test binaries (built by `nim c tests/...nim` for local debug compile)
+/tests/all_tests_common
+/tests/all_tests_waku
+/tests/all_tests_wakunode2
+
 # Generated Files
 *.generated.nim
 
@@ -41,6 +46,7 @@ node_modules/
 
 # RLN / keystore
 rlnKeystore.json
+rln_keystore*.json
 *.tar.gz
 
 # sqlite db
@@ -91,3 +97,6 @@ nimbledeps
 # Python bytecode from tests/simulator
 __pycache__/
 *.pyc
+
+# sim driver script (local dev tool, not part of build/CI)
+simulations/mixnet/roundtrip_check.sh

apps/chat2mix/chat2mix.nim

@@ -6,7 +6,7 @@ when not (compileOption("threads")):
 
 {.push raises: [].}
 
-import std/[strformat, strutils, times, options, random, sequtils]
+import std/[strformat, strutils, times, options, random, sequtils, sets]
 import
   confutils,
   chronicles,
@@ -31,9 +31,8 @@ import
     protocols/kademlia/types,
     protocols/service_discovery/types as sd_types,
     nameresolving/dnsresolver,
-    protocols/mix/curve25519,
-    protocols/mix/mix_protocol,
   ] # define DNS resolution
+import libp2p_mix/curve25519, libp2p_mix/mix_protocol
 import
   logos_delivery/waku/[
     waku_core,
@@ -51,6 +50,8 @@ import
     common/utils/nat,
     waku_store/common,
     waku_filter_v2/client,
+    waku_filter_v2/common as filter_common,
+    waku_mix/protocol,
     common/logging,
   ],
   ./config_chat2mix
@@ -61,6 +62,99 @@ import ../../logos_delivery/waku/waku_rln_relay
 logScope:
   topics = "chat2 mix"
 
+#########################
+## Mix Spam Protection ##
+#########################
+
+# Forward declaration
+proc maintainSpamProtectionSubscription(
+  node: WakuNode, contentTopics: seq[ContentTopic]
+) {.async.}
+
+proc setupMixSpamProtectionViaFilter(node: WakuNode): Future[void] =
+  ## Registers the spam-protection push handler and returns the long-lived
+  ## subscription-maintenance future so the caller can cancel it on shutdown.
+  # Register message handler for spam protection coordination
+  let spamTopics = node.wakuMix.getSpamProtectionContentTopics()
+
+  proc handleSpamMessage(pubsubTopic: PubsubTopic, message: WakuMessage) {.async.} =
+    await node.wakuMix.handleMessage(pubsubTopic, message)
+
+  node.wakuFilterClient.registerPushHandler(handleSpamMessage)
+
+  # Wait for filter peer and maintain subscription
+  maintainSpamProtectionSubscription(node, spamTopics)
+
+proc maintainSpamProtectionSubscription(
+    node: WakuNode, contentTopics: seq[ContentTopic]
+) {.async.} =
+  const RetryInterval = chronos.seconds(5)
+  const SubscriptionMaintenance = chronos.seconds(30)
+  const MaxFailedSubscribes = 3
+  var currentFilterPeer: Option[RemotePeerInfo] = none(RemotePeerInfo)
+  var noFailedSubscribes = 0
+
+  while true:
+    # Select or reuse filter peer
+    if currentFilterPeer.isNone():
+      let filterPeerOpt = node.peerManager.selectPeer(WakuFilterSubscribeCodec)
+      if filterPeerOpt.isNone():
+        debug "No filter peer available yet for spam protection, retrying..."
+        await sleepAsync(RetryInterval)
+        continue
+      currentFilterPeer = some(filterPeerOpt.get())
+      info "Selected filter peer for spam protection",
+        peer = currentFilterPeer.get().peerId
+
+    # Check if subscription is still alive with ping
+    let pingErr = (await node.wakuFilterClient.ping(currentFilterPeer.get())).errorOr:
+      # Subscription is alive, wait before next check
+      await sleepAsync(SubscriptionMaintenance)
+      if noFailedSubscribes > 0:
+        noFailedSubscribes = 0
+      continue
+
+    # Subscription lost, need to re-subscribe
+    warn "Spam protection filter subscription ping failed, re-subscribing",
+      error = pingErr, peer = currentFilterPeer.get().peerId
+
+    # Determine pubsub topic from content topics (using auto-sharding)
+    if node.wakuAutoSharding.isNone():
+      error "Auto-sharding not configured, cannot determine pubsub topic for spam protection"
+      await sleepAsync(RetryInterval)
+      continue
+
+    let shardRes = node.wakuAutoSharding.get().getShard(contentTopics[0])
+    if shardRes.isErr():
+      error "Failed to determine shard for spam protection", error = shardRes.error
+      await sleepAsync(RetryInterval)
+      continue
+
+    let shard = shardRes.get()
+    let pubsubTopic: PubsubTopic = shard # converter toPubsubTopic
+
+    # Subscribe to spam protection topics
+    let res = await node.wakuFilterClient.subscribe(
+      currentFilterPeer.get(), pubsubTopic, contentTopics
+    )
+    if res.isErr():
+      noFailedSubscribes += 1
+      warn "Failed to subscribe to spam protection topics via filter",
+        error = res.error, topics = contentTopics, failCount = noFailedSubscribes
+
+      if noFailedSubscribes >= MaxFailedSubscribes:
+        # Try with a different peer
+        warn "Max subscription failures reached, selecting new filter peer"
+        currentFilterPeer = none(RemotePeerInfo)
+        noFailedSubscribes = 0
+
+      await sleepAsync(RetryInterval)
+    else:
+      info "Successfully subscribed to spam protection topics via filter",
+        topics = contentTopics, peer = currentFilterPeer.get().peerId
+      noFailedSubscribes = 0
+      await sleepAsync(SubscriptionMaintenance)
+
 const Help = """
   Commands: /[?|help|connect|nick|exit]
   help: Prints this help
@@ -82,6 +176,7 @@ type Chat = ref object
   prompt: bool # chat prompt is showing
   contentTopic: string # default content topic for chat messages
   conf: Chat2Conf # configuration for chat2
+  mixSpamProtectionFut: Future[void] # mix spam-protection filter maintenance loop
 
 type
   PrivateKey* = crypto.PrivateKey
@@ -213,20 +308,20 @@ proc publish(c: Chat, line: string) {.async.} =
   try:
     if not c.node.wakuLightpushClient.isNil():
       # Attempt lightpush with mix
-
-      (
-        waitFor c.node.lightpushPublish(
-          some(c.conf.getPubsubTopic(c.node, c.contentTopic)),
-          message,
-          none(RemotePeerInfo),
-          true,
-        )
-      ).isOkOr:
-        error "failed to publish lightpush message", error = error
+      let res = await c.node.lightpushPublish(
+        some(c.conf.getPubsubTopic(c.node, c.contentTopic)),
+        message,
+        none(RemotePeerInfo),
+        true,
+      )
+      if res.isErr():
+        error "failed to publish lightpush message", error = res.error
     else:
       error "failed to publish message as lightpush client is not initialized"
+      echo "Error: lightpush client is not initialized"
   except CatchableError:
     error "caught error publishing message: ", error = getCurrentExceptionMsg()
+    echo "Error: " & getCurrentExceptionMsg()
 
 # TODO This should read or be subscribe handler subscribe
 proc readAndPrint(c: Chat) {.async.} =
@@ -276,6 +371,9 @@ proc writeAndPrint(c: Chat) {.async.} =
     elif line.startsWith("/exit"):
       echo "quitting..."
 
+      if not c.mixSpamProtectionFut.isNil():
+        await c.mixSpamProtectionFut.cancelAndWait()
+
       try:
         await c.node.stop()
       except:
@@ -455,7 +553,11 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
     error "failed to generate mix key pair", error = error
     return
 
-  (await node.mountMix(conf.clusterId, mixPrivKey, conf.mixnodes)).isOkOr:
+  (
+    await node.mountMix(
+      conf.clusterId, mixPrivKey, conf.mixnodes, some(conf.rlnUserMessageLimit)
+    )
+  ).isOkOr:
     error "failed to mount waku mix protocol: ", error = $error
     quit(QuitFailure)
 
@@ -464,7 +566,7 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
     var kadBootstrapPeers: seq[(PeerId, seq[MultiAddress])]
     for nodeStr in conf.kadBootstrapNodes:
       let (peerId, ma) = block:
-        parseFullAddress(nodeStr).isOkOr:
+        parseFullAddress(nodeStr).valueOr:
           error "Failed to parse kademlia bootstrap node", node = nodeStr, error
           continue
 
@@ -474,7 +576,7 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
       node.mountKademlia(
         KademliaDiscoveryConf(
           bootstrapNodes: kadBootstrapPeers,
-          servicesToDiscover: @[MixProtocolID],
+          servicesToDiscover: toHashSet(@[MixProtocolID]),
           randomLookupInterval: chronos.seconds(60),
           serviceLookupInterval: chronos.seconds(60),
           kadDhtConfig: KadDHTConfig.new(),
@@ -488,6 +590,11 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
 
   #await node.mountRendezvousClient(conf.clusterId)
 
+  # Subscribe to spam protection coordination topics via filter since chat2mix doesn't use relay
+  var mixSpamProtectionFut: Future[void] = nil
+  if not node.wakuFilterClient.isNil():
+    mixSpamProtectionFut = setupMixSpamProtectionViaFilter(node)
+
   await node.start()
 
   node.peerManager.start()
@@ -509,6 +616,7 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
     prompt: false,
     contentTopic: conf.contentTopic,
     conf: conf,
+    mixSpamProtectionFut: mixSpamProtectionFut,
   )
 
   var dnsDiscoveryUrl = none(string)

apps/chat2mix/config_chat2mix.nim

@@ -240,6 +240,13 @@ type
       name: "kad-bootstrap-node"
     .}: seq[string]
 
+    ## RLN spam protection config
+    rlnUserMessageLimit* {.
+      desc: "Maximum messages per epoch for RLN spam protection.",
+      defaultValue: 100,
+      name: "rln-user-message-limit"
+    .}: int
+
 proc parseCmdArg*(T: type MixNodePubInfo, p: string): T =
   let elements = p.split(":")
   if elements.len != 2:

config.nims

@@ -111,6 +111,8 @@ if not defined(macosx) and not defined(android):
       nimStackTraceOverride
     switch("import", "libbacktrace")
 
+switch("import", "logos_delivery/waku/compat/option_valueor")
+
 --define:
   nimOldCaseObjects
   # https://github.com/status-im/nim-confutils/issues/9

logos_delivery.nimble

@@ -62,14 +62,22 @@ requires "nim >= 2.2.4",
 # Packages not on nimble (use git URLs)
 
 requires "https://github.com/logos-messaging/nim-ffi#v0.1.3"
+requires "https://github.com/logos-co/mix-rln-spam-protection-plugin.git#61ee3e5aacb6b224b70e164ef7d0a5714fe66b26"
+
+# nim-libp2p-mix: extracted mix protocol used by the plugin and by waku's
+# mix integration layer. Tip of experiment/drop-nimble-lock (PR #14, stacked
+# on chore/bump-libp2p-v2.0.0). Carries the v2.0.0 bump + sink overrides +
+# AddressConfidence.Infinite + deeper move-semantics propagation + the
+# lockfile-as-build-artefact cleanup. Re-bump to master SHA once #14 lands.
+# The plugin pins the same SHA — keeps the diamond dep collapsed.
+requires "https://github.com/logos-co/nim-libp2p-mix.git#50c4ab4fa788a33eb12a0a2cecaa708873352b58"
 
 requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5"
 
 requires "https://github.com/NagyZoltanPeter/nim-brokers.git#v3.1.1"
 
 requires "https://github.com/vacp2p/nim-lsquic.git#v0.5.1"
 requires "https://github.com/vacp2p/nim-jwt.git#057ec95eb5af0eea9c49bfe9025b3312c95dc5f2"
-requires "https://github.com/logos-co/nim-libp2p-mix#380513117d556bf8f70066f5e72a7fd74fe36ba6"
 
 proc getMyCPU(): string =
   ## Need to set cpu more explicit manner to avoid arch issues between dependencies

logos_delivery/waku/factory/node_factory.nim

@@ -183,17 +183,6 @@ proc setupProtocols(
     node.mountKademlia(kadConf).isOkOr:
       return err("failed to setup service discovery: " & error)
 
-    # Register ServicePeersRequest provider
-    ServicePeersRequest.setProvider(
-      node.brokerCtx,
-      proc(serviceId: string): Future[Result[ServicePeersRequest, string]] {.async.} =
-        let peers = (await node.wakuKademlia.lookupServicePeers(serviceId)).valueOr:
-          return err("failed call to lookupServicePeers: " & error)
-        return ok(ServicePeersRequest(serviceId: serviceId, peers: peers)),
-    ).isOkOr:
-      error "Can't set provider for ServicePeersRequest", error = error
-      return err("Can't set provider for ServicePeersRequest: " & error)
-
   if conf.storeServiceConf.isSome():
     let storeServiceConf = conf.storeServiceConf.get()
 

logos_delivery/waku/node/subscription_manager.nim

@@ -11,6 +11,7 @@ import
     node/node_telemetry,
     waku_relay,
     waku_archive,
+    waku_mix,
     waku_store_sync,
     waku_filter_v2/common as filter_common,
     waku_filter_v2/client as filter_client,
@@ -66,6 +67,12 @@ proc registerRelayHandler(
 
     node.wakuStoreReconciliation.messageIngress(topic, msg)
 
+  proc mixHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
+    if node.wakuMix.isNil():
+      return
+
+    await node.wakuMix.handleMessage(topic, msg)
+
   proc internalHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
     MessageSeenEvent.emit(node.brokerCtx, topic, msg)
 
@@ -76,6 +83,7 @@ proc registerRelayHandler(
     await filterHandler(topic, msg)
     await archiveHandler(topic, msg)
     await syncHandler(topic, msg)
+    await mixHandler(topic, msg)
     await internalHandler(topic, msg)
 
     if node.legacyAppHandlers.hasKey(topic) and not node.legacyAppHandlers[topic].isNil():