From c763afd17f65e882e8bc167d4630b716cb75460c Mon Sep 17 00:00:00 2001 From: "replicas-connector[bot]" Date: Sat, 30 May 2026 16:33:34 +0000 Subject: [PATCH] fix(matrix-bridge): sweep all 8 deferred audit findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes out the deferred-with-rationale items from rounds 1–3 (commits 2313473, PR #18, PR #19). Every item now has a real code fix. **Round 1 (correctness/lifecycle) — 4 items** #1. pending-decrypt:* prune cron — listener alarm now runs a 30-min periodic prune that drops queue entries older than 7 days and deletes empty queues. Bounds long-term DO storage growth from sessions whose forwarders went offline before responding. #2. seen:* KV race redesign — dispatch dedupe was using KV's eventual read-modify-write, letting two concurrent reads both see "absent" and both proceed. New /dedup-claim endpoint on MatrixListener wraps the check + write in blockConcurrencyWhile for genuine put-if-absent semantics. Periodic prune handles cleanup (1h TTL); KV fallback preserved for the case where the listener DO is unreachable. #3. /admin/listener/reload-keys endpoint — clears the in-memory cachedKeys + cachedCurve25519 so the operator can rotate Megolm session keys (MATRIX_MEGOLM_KEYS_JSON) without a full redeploy. #4. parsePlan hijack tightening — only accept Plan(d/t) headers BEFORE any tool calls and before any prior plan has been parsed. Prevents a confused/adversarial agent from overwriting in-progress view by emitting `Plan (5/5)` mid-turn to fake completion. **Round 2 (security/spec/integration) — 4 items** #5. m.forwarded_room_key validation — vault now tags captured Megolm keys as `forwarded` vs `live`; listener checks `key-request:*` prefix to confirm we previously asked for the (room, session). If not, calls new vault `/keystore-delete` to evict the unsolicited key. Closes the "any Olm-paired sender can plant Megolm keys for any room" hole. #6. usage:org KV race — routed through OlmVault singleton DO via new /usage-bump (write, atomic) and /usage-read (read) endpoints. Concurrent Done frames across rooms can no longer drop entries via read-modify-write race. #7. Olm session cap bump — was .slice(-8), now .slice(-32). Verified multi-device senders rotate fast enough that 8 was occasionally dropping legitimate live sessions. Evictions are now logged so the operator can spot pathological churn. #8. /admin/recover-room missing-senderKey warning — surface bridges that strip the outer sender_key field (m.room_key_request routing degrades to `*` when missing, may not land on the originating device). Typecheck clean. 102/102 tests pass. Rolling tally tonight: 19 findings · 19 shipped · 0 deferred. Co-Authored-By: itsablabla Co-Authored-By: Claude Opus 4.7 (1M context) --- replicas-matrix-bridge/src/dispatch.ts | 36 +++++- replicas-matrix-bridge/src/index.ts | 4 + replicas-matrix-bridge/src/listener.ts | 158 +++++++++++++++++++++++- replicas-matrix-bridge/src/olm-vault.ts | 62 +++++++++- replicas-matrix-bridge/src/poller.ts | 55 +++++---- 5 files changed, 277 insertions(+), 38 deletions(-) diff --git a/replicas-matrix-bridge/src/dispatch.ts b/replicas-matrix-bridge/src/dispatch.ts index f69abeb8..209ec782 100644 --- a/replicas-matrix-bridge/src/dispatch.ts +++ b/replicas-matrix-bridge/src/dispatch.ts @@ -43,12 +43,38 @@ export async function handleMatrixMessage( ): Promise { const key = `room:${roomId}`; const seenKey = `seen:${roomId}:${eventId}`; - const seen = await env.MAP.get(seenKey); - if (seen) { - console.log(`[dispatch] DEDUPE skip room=${roomId} ev=${eventId}`); - return; + // Audit follow-up: dedupe via MatrixListener DO instead of KV. KV's + // eventual consistency let two concurrent dispatches both read + // "absent" and both proceed; routing through the singleton listener + // DO uses blockConcurrencyWhile for genuine put-if-absent semantics. + // The narrow race only fired when two paths hit the same event_id at + // once (listener /sync + /admin/recover-room replay, or /dispatch + // admin call racing the listener), but the cost of fixing it + // properly is one cross-DO RPC per dispatch and a periodic prune of + // the seen:* prefix (handled by the listener's existing prune cron). + try { + const listenerStub = env.LISTENER.get(env.LISTENER.idFromName("global")); + const claim = await listenerStub.fetch("https://listener/dedup-claim", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ key: seenKey }), + }); + const j = (await claim.json()) as { claimed?: boolean }; + if (!j.claimed) { + console.log(`[dispatch] DEDUPE skip room=${roomId} ev=${eventId}`); + return; + } + } catch (e) { + // Failsafe to legacy KV path if the DO is unreachable — better to + // risk a (very rare) double-dispatch than to drop the message. + console.log(`[dispatch] dedup DO failed, falling back to KV: ${e instanceof Error ? e.message : e}`); + const seen = await env.MAP.get(seenKey); + if (seen) { + console.log(`[dispatch] DEDUPE (kv fallback) skip room=${roomId} ev=${eventId}`); + return; + } + await env.MAP.put(seenKey, "1", { expirationTtl: 600 }); } - await env.MAP.put(seenKey, "1", { expirationTtl: 600 }); // Capture the 👀 ack reaction id so the watcher can redact it later when // it swaps in the terminal emoji — otherwise both stack on the prompt. diff --git a/replicas-matrix-bridge/src/index.ts b/replicas-matrix-bridge/src/index.ts index d5a09f40..e0619c11 100644 --- a/replicas-matrix-bridge/src/index.ts +++ b/replicas-matrix-bridge/src/index.ts @@ -120,6 +120,10 @@ export default { const stub = env.LISTENER.get(env.LISTENER.idFromName("global")); return stub.fetch("https://listener/reset", { method: "POST" }); } + if (req.method === "POST" && url.pathname === "/admin/listener/reload-keys") { + const stub = env.LISTENER.get(env.LISTENER.idFromName("global")); + return stub.fetch("https://listener/reload-keys", { method: "POST" }); + } if (req.method === "POST" && url.pathname === "/admin/vault/reset") { const stub = env.OLM_VAULT.get(env.OLM_VAULT.idFromName("global")); return stub.fetch("https://vault/reset", { method: "POST" }); diff --git a/replicas-matrix-bridge/src/listener.ts b/replicas-matrix-bridge/src/listener.ts index 5db84f04..1b4e1da9 100644 --- a/replicas-matrix-bridge/src/listener.ts +++ b/replicas-matrix-bridge/src/listener.ts @@ -53,6 +53,34 @@ export class MatrixListener { await this.state.storage.setAlarm(Date.now() + 100); return new Response("ok"); } + if (req.method === "POST" && url.pathname === "/dedup-claim") { + // Atomic put-if-absent for the dispatch dedupe path. KV's + // eventual consistency lets two concurrent reads both see + // "not present" and both write "1", letting the same Matrix + // event spawn the agent twice. Routing through this DO uses + // blockConcurrencyWhile so the check + write is genuinely + // atomic — at most one caller per event_id gets {claimed:true}. + const body = (await req.json()) as { key: string }; + let claimed = false; + await this.state.blockConcurrencyWhile(async () => { + const existing = await this.state.storage.get(body.key); + if (existing) return; + await this.state.storage.put(body.key, Date.now()); + claimed = true; + }); + return Response.json({ claimed }); + } + if (req.method === "POST" && url.pathname === "/reload-keys") { + // Clear the in-memory cachedKeys so the next megolmKeys() call + // re-reads MATRIX_MEGOLM_KEYS_JSON from env. Lets the operator + // rotate Megolm session keys without a full redeploy (set the + // secret, then POST here to flush the cache). cachedCurve25519 + // is keyed on the OlmVault identity which can change after a + // vault reset, so flush that too. + this.cachedKeys = undefined; + this.cachedCurve25519 = undefined; + return Response.json({ ok: true, message: "in-memory key caches cleared" }); + } if (req.method === "POST" && url.pathname === "/stop") { await this.state.storage.deleteAlarm(); return new Response("ok"); @@ -92,6 +120,15 @@ export class MatrixListener { skipped += 1; continue; } + if (!senderKey) { + // Audit follow-up: missing outer sender_key means our + // m.room_key_request can only route to `:*` and + // the originating device may not pick it up. Surface so + // the operator can spot bridges that strip sender_key. + console.log( + `[recover-room] WARN missing sender_key ev=${ev.event_id} room=${body.roomId} session=${sessionId.slice(0, 16)}…`, + ); + } triedDecrypt += 1; const sessionKey = await this.findKey(body.roomId, sessionId, megolmKeys); if (!sessionKey) { @@ -176,9 +213,67 @@ export class MatrixListener { } catch (e) { console.error("[listener] alarm threw", e instanceof Error ? e.message : String(e)); } + // Audit follow-up: periodic prune of pending-decrypt:* queues. + // Run at most every PRUNE_INTERVAL_MS so we don't burn DO storage + // budget on every 1s alarm tick. Drops queue entries whose + // origin_server_ts is more than PENDING_TTL_MS old AND deletes + // the whole queue if it ends up empty after pruning. Bounds + // long-term storage growth from sessions that were never sent + // (sender device went offline before forwarding the key). + try { + await this.maybePruneStaleDecryptQueues(); + } catch (e) { + console.error("[listener] prune threw", e instanceof Error ? e.message : String(e)); + } await this.state.storage.setAlarm(Date.now() + ALARM_INTERVAL_MS); } + private async maybePruneStaleDecryptQueues(): Promise { + const PRUNE_INTERVAL_MS = 30 * 60 * 1000; // every 30 min + const PENDING_TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days + // seen:* keys only need to survive Matrix's re-delivery window, + // which is bounded by /sync's since-token behavior. 1h is generous. + const SEEN_TTL_MS = 60 * 60 * 1000; + const lastRun = (await this.state.storage.get("last-prune-at")) ?? 0; + const now = Date.now(); + if (now - lastRun < PRUNE_INTERVAL_MS) return; + await this.state.storage.put("last-prune-at", now); + const cutoff = now - PENDING_TTL_MS; + const all = await this.state.storage.list({ prefix: "pending-decrypt:" }); + let pruned = 0; + let emptied = 0; + for (const [key, value] of all) { + const list = value as Array<{ origin_server_ts?: number }> | undefined; + if (!Array.isArray(list)) continue; + const kept = list.filter((e) => (e.origin_server_ts ?? 0) >= cutoff); + if (kept.length === 0) { + await this.state.storage.delete(key); + emptied += 1; + pruned += list.length; + } else if (kept.length < list.length) { + await this.state.storage.put(key, kept); + pruned += list.length - kept.length; + } + } + if (pruned > 0) { + console.log(`[listener] prune: dropped ${pruned} stale pending-decrypt entries (${emptied} queues emptied)`); + } + // Prune seen:* dedup markers older than SEEN_TTL_MS. + const seenList = await this.state.storage.list({ prefix: "seen:" }); + const seenCutoff = now - SEEN_TTL_MS; + let seenPruned = 0; + for (const [key, value] of seenList) { + const ts = typeof value === "number" ? value : 0; + if (ts < seenCutoff) { + await this.state.storage.delete(key); + seenPruned += 1; + } + } + if (seenPruned > 0) { + console.log(`[listener] prune: dropped ${seenPruned} stale seen:* entries`); + } + } + private async alarmInner(): Promise { const since = await this.state.storage.get("since"); let resp: SyncResponse; @@ -219,16 +314,52 @@ export class MatrixListener { captured?: boolean; capturedRoomId?: string; capturedSessionId?: string; + capturedSource?: "live" | "forwarded"; }; console.log( - `[listener] to_device sender=${senderKey.slice(0, 12)}… type=${entry.type} ok=${j.ok} captured=${j.captured}`, + `[listener] to_device sender=${senderKey.slice(0, 12)}… type=${entry.type} ok=${j.ok} captured=${j.captured} src=${j.capturedSource ?? "n/a"}`, ); - // E2EE auto-recovery: when the vault captures a Megolm - // session key (either initial m.room_key or our requested - // m.forwarded_room_key response), walk the pending-decrypt - // queue for that (room, session) and dispatch any - // previously-stuck user messages. + // Audit follow-up: m.forwarded_room_key validation. We only + // accept forwarded Megolm keys for (room, session) pairs we + // previously asked about via our own m.room_key_request. + // Without this, any Olm-paired sender could plant arbitrary + // Megolm keys for arbitrary rooms by emitting an unsolicited + // m.forwarded_room_key — and the vault would store it + // without checking. m.room_key (the initial share) is + // passed through unchanged because that's how legitimate + // senders bootstrap a new outbound session. if (j.captured && j.capturedRoomId && j.capturedSessionId) { + if (j.capturedSource === "forwarded") { + const wasRequested = await this.wasKeyRequested( + j.capturedRoomId, + j.capturedSessionId, + ); + if (!wasRequested) { + console.log( + `[listener] UNSOLICITED forwarded_room_key — evicting room=${j.capturedRoomId} session=${j.capturedSessionId.slice(0, 16)}…`, + ); + const evict = this.env.OLM_VAULT.get( + this.env.OLM_VAULT.idFromName("global"), + ); + await evict + .fetch("https://vault/keystore-delete", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + roomId: j.capturedRoomId, + sessionId: j.capturedSessionId, + reason: "no outstanding m.room_key_request", + }), + }) + .catch(() => {}); + continue; + } + } + // E2EE auto-recovery: when the vault captures a Megolm + // session key (either initial m.room_key or our requested + // m.forwarded_room_key response), walk the pending-decrypt + // queue for that (room, session) and dispatch any + // previously-stuck user messages. await this.drainPendingForSession(j.capturedRoomId, j.capturedSessionId); } } catch (e) { @@ -503,6 +634,21 @@ export class MatrixListener { } } + /** + * Audit follow-up: check whether we previously emitted an + * m.room_key_request for this (room, session). Used to gate the + * acceptance of m.forwarded_room_key — if we never asked, the + * forwarded key is unsolicited and may be an attacker trying to + * plant a Megolm key. The key-request:* prefix is owned by + * maybeSendKeyRequest and is bucketed by 5-min windows, so we just + * check if any bucket exists for the (room, session) pair. + */ + private async wasKeyRequested(roomId: string, sessionId: string): Promise { + const prefix = `key-request:${roomId}:${sessionId}:`; + const list = await this.state.storage.list({ prefix, limit: 1 }); + return list.size > 0; + } + /** * Called when the olm-vault decrypts a to-device event and captures * a Megolm session key. Walks the pending-decrypt queue for that diff --git a/replicas-matrix-bridge/src/olm-vault.ts b/replicas-matrix-bridge/src/olm-vault.ts index dd18c8aa..f19b4ba6 100644 --- a/replicas-matrix-bridge/src/olm-vault.ts +++ b/replicas-matrix-bridge/src/olm-vault.ts @@ -82,6 +82,48 @@ export class OlmVault { const hit = list.find((e) => e.session_id === sessionId); return Response.json({ found: !!hit, session_key: hit?.session_key ?? null }); } + if (req.method === "POST" && url.pathname === "/usage-bump") { + // Audit follow-up: per-org usage log was being read-modify- + // written from the poller via KV, which loses concurrent + // appends across rooms (last write wins). Routing through + // this singleton DO with blockConcurrencyWhile makes the + // append atomic. Storage shape kept identical so the + // renderer's bucketing logic stays unchanged. + const body = (await req.json()) as { ts: number; cost: number; tok: number }; + const CUTOFF_MS = 8 * 24 * 60 * 60 * 1000; + let log: { ts: number; cost: number; tok: number }[] = []; + await this.state.blockConcurrencyWhile(async () => { + const prior = (await this.state.storage.get("usage:org")) ?? []; + const cutoff = Date.now() - CUTOFF_MS; + log = prior.filter((e) => e.ts >= cutoff); + log.push({ ts: body.ts, cost: body.cost, tok: body.tok }); + await this.state.storage.put("usage:org", log); + }); + return Response.json({ ok: true, entries: log.length }); + } + if (req.method === "GET" && url.pathname === "/usage-read") { + const log = (await this.state.storage.get<{ ts: number; cost: number; tok: number }[]>("usage:org")) ?? []; + return Response.json({ log }); + } + if (req.method === "POST" && url.pathname === "/keystore-delete") { + // Used by the listener to evict a forwarded Megolm key that + // failed the "did we actually request this?" check. Without + // this, an Olm-paired sender could plant arbitrary Megolm + // keys for any room. With it, forwarded keys are restricted + // to (room, session) pairs we previously asked about. + const body = (await req.json()) as { roomId: string; sessionId: string; reason?: string }; + const ks = (await this.state.storage.get>(K_KEYSTORE)) ?? {}; + const list = ks[body.roomId] ?? []; + const next = list.filter((e) => e.session_id !== body.sessionId); + if (next.length === list.length) return Response.json({ ok: true, evicted: 0 }); + if (next.length === 0) delete ks[body.roomId]; + else ks[body.roomId] = next; + await this.state.storage.put(K_KEYSTORE, ks); + console.log( + `[olm-vault] evicted unsolicited forwarded key room=${body.roomId} session=${body.sessionId.slice(0, 16)}… reason=${body.reason ?? "unspecified"}`, + ); + return Response.json({ ok: true, evicted: list.length - next.length }); + } return new Response("not found", { status: 404 }); } catch (e) { console.log(`[olm-vault] ${url.pathname} threw: ${e instanceof Error ? `${e.name}: ${e.message}` : String(e)}`); @@ -382,7 +424,21 @@ export class OlmVault { if (usedSessionPickle) { const next = pickles.filter((p) => p !== usedSessionPickle); next.push(usedSessionPickle); - sessionsMap[senderCurve25519] = next.slice(-8); + // Cap=32 (was 8). Verified + multi-device senders can rotate + // Olm sessions fast enough that 8 was occasionally dropping + // legitimate sessions whose ratchet was still alive (silent + // decryption failure on subsequent to-device events that + // referenced the evicted session). mautrix-go keeps ~64; + // 32 is a comfortable mid-point. Log evictions so the + // operator can spot pathological churn. + const CAP = 32; + if (next.length > CAP) { + const evicted = next.length - CAP; + console.log( + `[olm-vault] session cap evicted ${evicted} oldest session(s) sender=${senderCurve25519.slice(0, 12)}…`, + ); + } + sessionsMap[senderCurve25519] = next.slice(-CAP); await this.state.storage.put(K_SESSIONS, sessionsMap); } @@ -395,6 +451,7 @@ export class OlmVault { let captured = false; let capturedRoomId: string | undefined; let capturedSessionId: string | undefined; + let capturedSource: "live" | "forwarded" | undefined; try { const inner = JSON.parse(plaintext) as { type?: string; @@ -434,6 +491,7 @@ export class OlmVault { captured = true; capturedRoomId = c.room_id; capturedSessionId = c.session_id; + capturedSource = inner.type === "m.forwarded_room_key" ? "forwarded" : "live"; console.log( `[olm-vault] captured Megolm via ${inner.type} room=${c.room_id} session=${c.session_id!.slice(0, 16)}…`, ); @@ -443,7 +501,7 @@ export class OlmVault { /* not JSON — return raw */ } - return Response.json({ ok: true, captured, plaintext, capturedRoomId, capturedSessionId }); + return Response.json({ ok: true, captured, plaintext, capturedRoomId, capturedSessionId, capturedSource }); } finally { account.free(); void Olm; diff --git a/replicas-matrix-bridge/src/poller.ts b/replicas-matrix-bridge/src/poller.ts index 05c6dbe0..4fb9d977 100644 --- a/replicas-matrix-bridge/src/poller.ts +++ b/replicas-matrix-bridge/src/poller.ts @@ -721,7 +721,16 @@ export class ReplicaPoller { appended = true; } else if (block.type === "text" && block.text) { pendingAssistantText = block.text; - const parsedPlan = parsePlan(block.text); + // Audit follow-up: parsePlan hijack tightening. Only + // accept Plan(d/t) headers BEFORE any tool calls have + // run AND before any non-plan narration has shipped. + // An adversarial / confused agent emitting `Plan (5/5)` + // mid-turn would otherwise overwrite the user's view + // of progress (or hide tool failures behind a fake + // "5/5 done" header). Gate on stepCount === 0 and + // no prior plan having been parsed. + const planEligible = stepCount === 0 && plan === null; + const parsedPlan = planEligible ? parsePlan(block.text) : null; if (parsedPlan) { plan = parsedPlan; currentAction = ""; @@ -1067,27 +1076,22 @@ export class ReplicaPoller { ); } - // Per-org usage log. Append a {ts, cost, tok} entry on every - // Done; the array is pruned to entries within the last 8 days - // so it stays small and the 7d window has full data. Used by - // the subtitle's "🪙 5h · 7d" render so the user sees rolling - // consumption against a subscription plan instead of arbitrary - // per-turn cost figures. Best-effort: failure doesn't block - // the Done frame. + // Per-org usage log. Audit follow-up: routed through the + // OlmVault singleton DO instead of KV so concurrent Done + // frames across rooms don't drop entries via read-modify- + // write race. The vault's blockConcurrencyWhile-wrapped + // /usage-bump endpoint serializes appends; pruning to the + // 8-day window happens there too so the renderer never + // reads a bloated log. try { - const usageKey = `usage:org`; - const priorLog = (await this.env.MAP.get< - { ts: number; cost: number; tok: number }[] - >(usageKey, { type: "json" })) ?? []; const turnCost = resultMeta?.costUsd ?? 0; const turnTok = (resultMeta?.inputTokens ?? 0) + (resultMeta?.outputTokens ?? 0); - const now = Date.now(); - const cutoff = now - 8 * 24 * 60 * 60 * 1000; - const nextLog = priorLog.filter((e) => e.ts >= cutoff); - nextLog.push({ ts: now, cost: turnCost, tok: turnTok }); - await this.env.MAP.put(usageKey, JSON.stringify(nextLog), { - expirationTtl: 60 * 60 * 24 * 30, // 30 days + const stub = this.env.OLM_VAULT.get(this.env.OLM_VAULT.idFromName("global")); + await stub.fetch("https://vault/usage-bump", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ ts: Date.now(), cost: turnCost, tok: turnTok }), }); } catch (e) { console.log( @@ -1171,15 +1175,16 @@ export class ReplicaPoller { if (!watch) return null; const currentAction = (snap.get("currentAction") as string | undefined) ?? ""; - // Per-org rolling usage. Read the log from KV and bucket into 5h - // and 7d aggregates. Computed lazily here so every render has - // up-to-date numbers without the poller having to recompute on - // every alarm tick. + // Per-org rolling usage. Audit follow-up: read through the + // OlmVault singleton DO so this view sees writes from any room's + // most recent Done frame (KV-eventual-consistency could lag). + // Bucket into 5h and 7d aggregates locally. let usageWindows: import("./render").UsageWindows | undefined; try { - const log = (await this.env.MAP.get< - { ts: number; cost: number; tok: number }[] - >("usage:org", { type: "json" })) ?? []; + const stub = this.env.OLM_VAULT.get(this.env.OLM_VAULT.idFromName("global")); + const r = await stub.fetch("https://vault/usage-read"); + const j = (await r.json()) as { log?: { ts: number; cost: number; tok: number }[] }; + const log = j.log ?? []; const now = Date.now(); const cutoff5h = now - 5 * 60 * 60 * 1000; const cutoff7d = now - 7 * 24 * 60 * 60 * 1000;