diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index 51259504e..e45eff942 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/logos_delivery/waku/api/publish.nim @@ -38,14 +38,10 @@ proc hasLightpush*(self: Waku): bool = return not self.node.wakuLightpushClient.isNil() proc relayPushHandler*(self: Waku): PushMessageHandler = - ## Builds the relay publish handler (appending an RLN proof when RLN is - ## mounted) used by the send pipeline. Caller ensures relay is mounted. - let rln = - if self.node.rln.isNil(): - none[Rln]() - else: - some(self.node.rln) - return getRelayPushHandler(self.node.wakuRelay, rln) + ## Builds the relay publish handler used by the send pipeline. Caller + ## ensures relay is mounted. RLN proof generation is handled client-side + ## in (legacy)lightpushPublish; this handler only validates and republishes. + return getRelayPushHandler(self.node.wakuRelay) proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool = ## True if a lightpush service peer is available for `shard`. diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index 6b9d816fd..1bb4ddcbb 100644 --- a/logos_delivery/waku/node/waku_node/lightpush.nim +++ b/logos_delivery/waku/node/waku_node/lightpush.nim @@ -47,15 +47,7 @@ proc mountLegacyLightPush*( return err(MountWithoutRelayError) info "mounting legacy lightpush with relay" - let rlnPeer = - if node.rln.isNil(): - info "mounting legacy lightpush without rln-relay" - none(Rln) - else: - info "mounting legacy lightpush with rln-relay" - some(node.rln) - let pushHandler = - legacy_lightpush_protocol.getRelayPushHandler(node.wakuRelay, rlnPeer) + let pushHandler = legacy_lightpush_protocol.getRelayPushHandler(node.wakuRelay) node.wakuLegacyLightPush = WakuLegacyLightPush.new(node.peerManager, node.rng, pushHandler, some(rateLimit)) @@ -90,6 +82,19 @@ proc legacyLightpushPublish*( error "failed to publish message as legacy lightpush not available" return err("Waku lightpush not available") + # toRLNSignal includes the timestamp in the proof input, so the timestamp + # must be fixed before proof generation. The downstream ensureTimestampSet + # in the client publish becomes an idempotent no-op safety net. + let message = ensureTimestampSet(message) + + let rln = + if node.rln.isNil(): + none(Rln) + else: + some(node.rln) + let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: + return err("failed call checkAndGenerateRLNProof from lightpush: " & error) + let internalPublish = proc( node: WakuNode, pubsubTopic: PubsubTopic, @@ -115,7 +120,7 @@ proc legacyLightpushPublish*( await node.wakuLegacyLightPush.handleSelfLightPushRequest(pubsubTopic, message) try: if pubsubTopic.isSome(): - return await internalPublish(node, pubsubTopic.get(), message, peer) + return await internalPublish(node, pubsubTopic.get(), msgWithProof, peer) if node.wakuAutoSharding.isNone(): return err("Pubsub topic must be specified when static sharding is enabled") @@ -123,7 +128,7 @@ proc legacyLightpushPublish*( ?node.wakuAutoSharding.get().getShardsFromContentTopics(message.contentTopic) for pubsub, _ in topicMap.pairs: # There's only one pair anyway - return await internalPublish(node, $pubsub, message, peer) + return await internalPublish(node, $pubsub, msgWithProof, peer) except CatchableError: return err(getCurrentExceptionMsg()) @@ -158,14 +163,7 @@ proc mountLightPush*( return err(MountWithoutRelayError) info "mounting lightpush with relay" - let rlnPeer = - if node.rln.isNil(): - info "mounting lightpush without rln-relay" - none(Rln) - else: - info "mounting lightpush with rln-relay" - some(node.rln) - let pushHandler = lightpush_protocol.getRelayPushHandler(node.wakuRelay, rlnPeer) + let pushHandler = lightpush_protocol.getRelayPushHandler(node.wakuRelay) node.wakuLightPush = WakuLightPush.new( node.peerManager, node.rng, pushHandler, node.wakuAutoSharding, some(rateLimit) @@ -282,4 +280,18 @@ proc lightpushPublish*( error "lightpush publish error", error = msg return lighpushErrorResult(LightPushErrorCode.INTERNAL_SERVER_ERROR, msg) - return await lightpushPublishHandler(node, pubsubForPublish, message, toPeer, mixify) + # toRLNSignal includes the timestamp in the proof input, so the timestamp + # must be fixed before proof generation. The downstream ensureTimestampSet + # in the client publish becomes an idempotent no-op safety net. + let message = ensureTimestampSet(message) + + let rln = + if node.rln.isNil(): + none(Rln) + else: + some(node.rln) + let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: + return lighpushErrorResult(LightPushErrorCode.OUT_OF_RLN_PROOF, error) + + return + await lightpushPublishHandler(node, pubsubForPublish, msgWithProof, toPeer, mixify) diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index 8f1a63517..4a0c332c1 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -1,23 +1,23 @@ {.push raises: [].} -import std/[times, sequtils] -import chronos, results, stew/byteutils +import std/[options, times, sequtils] +import chronos, chronicles, results, stew/byteutils import ./types, ./protocol_types, ./conversion_utils, ./group_manager, ./nonce_manager import logos_delivery/waku/waku_core -proc calcEpoch*(rlnPeer: Rln, t: float64): Epoch = +proc calcEpoch*(rln: Rln, t: float64): Epoch = ## gets time `t` as `flaot64` with subseconds resolution in the fractional part ## and returns its corresponding rln `Epoch` value - let e = uint64(t / rlnPeer.rlnEpochSizeSec.float64) + let e = uint64(t / rln.rlnEpochSizeSec.float64) return toEpoch(e) -proc nextEpoch*(rlnPeer: Rln, time: float64): float64 = +proc nextEpoch*(rln: Rln, time: float64): float64 = let - currentEpoch = uint64(time / rlnPeer.rlnEpochSizeSec.float64) - nextEpochTime = float64(currentEpoch + 1) * rlnPeer.rlnEpochSizeSec.float64 + currentEpoch = uint64(time / rln.rlnEpochSizeSec.float64) + nextEpochTime = float64(currentEpoch + 1) * rln.rlnEpochSizeSec.float64 currentTime = epochTime() # Ensure we always return a future time @@ -26,8 +26,8 @@ proc nextEpoch*(rlnPeer: Rln, time: float64): float64 = else: return epochTime() -proc getCurrentEpoch*(rlnPeer: Rln): Epoch = - return rlnPeer.calcEpoch(epochTime()) +proc getCurrentEpoch*(rln: Rln): Epoch = + return rln.calcEpoch(epochTime()) proc absDiff*(e1, e2: Epoch): uint64 = ## returns the absolute difference between the two rln `Epoch`s `e1` and `e2` @@ -55,11 +55,31 @@ proc toRLNSignal*(wakumessage: WakuMessage): seq[byte] = return output proc generateRLNProof*( - rlnPeer: Rln, input: seq[byte], senderEpochTime: float64 + rln: Rln, input: seq[byte], senderEpochTime: float64 ): Future[RlnResult[seq[byte]]] {.async.} = - let epoch = rlnPeer.calcEpoch(senderEpochTime) - let nonce = rlnPeer.nonceManager.getNonce().valueOr: + let epoch = rln.calcEpoch(senderEpochTime) + let nonce = rln.nonceManager.getNonce().valueOr: return err("could not get new message id to generate an rln proof: " & $error) - let proof = (await rlnPeer.groupManager.generateProof(input, epoch, nonce)).valueOr: + let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr: return err("could not generate rln-v2 proof: " & $error) return ok(proof.encode().buffer) + +proc checkAndGenerateRLNProof*( + rln: Option[Rln], message: WakuMessage +): Future[Result[WakuMessage, string]] {.async.} = + if message.proof.len > 0: + return ok(message) + + if rln.isNone(): + notice "Publishing message without RLN proof" + return ok(message) + + let + time = getTime().toUnix() + senderEpochTime = float64(time) + var msgWithProof = message + msgWithProof.proof = ( + await rln.get().generateRLNProof(msgWithProof.toRLNSignal(), senderEpochTime) + ).valueOr: + return err("error in checkAndGenerateRLNProof: " & $error) + return ok(msgWithProof) diff --git a/logos_delivery/waku/waku_core/message/message.nim b/logos_delivery/waku/waku_core/message/message.nim index acd7055a0..f6e6c9336 100644 --- a/logos_delivery/waku/waku_core/message/message.nim +++ b/logos_delivery/waku/waku_core/message/message.nim @@ -27,3 +27,8 @@ type WakuMessage* = object # Data payload transmitted. # The proof attribute indicates that the message is not spam. This # attribute will be used in the rln-relay protocol. proof*: seq[byte] + +proc ensureTimestampSet*(message: WakuMessage): WakuMessage = + result = message + if result.timestamp == 0: + result.timestamp = getNowInNanosecondTime() diff --git a/logos_delivery/waku/waku_lightpush/callbacks.nim b/logos_delivery/waku/waku_lightpush/callbacks.nim index 68af190f1..80d9c2439 100644 --- a/logos_delivery/waku/waku_lightpush/callbacks.nim +++ b/logos_delivery/waku/waku_lightpush/callbacks.nim @@ -7,47 +7,20 @@ import ../waku_core, ../waku_relay, ./common, ../rln, ../rln/protocol_types import std/times, libp2p/peerid, stew/byteutils -proc checkAndGenerateRLNProof*( - rlnPeer: Option[Rln], message: WakuMessage -): Future[Result[WakuMessage, string]] {.async.} = - # check if the message already has RLN proof - if message.proof.len > 0: - return ok(message) - - if rlnPeer.isNone(): - notice "Publishing message without RLN proof" - return ok(message) - # generate and append RLN proof - let - time = getTime().toUnix() - senderEpochTime = float64(time) - var msgWithProof = message - msgWithProof.proof = ( - await rlnPeer.get().generateRLNProof(msgWithProof.toRLNSignal, senderEpochTime) - ).valueOr: - return err($error) - return ok(msgWithProof) - proc getNilPushHandler*(): PushMessageHandler = return proc( pubsubTopic: string, message: WakuMessage ): Future[WakuLightPushResult] {.async.} = return lightpushResultInternalError("no waku relay found") -proc getRelayPushHandler*( - wakuRelay: WakuRelay, rlnPeer: Option[Rln] = none[Rln]() -): PushMessageHandler = +proc getRelayPushHandler*(wakuRelay: WakuRelay): PushMessageHandler = return proc( pubsubTopic: string, message: WakuMessage ): Future[WakuLightPushResult] {.async.} = - # append RLN proof - let msgWithProof = (await checkAndGenerateRLNProof(rlnPeer, message)).valueOr: - return lighpushErrorResult(LightPushErrorCode.OUT_OF_RLN_PROOF, error) - - (await wakuRelay.validateMessage(pubSubTopic, msgWithProof)).isOkOr: + (await wakuRelay.validateMessage(pubSubTopic, message)).isOkOr: return lighpushErrorResult(LightPushErrorCode.INVALID_MESSAGE, $error) - let publishedResult = (await wakuRelay.publish(pubsubTopic, msgWithProof)).valueOr: + let publishedResult = (await wakuRelay.publish(pubsubTopic, message)).valueOr: let msgHash = computeMessageHash(pubsubTopic, message).to0xHex() notice "Lightpush request has not been published to any peers", msg_hash = msgHash, reason = $error diff --git a/logos_delivery/waku/waku_lightpush/client.nim b/logos_delivery/waku/waku_lightpush/client.nim index 680970a51..9572b9323 100644 --- a/logos_delivery/waku/waku_lightpush/client.nim +++ b/logos_delivery/waku/waku_lightpush/client.nim @@ -24,10 +24,6 @@ type WakuLightPushClient* = ref object proc new*(T: type WakuLightPushClient, peerManager: PeerManager, rng: crypto.Rng): T = WakuLightPushClient(peerManager: peerManager, rng: rng) -proc ensureTimestampSet(message: var WakuMessage) = - if message.timestamp == 0: - message.timestamp = getNowInNanosecondTime() - ## Short log string for peer identifiers (overloads for convenience) func shortPeerId(peer: PeerId): string = shortLog(peer) @@ -82,8 +78,7 @@ proc publish*( wakuMessage: WakuMessage, dest: Connection | PeerId | RemotePeerInfo, ): Future[WakuLightPushResult] {.async, gcsafe.} = - var message = wakuMessage - ensureTimestampSet(message) + let message = ensureTimestampSet(wakuMessage) let msgHash = computeMessageHash(pubSubTopic.get(""), message).to0xHex() diff --git a/logos_delivery/waku/waku_lightpush_legacy/callbacks.nim b/logos_delivery/waku/waku_lightpush_legacy/callbacks.nim index 4fc92d516..fa6ec92d4 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/callbacks.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/callbacks.nim @@ -10,45 +10,19 @@ import import std/times, libp2p/peerid, stew/byteutils -proc checkAndGenerateRLNProof*( - rlnPeer: Option[Rln], message: WakuMessage -): Future[Result[WakuMessage, string]] {.async.} = - # check if the message already has RLN proof - if message.proof.len > 0: - return ok(message) - - if rlnPeer.isNone(): - notice "Publishing message without RLN proof" - return ok(message) - # generate and append RLN proof - let - time = getTime().toUnix() - senderEpochTime = float64(time) - var msgWithProof = message - msgWithProof.proof = ( - await rlnPeer.get().generateRLNProof(msgWithProof.toRLNSignal(), senderEpochTime) - ).valueOr: - return err($error) - return ok(msgWithProof) - proc getNilPushHandler*(): PushMessageHandler = return proc( pubsubTopic: string, message: WakuMessage ): Future[WakuLightPushResult[void]] {.async.} = return err("no waku relay found") -proc getRelayPushHandler*( - wakuRelay: WakuRelay, rlnPeer: Option[Rln] = none[Rln]() -): PushMessageHandler = +proc getRelayPushHandler*(wakuRelay: WakuRelay): PushMessageHandler = return proc( pubsubTopic: string, message: WakuMessage ): Future[WakuLightPushResult[void]] {.async.} = - # append RLN proof - let msgWithProof = ?(await checkAndGenerateRLNProof(rlnPeer, message)) + ?(await wakuRelay.validateMessage(pubSubTopic, message)) - ?(await wakuRelay.validateMessage(pubSubTopic, msgWithProof)) - - (await wakuRelay.publish(pubsubTopic, msgWithProof)).isOkOr: + (await wakuRelay.publish(pubsubTopic, message)).isOkOr: ## Agreed change expected to the lightpush protocol to better handle such case. https://github.com/waku-org/pm/issues/93 let msgHash = computeMessageHash(pubsubTopic, message).to0xHex() notice "Lightpush request has not been published to any peers", diff --git a/logos_delivery/waku/waku_lightpush_legacy/client.nim b/logos_delivery/waku/waku_lightpush_legacy/client.nim index 511c3f543..0ed5ca823 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/client.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/client.nim @@ -73,9 +73,7 @@ proc publish*( ): Future[WakuLightPushResult[string]] {.async, gcsafe.} = ## On success, returns the msg_hash of the published message - var message = wakuMessage - if message.timestamp == 0: - message.timestamp = getNowInNanosecondTime() + let message = ensureTimestampSet(wakuMessage) let msg_hash_hex_str = computeMessageHash(pubsubTopic, message).to0xHex() let pushRequest = PushRequest(pubSubTopic: pubSubTopic, message: message) diff --git a/tests/node/test_wakunode_legacy_lightpush.nim b/tests/node/test_wakunode_legacy_lightpush.nim index c2e901d41..9301e9e01 100644 --- a/tests/node/test_wakunode_legacy_lightpush.nim +++ b/tests/node/test_wakunode_legacy_lightpush.nim @@ -99,6 +99,7 @@ suite "RLN Proofs as a Lightpush Service": client {.threadvar.}: WakuNode anvilProc {.threadvar.}: Process manager {.threadvar.}: OnchainGroupManager + wakuRlnConfig {.threadvar.}: WakuRlnConfig serverRemotePeerInfo {.threadvar.}: RemotePeerInfo pubsubTopic {.threadvar.}: PubsubTopic @@ -118,7 +119,7 @@ suite "RLN Proofs as a Lightpush Service": # mount rln-relay # match prod epoch window to reduce test flake - let wakuRlnConfig = getWakuRlnConfig( + wakuRlnConfig = getWakuRlnConfig( manager = manager, index = MembershipIndex(1), epochSizeSec = 600 ) @@ -161,9 +162,15 @@ suite "RLN Proofs as a Lightpush Service": let lightpushClient = newTestWakuNode(generateSecp256k1Key()) lightpushClient.mountLegacyLightPushClient() + # Attach the RLN proof. In production the client mounts RLN and generates the + # proof in legacyLightpushPublish; here we generate it using the server's RLN + # instance since both ends share group state via the in-memory manager. + let msgWithProof = + (await checkAndGenerateRLNProof(some(server.rln), message)).get() + # When the client publishes a message let publishResponse = await lightpushClient.legacyLightpushPublish( - some(pubsubTopic), message, serverRemotePeerInfo + some(pubsubTopic), msgWithProof, serverRemotePeerInfo ) if not publishResponse.isOk(): diff --git a/tests/node/test_wakunode_lightpush.nim b/tests/node/test_wakunode_lightpush.nim index cc82cb7a3..6e87ab237 100644 --- a/tests/node/test_wakunode_lightpush.nim +++ b/tests/node/test_wakunode_lightpush.nim @@ -96,6 +96,7 @@ suite "RLN Proofs as a Lightpush Service": client {.threadvar.}: WakuNode anvilProc {.threadvar.}: Process manager {.threadvar.}: OnchainGroupManager + wakuRlnConfig {.threadvar.}: WakuRlnConfig serverRemotePeerInfo {.threadvar.}: RemotePeerInfo pubsubTopic {.threadvar.}: PubsubTopic @@ -115,7 +116,7 @@ suite "RLN Proofs as a Lightpush Service": # mount rln-relay # match prod epoch window to reduce test flake - let wakuRlnConfig = getWakuRlnConfig( + wakuRlnConfig = getWakuRlnConfig( manager = manager, index = MembershipIndex(1), epochSizeSec = 600 ) @@ -158,9 +159,15 @@ suite "RLN Proofs as a Lightpush Service": let lightpushClient = newTestWakuNode(generateSecp256k1Key()) lightpushClient.mountLightPushClient() + # Attach the RLN proof. In production the client mounts RLN and generates the + # proof in lightpushPublish; here we generate it using the server's RLN instance + # since both ends share group state via the in-memory manager. + let msgWithProof = + (await checkAndGenerateRLNProof(some(server.rln), message)).get() + # When the client publishes a message let publishResponse = await lightpushClient.lightpushPublish( - some(pubsubTopic), message, some(serverRemotePeerInfo) + some(pubsubTopic), msgWithProof, some(serverRemotePeerInfo) ) if not publishResponse.isOk():