feat: validate RLN proofs on store sync transfer ingress

Closes the transfer.nim TODO: received transfer messages pass through an
injected validator before syncMessageIngress; rejects are dropped and
counted (total_transfer_messages_rejected). mountStoreSync wires a
nil-guarded closure that runs Rln.validateMessage with the new
checkFreshness=false switch - synced messages are old by design, so the
timestamp-recency bound is skipped while proof, membership root and
timestamp/epoch binding still verify. Known gap: proofs against roots
older than the acceptable root window are rejected, bounding history
sync across heavy membership churn.

Part of the "Store as a startup-only dependency" experiment (step 8).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

logos_delivery/waku/node/waku_node.nim

@@ -394,8 +394,28 @@ proc mountStoreSync*(
   reconMountRes.isOkOr:
     return err(error.msg)
 
+  let transferValidator: TransferValidator = proc(
+      msg: WakuMessage
+  ): Future[bool] {.async.} =
+    ## RLN mounts after store sync (but before the switch starts accepting
+    ## connections), so capture the node and check at call time; nil rln
+    ## means RLN is not configured on this network. Freshness is not
+    ## checked: synced messages are old by design. Known gap: proofs built
+    ## against roots older than the acceptable root window are rejected,
+    ## so history sync across heavy membership churn is bounded by it.
+    if node.rln.isNil():
+      return true
+
+    let res = await node.rln.validateMessage(msg, checkFreshness = false)
+    return res == MessageValidationResult.Valid
+
   let transfer = SyncTransfer.new(
-    node.peerManager, node.wakuArchive, idsChannel, wantsChannel, needsChannel
+    node.peerManager,
+    node.wakuArchive,
+    idsChannel,
+    wantsChannel,
+    needsChannel,
+    msgValidator = Opt.some(transferValidator),
   )
 
   node.wakuStoreTransfer = transfer

logos_delivery/waku/rln/rln.nim

@@ -48,7 +48,7 @@ proc stop*(rlnPeer: Rln) {.async: (raises: [Exception]).} =
   await rlnPeer.groupManager.stop()
 
 proc validateMessage*(
-    rlnPeer: Rln, msg: WakuMessage
+    rlnPeer: Rln, msg: WakuMessage, checkFreshness = true
 ): Future[MessageValidationResult] {.async.} =
   ## validate the supplied `msg` based on the waku-rln-relay routing protocol i.e.,
   ## the `msg`'s epoch is within MaxEpochGap of the current epoch
@@ -56,6 +56,9 @@ proc validateMessage*(
   ## the `msg` does not violate the rate limit
   ## `timeOption` indicates Unix epoch time (fractional part holds sub-seconds)
   ## if `timeOption` is supplied, then the current epoch is calculated based on that
+  ## `checkFreshness = false` skips the timestamp-recency bound: used for
+  ## messages that are legitimately old, e.g. received via store sync
+  ## transfer. Proof, root and timestamp/epoch binding are still verified.
 
   let proof = RateLimitProof.init(msg.proof).valueOr:
     return MessageValidationResult.Invalid
@@ -72,7 +75,7 @@ proc validateMessage*(
   info "time info",
     currentTime = currentTime, messageTime = messageTime, msgHash = msg.hash
 
-  if timeDiff > rlnPeer.rlnMaxTimestampGap:
+  if checkFreshness and timeDiff > rlnPeer.rlnMaxTimestampGap:
     warn "invalid message: timestamp difference exceeds threshold",
       timeDiff = timeDiff,
       maxTimestampGap = rlnPeer.rlnMaxTimestampGap,

logos_delivery/waku/waku_store_sync/protocols_metrics.nim

@@ -17,6 +17,9 @@ declarePublicHistogram reconciliation_differences,
 declarePublicCounter total_bytes_exchanged,
   "the number of bytes sent and received by the protocols", ["protocol", "direction"]
 
+declarePublicCounter total_transfer_messages_rejected,
+  "number of received transfer messages dropped by validation"
+
 declarePublicCounter total_transfer_messages_exchanged,
   "the number of messages sent and received by the transfer protocol", ["direction"]
 

logos_delivery/waku/waku_store_sync/transfer.nim

@@ -31,9 +31,14 @@ import
 logScope:
   topics = "waku transfer"
 
+type TransferValidator* = proc(msg: WakuMessage): Future[bool] {.gcsafe, raises: [].}
+  ## Returns false when a received transfer message must be dropped
+  ## (e.g. failed RLN proof verification).
+
 type SyncTransfer* = ref object of LPProtocol
   wakuArchive: WakuArchive
   peerManager: PeerManager
+  msgValidator: Opt[TransferValidator]
 
   # Send IDs to reconciliation protocol for storage
   idsTx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)]
@@ -169,8 +174,24 @@ proc initProtocolHandler(self: SyncTransfer) =
 
       let hash = computeMessageHash(pubsub, msg)
 
+      if self.msgValidator.isSome():
+        let valid =
+          try:
+            await self.msgValidator.get()(msg)
+          except CancelledError as exc:
+            raise exc
+          except CatchableError:
+            error "transfer message validation error",
+              remote_peer_id = conn.peerId, error = getCurrentExceptionMsg()
+            false
+
+        if not valid:
+          total_transfer_messages_rejected.inc()
+          warn "transfer message failed validation, dropping",
+            remote_peer_id = conn.peerId
+          continue
+
       try:
-        #TODO verify msg RLN proof...
         (await self.wakuArchive.syncMessageIngress(hash, pubsub, msg)).isOkOr:
           error "failed to archive message", error = $error
           continue
@@ -202,6 +223,7 @@ proc new*(
     idsTx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)],
     localWantsRx: AsyncQueue[PeerId],
     remoteNeedsRx: AsyncQueue[(PeerId, WakuMessageHash)],
