From cf2dd918717d9b6ef383f5104135cfdde014fe5d Mon Sep 17 00:00:00 2001 From: darshankabariya Date: Fri, 17 Jul 2026 16:17:04 +0530 Subject: [PATCH] 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 --- logos_delivery/waku/node/waku_node.nim | 22 ++++++++++++++- logos_delivery/waku/rln/rln.nim | 7 +++-- .../waku_store_sync/protocols_metrics.nim | 3 +++ .../waku/waku_store_sync/transfer.nim | 25 ++++++++++++++++- tests/waku_store_sync/test_full_node_sync.nim | 27 ++++++++++++++++++- 5 files changed, 79 insertions(+), 5 deletions(-) diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index 31ed0cc68..3dd9a6a32 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/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 diff --git a/logos_delivery/waku/rln/rln.nim b/logos_delivery/waku/rln/rln.nim index 633d433dd..3933f2aac 100644 --- a/logos_delivery/waku/rln/rln.nim +++ b/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, diff --git a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim b/logos_delivery/waku/waku_store_sync/protocols_metrics.nim index 53595f931..02b0df9b1 100644 --- a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim +++ b/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"] diff --git a/logos_delivery/waku/waku_store_sync/transfer.nim b/logos_delivery/waku/waku_store_sync/transfer.nim index 5d20afb18..caa7b92a5 100644 --- a/logos_delivery/waku/waku_store_sync/transfer.nim +++ b/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() diff --git a/tests/waku_store_sync/test_full_node_sync.nim b/tests/waku_store_sync/test_full_node_sync.nim index 6f27013e8..6f1377406 100644 --- a/tests/waku_store_sync/test_full_node_sync.nim +++ b/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)