From 7430c2aa176babf880d95b006ddcb27f9421d508 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:16:52 +0000 Subject: [PATCH] =?UTF-8?q?fix(quartz):=20audit=20fixes=20=E2=80=94=20Veri?= =?UTF-8?q?fyAuthOnlyPolicy=20+=20small=20wins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-audit of the event-ingestion-batching changes turned up one real bug + a handful of cleanups. Fix: AUTH events skipped signature verification when parallelVerify=true. Previous commit dropped VerifyPolicy from the policy chain to avoid double-verifying EVENTs (the IngestQueue does those off-thread). But VerifyPolicy.accept(AuthCmd) was the only thing checking AUTH signatures — FullAuthPolicy verifies challenge / relay / expiry but trusts the sig. Removing VerifyPolicy let a forged event mark a pubkey as authenticated. Split VerifyPolicy into a parameterised base class (VerifyEventsAndAuthPolicy) with two singletons: - VerifyPolicy: verifies both EVENT and AUTH (existing default). - VerifyAuthOnlyPolicy: verifies AUTH only — use when an IngestQueue does parallel EVENT verify, since AUTH commands bypass the queue entirely. Geode's composePolicy now selects VerifyAuthOnlyPolicy when parallelVerify is on, keeping AUTH signature checks intact while still letting EVENT verify run in parallel on the writer's CPU fan-out. Cleanups (no behavior change): - IngestQueue.processBatch was ~70 lines; split into verifyBatch / runInsertStage / dispatchOutcomes. - Single-event verify shortcut: skip the coroutineScope + async dance for batch-of-1 (the common low-load case) and call the hook directly. - Hoisted the "internal error: missing outcome" Rejected sentinel to a companion `missingOutcome` constant. - Imported ClosedSendChannelException / ClosedReceiveChannelException instead of using the fully-qualified form inline. - Dropped NostrServer's verifyEvent companion wrapper — direct lambda is just as cheap and the "single instance" comment was inaccurate. - ObservableEventStore.batchInsert now uses requireNoNulls() rather than @Suppress("UNCHECKED_CAST"), since every index is provably populated. --- .../kotlin/com/vitorpamplona/geode/Main.kt | 13 ++- .../nip01Core/relay/server/IngestQueue.kt | 94 ++++++++++++------- .../nip01Core/relay/server/NostrServer.kt | 10 +- .../nip01Core/relay/server/RelaySession.kt | 3 +- .../relay/server/policies/VerifyPolicy.kt | 32 ++++++- .../nip01Core/store/ObservableEventStore.kt | 7 +- 6 files changed, 103 insertions(+), 56 deletions(-) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 19648f07cc..a731890e88 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore @@ -104,10 +105,7 @@ fun main(args: Array) { val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl) val policyBuilder: () -> IRelayPolicy = { - // When parallel verify is enabled the IngestQueue runs - // Schnorr verify off the WS pump, so the policy chain skips - // VerifyPolicy to avoid double-verifying every event. - composePolicy(config, advertisedUrl, requireAuth, verifySigs && !parallelVerify) + composePolicy(config, advertisedUrl, requireAuth, verifySigs, parallelVerify) } val stateFile = config.admin.state_file?.let { File(it) } @@ -168,6 +166,7 @@ private fun composePolicy( advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl, requireAuth: Boolean, verifySigs: Boolean, + parallelVerify: Boolean, ): IRelayPolicy { val pieces = mutableListOf() @@ -188,7 +187,11 @@ private fun composePolicy( } if (verifySigs) { - pieces += VerifyPolicy + // When parallel verify is on, the IngestQueue handles EVENT + // verification on the writer's CPU fan-out — but AUTH events + // bypass the queue, so we still need the policy chain to + // verify those. `VerifyAuthOnlyPolicy` does exactly that. + pieces += if (parallelVerify) VerifyAuthOnlyPolicy else VerifyPolicy } return pieces.fold(EmptyPolicy) { acc, p -> diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt index b93b236ae9..5b4876693c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlin.coroutines.CoroutineContext @@ -168,43 +169,53 @@ class IngestQueue( processBatch(batch) batch.clear() } - } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { + } catch (_: ClosedReceiveChannelException) { // Normal shutdown via close(). } } private suspend fun processBatch(batch: List) { - // Tier 3: parallel pre-insert verify. Each batch entry that - // fails verification is pre-marked Rejected and excluded from - // the SQLite transaction. The remaining (verified) events go - // through batchInsert in original order; we re-stitch outcomes - // back to the full batch by index. - val verifyResults: BooleanArray? = - verify?.let { hook -> - coroutineScope { - batch - .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } - .awaitAll() - .toBooleanArray() - } - } + val verifyResults = verifyBatch(batch) + val finalOutcomes = runInsertStage(batch, verifyResults) + dispatchOutcomes(batch, finalOutcomes) + } - val toInsert: List - val insertIndices: IntArray - if (verifyResults == null) { - toInsert = batch.map { it.event } - insertIndices = IntArray(batch.size) { it } - } else { - val accepted = ArrayList(batch.size) - val mapping = ArrayList(batch.size) - for (i in batch.indices) { - if (verifyResults[i]) { - accepted.add(batch[i].event) - mapping.add(i) - } + /** + * Tier 3: per-row verify. Returns `null` when no [verify] hook is + * configured (skip the stage entirely). For multi-event batches + * each verify runs as its own `async(Default)` so they spread + * across CPU cores; single-event batches short-circuit to a + * direct call to avoid coroutine-scope overhead. + */ + private suspend fun verifyBatch(batch: List): BooleanArray? { + val hook = verify ?: return null + if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) } + return coroutineScope { + batch + .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } + .awaitAll() + .toBooleanArray() + } + } + + /** + * Run the SQLite transaction for the verified subset of [batch] + * and stitch outcomes back to a per-batch-index array. Failed + * verifies pre-mark `Rejected` and skip the insert. A whole-batch + * commit failure converts every persisted entry to `Rejected` + * with the throw message. + */ + private suspend fun runInsertStage( + batch: List, + verifyResults: BooleanArray?, + ): Array { + val toInsert = ArrayList(batch.size) + val insertIndices = ArrayList(batch.size) + for (i in batch.indices) { + if (verifyResults == null || verifyResults[i]) { + toInsert.add(batch[i].event) + insertIndices.add(i) } - toInsert = accepted - insertIndices = mapping.toIntArray() } val insertOutcomes: List = @@ -220,11 +231,10 @@ class IngestQueue( } } - // Build a per-batch-index outcome array. val finalOutcomes = arrayOfNulls(batch.size) for (j in insertIndices.indices) { - finalOutcomes[insertIndices[j]] = insertOutcomes.getOrNull(j) - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + finalOutcomes[insertIndices[j]] = + insertOutcomes.getOrNull(j) ?: missingOutcome } if (verifyResults != null) { for (i in batch.indices) { @@ -233,12 +243,16 @@ class IngestQueue( } } } + return finalOutcomes + } + private fun dispatchOutcomes( + batch: List, + outcomes: Array, + ) { for (i in batch.indices) { val sub = batch[i] - val outcome = - finalOutcomes[i] - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + val outcome = outcomes[i] ?: missingOutcome try { sub.onComplete(outcome) } catch (e: Throwable) { @@ -262,6 +276,14 @@ class IngestQueue( } companion object { + /** + * Default for a missing per-row outcome — only reachable on a + * contract violation (the store returned fewer outcomes than + * inserts), so the message is informational, not user-facing. + */ + private val missingOutcome = + IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + /** * Cap per batch. Sized to keep per-batch latency low (each * transaction holds the SQLite writer mutex; over-large diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 54c23fbe00..0ec8e602be 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -20,7 +20,6 @@ */ package com.vitorpamplona.quartz.nip01Core.relay.server -import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore @@ -65,7 +64,7 @@ class NostrServer( IngestQueue( store = store, parentContext = parentContext, - verify = if (parallelVerify) ::verifyEvent else null, + verify = if (parallelVerify) ({ it.verify() }) else null, ) private val subStore = LiveEventStore(store, ingest) @@ -128,11 +127,4 @@ class NostrServer( */ @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()")) fun shutdown() = close() - - private companion object { - // Function reference (`Event::verify`) wrapper so the - // ingest hook keeps a single instance per server rather - // than allocating a fresh lambda on every call site. - private fun verifyEvent(event: Event): Boolean = event.verify() - } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 5c2c56f33b..c1698492a3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -43,6 +43,7 @@ import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.launch /** @@ -147,7 +148,7 @@ class RelaySession( } } } - } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) { + } catch (_: ClosedSendChannelException) { // Server is shutting down — the queue is closed. Reply // with a transient failure so a client re-trying against // the next instance gets a sane signal; the WS itself is diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt index 283fda6707..e59bd53088 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt @@ -31,13 +31,24 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.PolicyResult /** - * Allows all commands without authentication. This is the default policy. + * Verifies the Schnorr signature + id hash of every incoming + * `EVENT` and `AUTH` event. Other commands pass through. + * + * The default `VerifyPolicy` singleton verifies both. The + * [VerifyAuthOnlyPolicy] singleton skips the EVENT path — use it + * when the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * is doing parallel verify (Tier 3 of the event-ingestion plan) + * so EVENTs aren't verified twice. AUTH is still verified inline + * because the AUTH path bypasses the queue entirely; without it, + * a forged event could mark a pubkey as authenticated. */ -object VerifyPolicy : IRelayPolicy { +open class VerifyEventsAndAuthPolicy( + private val verifyEvents: Boolean, +) : IRelayPolicy { override fun onConnect(send: (Message) -> Unit) { } override fun accept(cmd: EventCmd) = - if (cmd.event.verify()) { + if (!verifyEvents || cmd.event.verify()) { PolicyResult.Accepted(cmd) } else { PolicyResult.Rejected("invalid: bad signature or id") @@ -56,3 +67,18 @@ object VerifyPolicy : IRelayPolicy { override fun canSendToSession(event: Event) = true } + +/** + * Default verify policy — checks every EVENT and every AUTH. Use + * this when nothing else in the stack verifies signatures. + */ +object VerifyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = true) + +/** + * Verify policy that skips the EVENT path. Use when the + * `IngestQueue` is configured with `parallelVerify`, so EVENTs are + * verified once on the writer's CPU fan-out instead of inline on + * the WebSocket pump. AUTH is still verified inline because that + * command never reaches the queue. + */ +object VerifyAuthOnlyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = false) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index eedca33713..cb18c13a96 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -137,8 +137,11 @@ class ObservableEventStore( } } - @Suppress("UNCHECKED_CAST") - return outcomes.toList() as List + // Every index in `events.indices` was populated above (either + // from the ephemeral pre-pass or the `inner.batchInsert` + // result), so no nulls remain. `requireNoNulls()` enforces + // that with a runtime check instead of an unchecked cast. + return outcomes.requireNoNulls().asList() } override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) {