From a2b38cd1a3ef5c4ac1f186641288be4959071fb7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Jul 2026 21:54:49 +0000 Subject: [PATCH] feat(geode): per-upstream [[mirror]] filters, strfry-router parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each [[mirror]] entry takes an optional NIP-01 filter as a JSON object string, like strfry-router's per-stream filter. It is applied twice, matching strfry's design (cmd_router.cpp): - it shapes the REQ sent upstream (the mirror still owns since via backfill_seconds and strips limit — the subscription is unbounded); - every delivered event is re-checked against it before ingest, so an upstream answering outside its REQ — including a trusted one whose events skip signature verification — can only inject events inside the operator-declared scope. Out-of-scope deliveries surface on a 'filtered' counter. Malformed filter JSON fails the boot, not the first delivery. Several disjoint scopes for one upstream = repeat [[mirror]] with the same url. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A --- geode/config.example.toml | 10 +++++ .../kotlin/com/vitorpamplona/geode/Main.kt | 28 ++++++++++++-- .../geode/config/StaticConfig.kt | 12 ++++++ .../geode/mirror/MirrorWorker.kt | 37 ++++++++++++++++++- .../geode/config/StaticConfigTest.kt | 18 ++++++++- .../geode/mirror/MirrorWorkerTest.kt | 37 +++++++++++++++++-- 6 files changed, 132 insertions(+), 10 deletions(-) diff --git a/geode/config.example.toml b/geode/config.example.toml index 98ad0fc2d4..6b18b478d2 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -99,10 +99,20 @@ require_auth = false # mirror-but-verify. Only trust relays you operate or whose ingest # discipline you'd stake your own db on. # +# `filter` (optional) scopes an upstream, as a NIP-01 filter JSON +# object — same idea as strfry-router's per-stream filter. It shapes +# the REQ sent upstream AND every delivered event is re-checked +# against it before ingest, so even a trusted upstream can only +# inject events inside the declared scope. `since`/`limit` inside it +# are ignored (backfill_seconds owns the time window). Omit to mirror +# everything; for several disjoint scopes, repeat [[mirror]] with the +# same url. +# # [[mirror]] # url = "wss://upstream.example.com/" # trusted = true # backfill_seconds = 3600 +# filter = '{"kinds":[0,1,3,7],"#t":["nostr"]}' [admin] # NIP-86 relay management API. When `pubkeys` is non-empty, the relay diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 89dc1c989f..d096b34f8f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -26,6 +26,8 @@ import com.vitorpamplona.geode.config.RuntimeConfigData import com.vitorpamplona.geode.config.StaticConfig import com.vitorpamplona.geode.mirror.MirrorUpstream import com.vitorpamplona.geode.mirror.MirrorWorker +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy @@ -181,11 +183,29 @@ fun main(args: Array) { // anyway. Never mirror ourselves: a self-URL would echo every local // publish back forever. val upstreams = - config.mirror.map { + config.mirror.map { m -> + // The optional scope filter is parsed eagerly so a malformed + // JSON object fails the boot, not the first delivery. Its + // since/limit are stripped: the mirror owns the time window + // (backfill_seconds) and never bounds the subscription. + val scope = + m.filter?.let { json -> + val parsed = + try { + OptimizedJsonMapper.fromJsonTo(json) + } catch (e: Exception) { + throw IllegalArgumentException( + "[[mirror]] filter for ${m.url} is not a valid NIP-01 filter object: $json", + e, + ) + } + parsed.copy(since = null, limit = null) + } MirrorUpstream( - url = it.url.normalizeRelayUrl(), - trusted = it.trusted, - backfillSeconds = it.backfill_seconds, + url = m.url.normalizeRelayUrl(), + trusted = m.trusted, + backfillSeconds = m.backfill_seconds, + filter = scope, ) } require(upstreams.none { it.url == advertisedUrl }) { diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt index ba7291330a..7191da08f3 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt @@ -181,6 +181,18 @@ data class StaticConfig( val trusted: Boolean = false, /** How far back the initial subscription reaches. 0 = live-only. */ val backfill_seconds: Long = 0L, + /** + * Optional NIP-01 filter as a JSON object string (strfry-router + * parity), e.g. `'{"kinds":[0,1,3],"#t":["nostr"]}'`. Scopes what + * this upstream is asked for AND what it is allowed to deliver — + * every received event is re-checked against it before ingest, so + * even a trusted upstream can't push events outside the declared + * scope. `since` is managed by the mirror (see [backfill_seconds]) + * and `limit` is transport-level, so both are ignored if present. + * Omitted = mirror everything. For several disjoint filters, add + * several `[[mirror]]` entries with the same url. + */ + val filter: String? = null, ) /** diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt index f6f5b18468..2350bf72c5 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -55,6 +55,16 @@ class MirrorUpstream( val trusted: Boolean, /** How far back the initial REQ reaches. 0 = live-only from connect. */ val backfillSeconds: Long = 0L, + /** + * Optional scope for this upstream (strfry-router's per-stream + * `filter`). Used twice: it shapes the REQ sent upstream, and every + * delivered event is re-checked against it before ingest — so even a + * [trusted] upstream can only inject events inside the declared + * scope. Its `since`/`limit` are ignored ([backfillSeconds] owns the + * time window; the subscription is unbounded). `null` mirrors + * everything. + */ + val filter: Filter? = null, ) /** @@ -113,6 +123,13 @@ class MirrorWorker( /** Events the store rejected — mostly duplicate replays after a reconnect. */ val rejected = AtomicLong(0) + /** + * Deliveries dropped by the [MirrorUpstream.filter] re-check before + * ever reaching the store — an upstream sending these is answering + * outside the REQ it was given. + */ + val filtered = AtomicLong(0) + /** Dials every upstream and starts streaming. Call once. */ fun start() { scope.launch { @@ -151,6 +168,17 @@ class MirrorWorker( relay: NormalizedRelayUrl, forFilters: List?, ) { + // strfry-router parity: never take the upstream's + // word for what matched. Re-checking the configured + // scope here means even a trusted (skip-verify) + // upstream can only inject events the operator + // declared — the REQ shapes what we ask for, this + // shapes what we accept. + if (up.filter != null && !up.filter.match(event)) { + filtered.incrementAndGet() + Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" } + return + } inbound.trySend(Inbound(event, up.trusted)) } @@ -162,9 +190,16 @@ class MirrorWorker( Log.w("MirrorWorker") { "cannot reach upstream ${relay.url}: $message" } } } + // The operator's filter scopes the REQ; the mirror owns the + // time window (since) and never bounds the result (limit). + val reqFilter = + (up.filter ?: Filter()).copy( + since = since - up.backfillSeconds, + limit = null, + ) client.subscribe( subId = "geode-mirror-$i", - filters = mapOf(up.url to listOf(Filter(since = since - up.backfillSeconds))), + filters = mapOf(up.url to listOf(reqFilter)), listener = listener, ) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt index 2d478e52ef..765bc58881 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt @@ -20,6 +20,8 @@ */ package com.vitorpamplona.geode.config +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import java.io.File import kotlin.test.Test import kotlin.test.assertEquals @@ -51,6 +53,16 @@ class StaticConfigTest { assertTrue(StaticConfig.fromToml("").mirror.isEmpty()) } + @Test + fun mirrorFilterJsonParsesToANip01Filter() { + // The exact parse Main.kt runs on [[mirror]].filter at boot. + val f = OptimizedJsonMapper.fromJsonTo("""{"kinds":[0,1],"#t":["nostr"],"since":123,"limit":5}""") + assertEquals(listOf(0, 1), f.kinds) + assertEquals(listOf("nostr"), f.tags?.get("t")) + assertEquals(123L, f.since) + assertEquals(5, f.limit) + } + @Test fun parsesMirrorUpstreams() { val toml = @@ -59,6 +71,7 @@ class StaticConfigTest { url = "wss://trusted.upstream.example/" trusted = true backfill_seconds = 3600 + filter = '{"kinds":[0,1,3],"#t":["nostr"]}' [[mirror]] url = "wss://public.upstream.example/" @@ -70,10 +83,13 @@ class StaticConfigTest { assertEquals("wss://trusted.upstream.example/", c.mirror[0].url) assertEquals(true, c.mirror[0].trusted) assertEquals(3600L, c.mirror[0].backfill_seconds) - // Trust is opt-in per upstream: the default is mirror-but-verify. + assertEquals("""{"kinds":[0,1,3],"#t":["nostr"]}""", c.mirror[0].filter) + // Trust and scoping are opt-in per upstream: the default is + // mirror-everything-but-verify. assertEquals("wss://public.upstream.example/", c.mirror[1].url) assertEquals(false, c.mirror[1].trusted) assertEquals(0L, c.mirror[1].backfill_seconds) + assertEquals(null, c.mirror[1].filter) } @Test diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt index 440e40abf7..e2f62e2531 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt @@ -77,9 +77,12 @@ class MirrorWorkerTest { hub.close() } - private fun startMirror(trusted: Boolean): MirrorWorker = + private fun startMirror( + trusted: Boolean, + filter: Filter? = null, + ): MirrorWorker = MirrorWorker( - upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600)), + upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600, filter = filter)), server = downstream.server, websocketBuilder = hub, ).also { @@ -87,12 +90,15 @@ class MirrorWorkerTest { it.start() } - private fun forgedEvent(idSeed: Int): Event = + private fun forgedEvent( + idSeed: Int, + kind: Int = 1, + ): Event = Event( id = idSeed.toString().padStart(64, '0'), pubKey = "1".repeat(64), createdAt = TimeUtils.now() - idSeed, - kind = 1, + kind = kind, tags = emptyArray(), content = "forged $idSeed", sig = "f".repeat(128), @@ -144,6 +150,29 @@ class MirrorWorkerTest { assertEquals(setOf(stored.id, live.id), ids) } + @Test + fun filterScopesTheMirrorToDeclaredKinds() = + runBlocking { + val wantedStored = forgedEvent(1, kind = 1) + val unwantedStored = forgedEvent(2, kind = 7) + hub.getOrCreate(upstreamUrl).preload(wantedStored, unwantedStored) + + startMirror(trusted = true, filter = Filter(kinds = listOf(1))) + awaitDownstreamCount(1) + + // Live tail: the out-of-scope kind is published FIRST, so by the + // time the in-scope one lands downstream (same connection, same + // ordered pipeline), the kind-7 has already had its chance. + hub.getOrCreate(upstreamUrl).publish(forgedEvent(3, kind = 7)) + val wantedLive = forgedEvent(4, kind = 1) + hub.getOrCreate(upstreamUrl).publish(wantedLive) + awaitDownstreamCount(2) + + val stored = downstreamStore.query(Filter()) + assertEquals(setOf(wantedStored.id, wantedLive.id), stored.map { it.id }.toSet()) + assertTrue(stored.all { it.kind == 1 }) + } + @Test fun untrustedUpstreamStillVerifiesEverything() = runBlocking {