+    msgValidator: Opt[TransferValidator] = Opt.none(TransferValidator),
 ): T =
   var transfer = SyncTransfer(
     peerManager: peerManager,
@@ -209,6 +231,7 @@ proc new*(
     idsTx: idsTx,
     localWantsRx: localWantsRx,
     remoteNeedsRx: remoteNeedsRx,
+    msgValidator: msgValidator,
   )
 
   transfer.initProtocolHandler()

tests/waku_store_sync/test_full_node_sync.nim

@@ -31,7 +31,9 @@ type FullNode = object
   peerInfo: RemotePeerInfo
   peerManager: PeerManager
 
-proc newFullNode(): Future[FullNode] {.async.} =
+proc newFullNode(
+    msgValidator: Opt[TransferValidator] = Opt.none(TransferValidator)
+): Future[FullNode] {.async.} =
   ## Mirrors the wiring of WakuNode.mountStoreSync: one peer manager and
   ## three shared channels connect reconciliation and transfer.
   let switch = newTestSwitch()
@@ -68,6 +70,7 @@ proc newFullNode(): Future[FullNode] {.async.} =
     idsTx = idsChannel,
     localWantsRx = wantsChannel,
     remoteNeedsRx = needsChannel,
+    msgValidator = msgValidator,
   )
 
   await transfer.start()
@@ -153,3 +156,25 @@ suite "Waku Sync: full node miss recovery":
     check:
       await nodeA.hasMessage(hashB)
       await nodeB.hasMessage(hashA)
+
+  asyncTest "messages failing transfer validation are not archived":
+    await nodeB.stop()
+
+    let rejectAll: TransferValidator = proc(msg: WakuMessage): Future[bool] {.async.} =
+      return false
+
+    nodeB = await newFullNode(msgValidator = Opt.some(rejectAll))
+    nodeA.peerManager.addPeer(nodeB.peerInfo)
+    nodeB.peerManager.addPeer(nodeA.peerInfo)
+
+    let msg = fakeWakuMessage(contentTopic = DefaultContentTopic)
+    let hash = computeMessageHash(DefaultPubsubTopic, msg)
+
+    nodeA.insertMessage(msg)
+
+    let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo))
+    assert res.isOk(), $res.error
+
+    await sleepAsync(1.seconds)
+
+    check not await nodeB.hasMessage(hash)