From 3accb291c9c0857fc0f5be5e970ed9a260260149 Mon Sep 17 00:00:00 2001 From: darshankabariya Date: Fri, 24 Jul 2026 01:42:26 +0530 Subject: [PATCH] feat: persist RLN proofs in the archive so synced history stays verifiable Live messages carry an RLN proof, but archives dropped it at insert time - so history handed over via store-sync transfer or store resume arrived proofless and had to be accepted on faith, sidestepping the spam limiter exactly where it should be load-bearing. A malicious peer could invent 'history you missed' and a victim would swallow and re-serve it. - archive: proof column (sqlite schema v11, postgres v8; nullable, so existing rows migrate as a no-op and in-memory archives need no migration); insert/select thread message.proof through both drivers; the queue driver stores whole messages and needed no change - transfer and store resume serve proof-bearing messages automatically (the wire codecs already carry the field) and both re-verify the original author's proof on arrival: the receiving node's FIRST verification of a message it missed on relay, not a repeat - same work the relay path would have done, deferred to catch-up - store resume gains the same validator the sync transfer uses, since a store node could also serve invented history - new --store-sync-require-proof flag (default off): the rollout lever; once the fleet persists and serves proofs, flipping it makes nodes reject proofless synced history when RLN is enabled - reconciliation untouched: proofs are not part of the message hash, so identity, fingerprints and cursors are unchanged Tested end-to-end: a proof survives full-node -> full-node sync transfer and the store-query -> resume -> archive trip intact. Co-Authored-By: Claude Fable 5 --- .../conf_builder/store_sync_conf_builder.nim | 5 ++++ logos_delivery/waku/factory/node_factory.nim | 5 +++- logos_delivery/waku/factory/waku_conf.nim | 3 ++ logos_delivery/waku/node/waku_node.nim | 17 +++++++---- logos_delivery/waku/node/waku_node/store.nim | 18 ++++++++++- .../driver/postgres_driver/migrations.nim | 2 +- .../postgres_driver/postgres_driver.nim | 29 ++++++++++++++---- .../driver/sqlite_driver/migrations.nim | 2 +- .../driver/sqlite_driver/queries.nim | 20 ++++++++----- .../driver/sqlite_driver/sqlite_driver.nim | 1 + logos_delivery/waku/waku_store/resume.nim | 23 +++++++++++++- .../message_store/00011_addProofColumn.up.sql | 4 +++ .../content_script_version_8.nim | 12 ++++++++ .../pg_migration_manager.nim | 3 +- tests/waku_archive/test_driver_sqlite.nim | 30 +++++++++++++++++++ tests/waku_store/test_resume.nim | 14 ++++++++- tests/waku_store_sync/test_full_node_sync.nim | 23 ++++++++++++++ tools/confutils/cli_args.nim | 8 +++++ 18 files changed, 193 insertions(+), 26 deletions(-) create mode 100644 migrations/message_store/00011_addProofColumn.up.sql create mode 100644 migrations/message_store_postgres/content_script_version_8.nim diff --git a/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim b/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim index 607fd9e21..cc19de992 100644 --- a/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim +++ b/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim @@ -18,6 +18,7 @@ type StoreSyncConfBuilder* = object intervalSec*: Opt[uint32] relayJitterSec*: Opt[uint32] dbUrl*: Opt[string] + requireProof*: Opt[bool] proc init*(T: type StoreSyncConfBuilder): StoreSyncConfBuilder = StoreSyncConfBuilder() @@ -37,6 +38,9 @@ proc withRelayJitterSec*(b: var StoreSyncConfBuilder, relayJitterSec: uint32) = proc withDbUrl*(b: var StoreSyncConfBuilder, dbUrl: string) = b.dbUrl = Opt.some(dbUrl) +proc withRequireProof*(b: var StoreSyncConfBuilder, requireProof: bool) = + b.requireProof = Opt.some(requireProof) + proc build*(b: StoreSyncConfBuilder): Result[Opt[StoreSyncConf], string] = if not b.enabled.get(DefaultStoreSyncEnabled): return ok(Opt.none(StoreSyncConf)) @@ -64,6 +68,7 @@ proc build*(b: StoreSyncConfBuilder): Result[Opt[StoreSyncConf], string] = intervalSec: b.intervalSec.get(), relayJitterSec: b.relayJitterSec.get(), dbUrl: dbUrl, + requireProof: b.requireProof.get(false), ) ) ) diff --git a/logos_delivery/waku/factory/node_factory.nim b/logos_delivery/waku/factory/node_factory.nim index 6eb6ad011..b77ffbc15 100644 --- a/logos_delivery/waku/factory/node_factory.nim +++ b/logos_delivery/waku/factory/node_factory.nim @@ -260,6 +260,7 @@ proc setupProtocols( await node.mountStoreSync( conf.clusterId, conf.subscribeShards, conf.contentTopics, confStoreSync.rangeSec, confStoreSync.intervalSec, confStoreSync.relayJitterSec, + confStoreSync.requireProof, ) ).isOkOr: return err("failed to mount waku store sync protocol: " & $error) @@ -283,7 +284,9 @@ proc setupProtocols( conf.storeSyncConf.isSome(): # sync-enabled full nodes track last-online so the startup catch-up can # choose between neighbour reconciliation and store resume - node.setupStoreResume() + let resumeRequireProof = + conf.storeSyncConf.isSome() and conf.storeSyncConf.get().requireProof + node.setupStoreResume(requireProof = resumeRequireProof) if conf.shardingConf.kind == AutoSharding: node.mountAutoSharding(conf.clusterId, conf.shardingConf.numShardsInCluster).isOkOr: diff --git a/logos_delivery/waku/factory/waku_conf.nim b/logos_delivery/waku/factory/waku_conf.nim index 7374f12e9..4a06302cb 100644 --- a/logos_delivery/waku/factory/waku_conf.nim +++ b/logos_delivery/waku/factory/waku_conf.nim @@ -58,6 +58,9 @@ type StoreSyncConf* {.requiresInit.} = object dbUrl*: string ## Backs reconciliation & transfer with a small bounded archive ## when the store service is not enabled. Ignored otherwise. + requireProof*: bool + ## Reject synced messages that carry no RLN proof. Only effective + ## when RLN is enabled; keep false until the fleet persists proofs. type MixConf* = ref object mixKey*: Curve25519Key diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index 21de0f360..d6e84c0e9 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -371,6 +371,7 @@ proc mountStoreSync*( storeSyncRange: uint32, storeSyncInterval: uint32, storeSyncRelayJitter: uint32, + storeSyncRequireProof: bool = false, ): Future[Result[void, string]] {.async.} = if node.wakuArchive.isNil(): return err("store sync requires a mounted archive") @@ -412,16 +413,20 @@ proc mountStoreSync*( ## 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 gaps: archives do - ## not persist RLN proofs, so archive-served messages arrive proofless - ## and can only be counted, not verified (enforcement needs proof - ## persistence); and proofs built against roots older than the - ## acceptable root window are rejected, bounding history sync across - ## heavy membership churn. + ## checked: synced messages are old by design. Archives persist the + ## original author's proof, so transferred history is re-verified as + ## it arrives — the node's first verification of a message it missed + ## on relay. Proofless messages are accepted (and counted) only while + ## requireProof is off: the rollout lever until the whole fleet + ## persists and serves proofs. Proofs built against roots older than + ## the acceptable root window are rejected, bounding history sync + ## across heavy membership churn. if node.rln.isNil(): return true if msg.proof.len == 0: + if storeSyncRequireProof: + return false total_transfer_messages_unverified.inc() return true diff --git a/logos_delivery/waku/node/waku_node/store.nim b/logos_delivery/waku/node/waku_node/store.nim index 73dd1c634..b2d6fbba8 100644 --- a/logos_delivery/waku/node/waku_node/store.nim +++ b/logos_delivery/waku/node/waku_node/store.nim @@ -24,6 +24,7 @@ import ../../waku_store/common as store_common, ../../waku_store/resume, ../../waku_store_sync/reconciliation, + ../../rln, ../peer_manager, ../../common/rate_limit/setting, ../../waku_archive @@ -151,7 +152,7 @@ proc query*( return ok(response) -proc setupStoreResume*(node: WakuNode) = +proc setupStoreResume*(node: WakuNode, requireProof: bool = false) = # Resume-fetched messages also feed reconciliation, so the next sync # round does not re-request history the store already provided. The # reconciliation protocol may mount after resume is set up; check at @@ -162,11 +163,26 @@ proc setupStoreResume*(node: WakuNode) = if not node.wakuStoreReconciliation.isNil(): node.wakuStoreReconciliation.messageIngress(msgHash, pubsubTopic, msg) + # Same trust rule as the sync transfer: re-verify the original author's + # RLN proof on resume-fetched history (a store node could also serve + # invented history). RLN may mount after resume is set up; check at + # call time. nil rln means RLN is not configured on this network. + let msgValidator: ResumeValidator = proc(msg: WakuMessage): Future[bool] {.async.} = + if node.rln.isNil(): + return true + + if msg.proof.len == 0: + return not requireProof + + let res = await node.rln.validateMessage(msg, checkFreshness = false) + return res == MessageValidationResult.Valid + node.wakuStoreResume = StoreResume.new( node.peerManager, node.wakuArchive, node.wakuStoreClient, reconIngress = Opt.some(reconIngress), + msgValidator = Opt.some(msgValidator), ).valueOr: error "Failed to setup Store Resume", error = $error return diff --git a/logos_delivery/waku/waku_archive/driver/postgres_driver/migrations.nim b/logos_delivery/waku/waku_archive/driver/postgres_driver/migrations.nim index a1a3e93fe..ec2f3d1ff 100644 --- a/logos_delivery/waku/waku_archive/driver/postgres_driver/migrations.nim +++ b/logos_delivery/waku/waku_archive/driver/postgres_driver/migrations.nim @@ -9,7 +9,7 @@ import logScope: topics = "waku archive migration" -const SchemaVersion* = 7 # increase this when there is an update in the database schema +const SchemaVersion* = 8 # increase this when there is an update in the database schema proc breakIntoStatements*(script: string): seq[string] = ## Given a full migration script, that can potentially contain a list diff --git a/logos_delivery/waku/waku_archive/driver/postgres_driver/postgres_driver.nim b/logos_delivery/waku/waku_archive/driver/postgres_driver/postgres_driver.nim index 43a36821a..f9ef926e2 100644 --- a/logos_delivery/waku/waku_archive/driver/postgres_driver/postgres_driver.nim +++ b/logos_delivery/waku/waku_archive/driver/postgres_driver/postgres_driver.nim @@ -40,14 +40,14 @@ type PostgresDriver* = ref object of ArchiveDriver const InsertRowStmtName = "InsertRow" const InsertRowStmtDefinition = """INSERT INTO messages (id, messageHash, pubsubTopic, contentTopic, payload, - version, timestamp, meta) VALUES ($1, $2, $3, $4, $5, $6, $7, CASE WHEN $8 = '' THEN NULL ELSE $8 END) ON CONFLICT DO NOTHING;""" + version, timestamp, meta, proof) VALUES ($1, $2, $3, $4, $5, $6, $7, CASE WHEN $8 = '' THEN NULL ELSE $8 END, CASE WHEN $9 = '' THEN NULL ELSE $9 END) ON CONFLICT DO NOTHING;""" const InsertRowInMessagesLookupStmtName = "InsertRowMessagesLookup" const InsertRowInMessagesLookupStmtDefinition = """INSERT INTO messages_lookup (messageHash, timestamp) VALUES ($1, $2) ON CONFLICT DO NOTHING;""" const SelectClause = - """SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta FROM messages """ + """SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta, proof FROM messages """ const SelectNoCursorAscStmtName = "SelectWithoutCursorAsc" const SelectNoCursorAscStmtDef = @@ -245,6 +245,7 @@ proc rowCallbackImpl( rawVersion: string rawTimestamp: string rawMeta: string + rawProof: string hashHex: string msgHash: WakuMessageHash @@ -256,6 +257,7 @@ proc rowCallbackImpl( version: uint timestamp: Timestamp meta: string + proof: string wakuMessage: WakuMessage rawHash = $(pqgetvalue(pqResult, iRow, 0)) @@ -265,6 +267,7 @@ proc rowCallbackImpl( rawVersion = $(pqgetvalue(pqResult, iRow, 4)) rawTimestamp = $(pqgetvalue(pqResult, iRow, 5)) rawMeta = $(pqgetvalue(pqResult, iRow, 6)) + rawProof = $(pqgetvalue(pqResult, iRow, 7)) trace "db output", rawHash, pubSubTopic, contentTopic, rawPayload, rawVersion, rawTimestamp, rawMeta @@ -275,6 +278,7 @@ proc rowCallbackImpl( version = parseUInt(rawVersion) timestamp = parseInt(rawTimestamp) meta = parseHexStr(rawMeta) + proof = parseHexStr(rawProof) except ValueError: error "could not parse correctly", error = getCurrentExceptionMsg() @@ -285,6 +289,7 @@ proc rowCallbackImpl( wakuMessage.version = uint32(version) wakuMessage.timestamp = timestamp wakuMessage.meta = @(meta.toOpenArrayByte(0, meta.high)) + wakuMessage.proof = @(proof.toOpenArrayByte(0, proof.high)) outRows.add((msgHash, pubSubTopic, wakuMessage)) @@ -301,6 +306,7 @@ method put*( let version = $message.version let timestamp = $message.timestamp let meta = byteutils.toHex(message.meta) + let proof = byteutils.toHex(message.proof) trace "put PostgresDriver", messageHash, contentTopic, payload, version, timestamp, meta @@ -316,7 +322,7 @@ method put*( InsertRowStmtDefinition, @[ fakeId, messageHash, pubsubTopic, contentTopic, payload, version, timestamp, - meta, + meta, proof, ], @[ int32(fakeId.len), @@ -327,8 +333,19 @@ method put*( int32(version.len), int32(timestamp.len), int32(meta.len), + int32(proof.len), + ], + @[ + int32(0), + int32(0), + int32(0), + int32(0), + int32(0), + int32(0), + int32(0), + int32(0), + int32(0), ], - @[int32(0), int32(0), int32(0), int32(0), int32(0), int32(0), int32(0), int32(0)], ) ).isOkOr: return err("could not put msg in messages table: " & $error) @@ -356,7 +373,7 @@ method getAllMessages*( ( await s.readConnPool.pgQuery( - """SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta + """SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta, proof FROM messages ORDER BY timestamp ASC, messageHash ASC""", newSeq[string](0), @@ -823,7 +840,7 @@ proc getMessagesByMessageHashes( {hashes} ) ) - SELECT m.messageHash, pubsubTopic, contentTopic, payload, version, m.timestamp, meta + SELECT m.messageHash, pubsubTopic, contentTopic, payload, version, m.timestamp, meta, proof FROM messages m INNER JOIN messages_lookup l diff --git a/logos_delivery/waku/waku_archive/driver/sqlite_driver/migrations.nim b/logos_delivery/waku/waku_archive/driver/sqlite_driver/migrations.nim index b077de19a..864e5295d 100644 --- a/logos_delivery/waku/waku_archive/driver/sqlite_driver/migrations.nim +++ b/logos_delivery/waku/waku_archive/driver/sqlite_driver/migrations.nim @@ -7,7 +7,7 @@ import ../../../common/databases/db_sqlite, ../../../common/databases/common logScope: topics = "waku archive migration" -const SchemaVersion* = 10 # increase this when there is an update in the database schema +const SchemaVersion* = 11 # increase this when there is an update in the database schema template projectRoot(): string = currentSourcePath.rsplit(DirSep, 1)[0] / ".." / ".." / ".." / ".." diff --git a/logos_delivery/waku/waku_archive/driver/sqlite_driver/queries.nim b/logos_delivery/waku/waku_archive/driver/sqlite_driver/queries.nim index c7e0037cc..edac6c249 100644 --- a/logos_delivery/waku/waku_archive/driver/sqlite_driver/queries.nim +++ b/logos_delivery/waku/waku_archive/driver/sqlite_driver/queries.nim @@ -15,7 +15,7 @@ type SqlQueryStr = string proc queryRowWakuMessageCallback( s: ptr sqlite3_stmt, - contentTopicCol, payloadCol, versionCol, timestampCol, metaCol: cint, + contentTopicCol, payloadCol, versionCol, timestampCol, metaCol, proofCol: cint, ): WakuMessage = let topic = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, contentTopicCol)) @@ -24,13 +24,16 @@ proc queryRowWakuMessageCallback( p = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, payloadCol)) m = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, metaCol)) + pr = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, proofCol)) payloadLength = sqlite3_column_bytes(s, payloadCol) metaLength = sqlite3_column_bytes(s, metaCol) + proofLength = sqlite3_column_bytes(s, proofCol) payload = @(toOpenArray(p, 0, payloadLength - 1)) version = sqlite3_column_int64(s, versionCol) timestamp = sqlite3_column_int64(s, timestampCol) meta = @(toOpenArray(m, 0, metaLength - 1)) + proof = @(toOpenArray(pr, 0, proofLength - 1)) return WakuMessage( contentTopic: ContentTopic(contentTopic), @@ -38,6 +41,7 @@ proc queryRowWakuMessageCallback( version: uint32(version), timestamp: Timestamp(timestamp), meta: meta, + proof: proof, ) proc queryRowTimestampCallback(s: ptr sqlite3_stmt, timestampCol: cint): Timestamp = @@ -74,7 +78,7 @@ proc createTableQuery(table: string): SqlQueryStr = "CREATE TABLE IF NOT EXISTS " & table & " (" & " messageHash BLOB NOT NULL PRIMARY KEY," & " pubsubTopic BLOB NOT NULL," & " contentTopic BLOB NOT NULL," & " payload BLOB," & " version INTEGER NOT NULL," & - " timestamp INTEGER NOT NULL," & " meta BLOB" & ") WITHOUT ROWID;" + " timestamp INTEGER NOT NULL," & " meta BLOB," & " proof BLOB" & ") WITHOUT ROWID;" proc createTable*(db: SqliteDatabase): DatabaseResult[void] = let query = createTableQuery(DbTable) @@ -101,13 +105,13 @@ proc createOldestMessageTimestampIndex*(db: SqliteDatabase): DatabaseResult[void ## Insert message type InsertMessageParams* = - (seq[byte], seq[byte], seq[byte], seq[byte], int64, Timestamp, seq[byte]) + (seq[byte], seq[byte], seq[byte], seq[byte], int64, Timestamp, seq[byte], seq[byte]) proc insertMessageQuery(table: string): SqlQueryStr = return "INSERT INTO " & table & - "(messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta)" & - " VALUES (?, ?, ?, ?, ?, ?, ?);" + "(messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta, proof)" & + " VALUES (?, ?, ?, ?, ?, ?, ?, ?);" proc prepareInsertMessageStmt*( db: SqliteDatabase @@ -204,7 +208,7 @@ proc deleteOldestMessagesNotWithinLimit*( proc selectAllMessagesQuery(table: string): SqlQueryStr = return - "SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta" & + "SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta, proof" & " FROM " & table & " ORDER BY timestamp ASC" proc selectAllMessages*( @@ -223,6 +227,7 @@ proc selectAllMessages*( versionCol = 4, timestampCol = 5, metaCol = 6, + proofCol = 7, ) rows.add((hash, pubsubTopic, wakuMessage)) @@ -445,7 +450,7 @@ proc selectMessagesWithLimitQuery( var query: string query = - "SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta" + "SELECT messageHash, pubsubTopic, contentTopic, payload, version, timestamp, meta, proof" query &= " FROM " & table if where.isSome(): @@ -509,6 +514,7 @@ proc selectMessagesByStoreQueryWithLimit*( versionCol = 4, timestampCol = 5, metaCol = 6, + proofCol = 7, ) rows.add((hash, pubsubTopic, message)) diff --git a/logos_delivery/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim b/logos_delivery/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim index e12fb12bc..c05440891 100644 --- a/logos_delivery/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim +++ b/logos_delivery/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim @@ -57,6 +57,7 @@ method put*( int64(message.version), message.timestamp, message.meta, + message.proof, ) ) diff --git a/logos_delivery/waku/waku_store/resume.nim b/logos_delivery/waku/waku_store/resume.nim index 06963bc3c..a9f381400 100644 --- a/logos_delivery/waku/waku_store/resume.nim +++ b/logos_delivery/waku/waku_store/resume.nim @@ -39,6 +39,10 @@ type ## Feeds a resume-fetched message into the reconciliation storage so ## peers are not asked again for messages the store already provided. + ResumeValidator* = proc(msg: WakuMessage): Future[bool] {.gcsafe, raises: [].} + ## Returns false when a resume-fetched message must be dropped + ## (e.g. failed RLN proof verification). + StoreResume* = ref object handle: Future[void] @@ -73,6 +77,7 @@ proc initTransferHandler( wakuArchive: WakuArchive, wakuStoreClient: WakuStoreClient, reconIngress: Opt[ReconciliationIngress], + msgValidator: Opt[ResumeValidator], ) = # guard clauses to prevent faulty callback if self.peerManager.isNil(): @@ -117,6 +122,21 @@ proc initTransferHandler( msg = kv.message.get() msgHash = computeMessageHash(pubsubTopic, msg) + # Same trust rule as the sync transfer: resume-fetched history + # is validated (RLN proof of the original author) before it is + # archived, since a store node could also serve invented history. + if msgValidator.isSome(): + let validRes = catch: + await msgValidator.get()(msg) + + let valid = validRes.valueOr: + error "resume message validation error", error = error.msg + continue + + if not valid: + warn "resume message failed validation, dropping", msg_hash = msgHash + continue + # Catch-up messages are older than the archive's live-traffic # freshness window by definition, so they must enter through the # sync ingress, which skips that validation (same path the @@ -147,6 +167,7 @@ proc new*( wakuArchive: WakuArchive, wakuStoreClient: WakuStoreClient, reconIngress: Opt[ReconciliationIngress] = Opt.none(ReconciliationIngress), + msgValidator: Opt[ResumeValidator] = Opt.none(ResumeValidator), ): Result[T, string] = info "initializing store resume" @@ -159,7 +180,7 @@ proc new*( let resume = StoreResume(db: db, replaceStmt: replaceStmt, peerManager: peerManager) - resume.initTransferHandler(wakuArchive, wakuStoreClient, reconIngress) + resume.initTransferHandler(wakuArchive, wakuStoreClient, reconIngress, msgValidator) return ok(resume) diff --git a/migrations/message_store/00011_addProofColumn.up.sql b/migrations/message_store/00011_addProofColumn.up.sql new file mode 100644 index 000000000..e8a9fe92b --- /dev/null +++ b/migrations/message_store/00011_addProofColumn.up.sql @@ -0,0 +1,4 @@ +-- Persist the RLN proof alongside each stored message so that history served +-- via store-sync transfer stays verifiable: the receiving node can re-verify +-- the original author was rate-limited, instead of counting proofless messages. +ALTER TABLE Message ADD COLUMN proof BLOB; diff --git a/migrations/message_store_postgres/content_script_version_8.nim b/migrations/message_store_postgres/content_script_version_8.nim new file mode 100644 index 000000000..90f40e16f --- /dev/null +++ b/migrations/message_store_postgres/content_script_version_8.nim @@ -0,0 +1,12 @@ +const ContentScriptVersion_8* = """ + +-- Persist the original RLN proof alongside each stored message so that +-- history served via store-sync transfer or store queries stays verifiable: +-- the receiving node can re-verify the original author was rate-limited, +-- instead of accepting proofless messages. +ALTER TABLE messages ADD COLUMN IF NOT EXISTS proof VARCHAR; + +-- Update to new version +UPDATE version SET version = 8 WHERE version = 7; + +""" diff --git a/migrations/message_store_postgres/pg_migration_manager.nim b/migrations/message_store_postgres/pg_migration_manager.nim index 89a443609..79af6906f 100644 --- a/migrations/message_store_postgres/pg_migration_manager.nim +++ b/migrations/message_store_postgres/pg_migration_manager.nim @@ -1,7 +1,7 @@ import content_script_version_1, content_script_version_2, content_script_version_3, content_script_version_4, content_script_version_5, content_script_version_6, - content_script_version_7 + content_script_version_7, content_script_version_8 type MigrationScript* = object version*: int @@ -18,6 +18,7 @@ const PgMigrationScripts* = @[ MigrationScript(version: 5, scriptContent: ContentScriptVersion_5), MigrationScript(version: 6, scriptContent: ContentScriptVersion_6), MigrationScript(version: 7, scriptContent: ContentScriptVersion_7), + MigrationScript(version: 8, scriptContent: ContentScriptVersion_8), ] proc getMigrationScripts*(currentVersion: int64, targetVersion: int64): seq[string] = diff --git a/tests/waku_archive/test_driver_sqlite.nim b/tests/waku_archive/test_driver_sqlite.nim index 1eb30cb44..bd24ec208 100644 --- a/tests/waku_archive/test_driver_sqlite.nim +++ b/tests/waku_archive/test_driver_sqlite.nim @@ -52,3 +52,33 @@ suite "SQLite driver": ## Cleanup (waitFor driver.close()).expect("driver to close") + + test "insert a message with a proof and read it back": + ## The original author's RLN proof must survive archiving so that + ## history served via store-sync transfer stays verifiable. + let driver = newSqliteArchiveDriver() + + let proof = @[byte 1, 2, 3, 4] + let msg = fakeWakuMessage(proof = proof) + let msgHash = computeMessageHash(DefaultPubsubTopic, msg) + + ## When + let putRes = waitFor driver.put(msgHash, DefaultPubsubTopic, msg) + + ## Then + check: + putRes.isOk() + + let allRows = (waitFor driver.getAllMessages()).tryGet() + check: + allRows.len == 1 + allRows[0][2].proof == proof + + let queried = + (waitFor driver.getMessages(includeData = true, hashes = @[msgHash])).tryGet() + check: + queried.len == 1 + queried[0][2].proof == proof + + ## Cleanup + (waitFor driver.close()).expect("driver to close") diff --git a/tests/waku_store/test_resume.nim b/tests/waku_store/test_resume.nim index c0712a6dc..1a5cdd882 100644 --- a/tests/waku_store/test_resume.nim +++ b/tests/waku_store/test_resume.nim @@ -7,6 +7,7 @@ import node/peer_manager, waku_node, waku_core, + waku_core/message/digest, waku_store/resume, waku_store/common, waku_archive/driver, @@ -125,7 +126,7 @@ suite "Store Resume - End to End": ## would be silently dropped. let hourAgo = Timestamp(getNowInNanosecondTime() - 3_600_000_000_000) let oldMessages = @[ - fakeWakuMessage(@[byte 10], ts = hourAgo), + fakeWakuMessage(@[byte 10], ts = hourAgo, proof = @[byte 9, 9, 9]), fakeWakuMessage(@[byte 11], ts = hourAgo + 1), fakeWakuMessage(@[byte 12], ts = hourAgo + 2), fakeWakuMessage(@[byte 13], ts = hourAgo + 3), @@ -148,3 +149,14 @@ suite "Store Resume - End to End": check: countRes.get() == 15 + + # the original author's proof survives the store -> resume -> archive trip + let proofHash = computeMessageHash(DefaultPubsubTopic, oldMessages[0]) + let rows = ( + await clientDriver.getMessages(includeData = true, hashes = @[proofHash]) + ).valueOr: + raiseAssert $error + + check: + rows.len == 1 + rows[0][2].proof == @[byte 9, 9, 9] diff --git a/tests/waku_store_sync/test_full_node_sync.nim b/tests/waku_store_sync/test_full_node_sync.nim index 6f1377406..d72173556 100644 --- a/tests/waku_store_sync/test_full_node_sync.nim +++ b/tests/waku_store_sync/test_full_node_sync.nim @@ -157,6 +157,29 @@ suite "Waku Sync: full node miss recovery": await nodeA.hasMessage(hashB) await nodeB.hasMessage(hashA) + asyncTest "synced messages keep the original author's proof": + let proof = @[byte 4, 2, 4, 2] + let msg = fakeWakuMessage(contentTopic = DefaultContentTopic, proof = proof) + 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) + + var query = ArchiveQuery() + query.includeData = true + query.hashes = @[hash] + + let response = (await nodeB.archive.findMessages(query)).valueOr: + raiseAssert $error + + check: + response.messages.len == 1 + response.messages[0].proof == proof + asyncTest "messages failing transfer validation are not archived": await nodeB.stop() diff --git a/tools/confutils/cli_args.nim b/tools/confutils/cli_args.nim index ee7f513a2..9e7179644 100644 --- a/tools/confutils/cli_args.nim +++ b/tools/confutils/cli_args.nim @@ -451,6 +451,13 @@ hence would have reachability issues.""", name: "store-sync-db-url" .}: string + storeSyncRequireProof* {. + desc: + "Reject sync-transferred messages that carry no RLN proof. Only effective when RLN is enabled; keep false until every peer persists and serves proofs.", + defaultValue: false, + name: "store-sync-require-proof" + .}: bool + ## Filter config filter* {. desc: "Enable filter protocol: true|false", defaultValue: true, name: "filter" @@ -1138,6 +1145,7 @@ proc toWakuConf*(n: WakuNodeConf): ConfResult[WakuConf] = b.storeSyncConf.withRangeSec(n.storeSyncRange) b.storeSyncConf.withRelayJitterSec(n.storeSyncRelayJitter) b.storeSyncConf.withDbUrl(n.storeSyncDbUrl) + b.storeSyncConf.withRequireProof(n.storeSyncRequireProof) if n.mix.isSome(): b.mixConf.withEnabled(n.mix.get())