diff --git a/.changeset/fix-relay-reconnect-replay.md b/.changeset/fix-relay-reconnect-replay.md new file mode 100644 index 0000000..3b7f4de --- /dev/null +++ b/.changeset/fix-relay-reconnect-replay.md @@ -0,0 +1,7 @@ +--- +'@contextvm/sdk': patch +--- + +Prevent relay reconnects from dispatching previously seen server requests again after deduplication cache eviction. Retain verified plaintext and decrypted inner-event IDs until the server transport closes, while keeping envelope deduplication bounded as a decryption optimization. This also suppresses replay through a new envelope containing an old signed inner event. + +Replay-protection memory now grows with unique verified inbound events, including notifications, transfer frames, and events rejected after signature verification. Newly signed events remain distinct requests. This protection does not persist across transport closure or process restart, cache responses, or guarantee exactly-once side effects; applications still need idempotency for retries. diff --git a/README.md b/README.md index 40f4841..2d8ea8a 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,14 @@ import { Proxy } from '@contextvm/sdk/proxy'; Visit the [ContextVM documentation](https://contextvm.org) for information on how to use ContextVM. +### Server replay protection + +`NostrServerTransport` retains verified plaintext and decrypted inner-event IDs until `close()`. Relay reconnects, completed responses, and client-session eviction do not erase these IDs, so replaying the same signed event cannot dispatch it again during that transport's lifetime. This also covers the same inner event delivered in a new encrypted envelope. + +Memory usage grows with the number of unique verified events, including notifications and transfer frames. IDs are recorded before authorization and message processing, so events subsequently rejected by those checks also remain recorded. A separate bounded envelope cache avoids repeated decryption; eviction from that cache does not remove replay protection. + +This protection is scoped to signed event IDs. A newly signed event with a different ID is a new request, even if its JSON-RPC ID and payload match an earlier request. Responses are not cached or replayed, and deduplication state does not survive transport closure or process restart. Tools with side effects still need application-level idempotency for retries and crash recovery; transport deduplication does not guarantee exactly-once side effects. + ### Logging The SDK uses Pino for high-performance logging with structured JSON output. By default, logs are written to stderr to comply with the MCP protocol expectations. diff --git a/src/transport/nostr-server-transport.ts b/src/transport/nostr-server-transport.ts index 4e53950..bfd58fe 100644 --- a/src/transport/nostr-server-transport.ts +++ b/src/transport/nostr-server-transport.ts @@ -173,6 +173,9 @@ export type ListToolsAnnouncementTagsProducer = ( * This transport listens for incoming MCP requests via Nostr events and can send * responses back to the originating clients. It handles all request/response correlation * internally, making it a standalone MCP transport that works over Nostr. + * Verified inbound event IDs are retained until close() to prevent replay dispatch. + * Memory grows with unique verified events; this does not guarantee exactly-once + * side effects or persist replay protection across transport lifetimes. */ export class NostrServerTransport extends BaseNostrTransport @@ -215,12 +218,11 @@ export class NostrServerTransport private readonly listToolsAnnouncementTagsProducers: ListToolsAnnouncementTagsProducer[] = []; - /** - * Deduplicate inbound events to avoid redundant work. - * - * Used for gift-wrap envelopes (outer event ids) and decrypted inner events. - */ - private readonly seenEventIds = new LruCache(DEFAULT_LRU_SIZE); + /** Bounded envelope cache to avoid repeated decryption. */ + private readonly seenGiftWrapIds = new LruCache(DEFAULT_LRU_SIZE); + + /** Retain verified inner/plain IDs until close: relays can replay the full session. */ + private readonly seenEventIds = new Set(); /** Receives inbound oversized-transfer frames from clients (client→server requests). */ private readonly oversizedReceiver: OversizedTransferReceiver; @@ -469,6 +471,7 @@ export class NostrServerTransport this.eventPipeline = new ServerEventPipeline({ signer: this.signer, + seenGiftWrapIds: this.seenGiftWrapIds, seenEventIds: this.seenEventIds, encryptionMode: this.encryptionMode, giftWrapMode: this.giftWrapMode, @@ -606,6 +609,7 @@ export class NostrServerTransport this.sessionStore.clear(); this.correlationStore.clear(); this.subscriptionStore.clear(); + this.seenGiftWrapIds.clear(); this.seenEventIds.clear(); this.oversizedReceiver.clear(); this.openStreamFactory.getReceiver().clear(); diff --git a/src/transport/nostr-server/event-pipeline.test.ts b/src/transport/nostr-server/event-pipeline.test.ts new file mode 100644 index 0000000..a33479d --- /dev/null +++ b/src/transport/nostr-server/event-pipeline.test.ts @@ -0,0 +1,273 @@ +import { describe, expect, mock, test } from 'bun:test'; +import type { NostrEvent } from 'nostr-tools'; +import { finalizeEvent, generateSecretKey } from 'nostr-tools/pure'; +import { + CTXVM_MESSAGES_KIND, + EPHEMERAL_GIFT_WRAP_KIND, + GIFT_WRAP_KIND, +} from '../../core/constants.js'; +import { encryptMessage } from '../../core/encryption.js'; +import { EncryptionMode, GiftWrapMode } from '../../core/interfaces.js'; +import { createLogger } from '../../core/utils/logger.js'; +import { LruCache } from '../../core/utils/lru-cache.js'; +import { PrivateKeySigner } from '../../signer/private-key-signer.js'; +import { ServerEventPipeline } from './event-pipeline.js'; + +const CACHE_CAPACITY = 4; + +function relayCopy(event: NostrEvent): NostrEvent { + return JSON.parse(JSON.stringify(event)) as NostrEvent; +} + +async function createFixture() { + const signer = new PrivateKeySigner(); + const serverPubkey = await signer.getPublicKey(); + const clientKey = generateSecretKey(); + const seenGiftWrapIds = new LruCache(CACHE_CAPACITY); + const seenEventIds = new Set(); + const decrypt = mock(signer.nip44.decrypt); + signer.nip44.decrypt = decrypt; + const onerror = mock((_error: Error): void => {}); + const pipeline = new ServerEventPipeline({ + signer, + seenGiftWrapIds, + seenEventIds, + encryptionMode: EncryptionMode.OPTIONAL, + giftWrapMode: GiftWrapMode.OPTIONAL, + logger: createLogger('server-event-pipeline-test', { level: 'silent' }), + onerror, + }); + + const makeEvent = ( + sequence: number, + createdAt = 100, + content = JSON.stringify({ + jsonrpc: '2.0', + id: sequence, + method: 'tools/call', + params: { name: 'counter' }, + }), + ): NostrEvent => + relayCopy( + finalizeEvent( + { + kind: CTXVM_MESSAGES_KIND, + created_at: createdAt, + tags: [ + ['p', serverPubkey], + ['test-sequence', String(sequence)], + ], + content, + }, + clientKey, + ), + ); + + const wrap = (event: NostrEvent, kind = GIFT_WRAP_KIND): NostrEvent => + encryptMessage(JSON.stringify(event), serverPubkey, kind); + + const fillEnvelopeCache = async (): Promise => { + for (let index = 1; index <= CACHE_CAPACITY + 1; index++) { + expect(await pipeline.unwrap(wrap(makeEvent(index)))).not.toBeNull(); + } + }; + + return { + pipeline, + signer, + serverPubkey, + seenGiftWrapIds, + decrypt, + onerror, + makeEvent, + wrap, + fillEnvelopeCache, + }; +} + +describe('ServerEventPipeline replay protection', () => { + test.each([GIFT_WRAP_KIND, EPHEMERAL_GIFT_WRAP_KIND])( + 'does not unwrap a verified event again after its kind %i envelope is evicted', + async (kind) => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + const envelope = fixture.wrap(original, kind); + expect((await fixture.pipeline.unwrap(envelope))?.event.id).toBe( + original.id, + ); + + await fixture.fillEnvelopeCache(); + expect(fixture.seenGiftWrapIds.has(envelope.id)).toBe(false); + const decryptCalls = fixture.decrypt.mock.calls.length; + + expect(await fixture.pipeline.unwrap(relayCopy(envelope))).toBeNull(); + expect(fixture.decrypt.mock.calls.length).toBe(decryptCalls + 1); + expect(fixture.onerror).not.toHaveBeenCalled(); + }, + ); + + test('rejects an old inner event in a fresh envelope after cache pressure', async () => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + const firstEnvelope = fixture.wrap(original); + expect(await fixture.pipeline.unwrap(firstEnvelope)).not.toBeNull(); + + await fixture.fillEnvelopeCache(); + const newEnvelope = fixture.wrap(original); + expect(newEnvelope.id).not.toBe(firstEnvelope.id); + expect(await fixture.pipeline.unwrap(newEnvelope)).toBeNull(); + }); + + test('rejects a previously seen plaintext event after more events than the cache capacity', async () => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + expect(await fixture.pipeline.unwrap(original)).not.toBeNull(); + + for (let index = 1; index <= CACHE_CAPACITY + 1; index++) { + expect( + await fixture.pipeline.unwrap(fixture.makeEvent(index)), + ).not.toBeNull(); + } + + expect(await fixture.pipeline.unwrap(relayCopy(original))).toBeNull(); + expect(fixture.decrypt).not.toHaveBeenCalled(); + }); + + test('shares verified identity between plaintext and encrypted delivery after cache pressure', async () => { + const fixture = await createFixture(); + const plainFirst = fixture.makeEvent(-1); + const encryptedFirst = fixture.makeEvent(0); + expect(await fixture.pipeline.unwrap(plainFirst)).not.toBeNull(); + expect( + await fixture.pipeline.unwrap(fixture.wrap(encryptedFirst)), + ).not.toBeNull(); + + await fixture.fillEnvelopeCache(); + expect(await fixture.pipeline.unwrap(fixture.wrap(plainFirst))).toBeNull(); + expect(await fixture.pipeline.unwrap(relayCopy(encryptedFirst))).toBeNull(); + }); + + test('deduplicates an envelope before decryption while its cache entry remains', async () => { + const fixture = await createFixture(); + const envelope = fixture.wrap(fixture.makeEvent(0)); + const results = await Promise.all([ + fixture.pipeline.unwrap(envelope), + fixture.pipeline.unwrap(relayCopy(envelope)), + ]); + + expect(results.filter((result) => result !== null)).toHaveLength(1); + expect(fixture.decrypt).toHaveBeenCalledTimes(1); + }); + + test('claims an inner identity once when distinct envelopes finish asynchronous decryption together', async () => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + let releaseDecrypt!: () => void; + const decryptGate = new Promise((resolve) => { + releaseDecrypt = resolve; + }); + fixture.signer.nip44.decrypt = async ( + pubkey: string, + ciphertext: string, + ): Promise => { + await decryptGate; + return fixture.decrypt(pubkey, ciphertext); + }; + + const resultsPromise = Promise.all([ + fixture.pipeline.unwrap(fixture.wrap(original)), + fixture.pipeline.unwrap(fixture.wrap(original)), + ]); + expect(fixture.decrypt).not.toHaveBeenCalled(); + releaseDecrypt(); + + const results = await resultsPromise; + expect(results.filter((result) => result !== null)).toHaveLength(1); + expect(fixture.decrypt).toHaveBeenCalledTimes(2); + expect(fixture.onerror).not.toHaveBeenCalled(); + }); + + test.each(['plain', 'encrypted'] as const)( + 'does not let an invalid %s signature reserve the genuine event identity', + async (delivery) => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + const invalid = { ...relayCopy(original), sig: '0'.repeat(128) }; + const deliver = (event: NostrEvent): Promise => + fixture.pipeline.unwrap( + delivery === 'encrypted' ? fixture.wrap(event) : event, + ); + + expect(await deliver(invalid)).toBeNull(); + expect(await deliver(relayCopy(original))).not.toBeNull(); + expect(fixture.onerror).not.toHaveBeenCalled(); + }, + ); + + test('reports decryption errors without reserving the enclosed event identity', async () => { + const fixture = await createFixture(); + const original = fixture.makeEvent(0); + const failure = new Error('signer unavailable'); + fixture.decrypt.mockRejectedValueOnce(failure); + + expect(await fixture.pipeline.unwrap(fixture.wrap(original))).toBeNull(); + expect(fixture.onerror).toHaveBeenCalledTimes(1); + expect(fixture.onerror).toHaveBeenCalledWith(failure); + expect( + await fixture.pipeline.unwrap(fixture.wrap(original)), + ).not.toBeNull(); + }); + + test('reports malformed decrypted JSON and accepts a later valid envelope', async () => { + const fixture = await createFixture(); + const malformed = encryptMessage('{', fixture.serverPubkey); + + expect(await fixture.pipeline.unwrap(malformed)).toBeNull(); + expect(fixture.onerror).toHaveBeenCalledTimes(1); + expect(fixture.onerror.mock.calls[0]?.[0]).toBeInstanceOf(SyntaxError); + expect( + await fixture.pipeline.unwrap(fixture.wrap(fixture.makeEvent(0))), + ).not.toBeNull(); + }); + + test('accepts distinct signed events with identical payloads in the same second and out of order', async () => { + const fixture = await createFixture(); + const payload = fixture.makeEvent(0).content; + const events = [ + fixture.makeEvent(1, 100, payload), + fixture.makeEvent(2, 100, payload), + fixture.makeEvent(3, 99, payload), + ]; + + for (const event of events) { + expect( + (await fixture.pipeline.unwrap(fixture.wrap(event)))?.event.id, + ).toBe(event.id); + } + expect(fixture.onerror).not.toHaveBeenCalled(); + }); + + test.each(['oversized-transfer', 'open-stream'])( + 'retains %s notification frame identities after cache pressure', + async (type) => { + const fixture = await createFixture(); + const frame = fixture.makeEvent( + 0, + 100, + JSON.stringify({ + jsonrpc: '2.0', + method: 'notifications/progress', + params: { + progressToken: 'request-stream', + progress: 1, + cvm: { type, frameType: 'abort', reason: 'request cancelled' }, + }, + }), + ); + expect(await fixture.pipeline.unwrap(fixture.wrap(frame))).not.toBeNull(); + + await fixture.fillEnvelopeCache(); + expect(await fixture.pipeline.unwrap(fixture.wrap(frame))).toBeNull(); + }, + ); +}); diff --git a/src/transport/nostr-server/event-pipeline.ts b/src/transport/nostr-server/event-pipeline.ts index 0f185d7..439de64 100644 --- a/src/transport/nostr-server/event-pipeline.ts +++ b/src/transport/nostr-server/event-pipeline.ts @@ -18,7 +18,8 @@ import { withTimeout } from '../../core/utils/utils.js'; /** Dependencies for the server-side event decryption and verification pipeline. */ export interface ServerEventPipelineDeps { signer: NostrSigner; - seenEventIds: LruCache; + seenGiftWrapIds: LruCache; + seenEventIds: Set; encryptionMode: EncryptionMode; giftWrapMode: GiftWrapMode; logger: Logger; @@ -57,13 +58,13 @@ export class ServerEventPipeline { } // Deduplicate gift-wrap envelopes before any expensive decryption. - if (this.deps.seenEventIds.has(event.id)) { + if (this.deps.seenGiftWrapIds.has(event.id)) { this.deps.logger.debug('Skipping duplicate gift-wrapped event', { eventId: event.id, }); return null; } - this.deps.seenEventIds.set(event.id, true); + this.deps.seenGiftWrapIds.set(event.id, true); return await this.handleEncryptedEvent(event); } else { @@ -133,7 +134,7 @@ export class ServerEventPipeline { }); return null; } - this.deps.seenEventIds.set(currentEvent.id, true); + this.deps.seenEventIds.add(currentEvent.id); return { event: currentEvent, isEncrypted: true, wrapKind: event.kind }; } catch (error) { @@ -178,7 +179,7 @@ export class ServerEventPipeline { }); return null; } - this.deps.seenEventIds.set(event.id, true); + this.deps.seenEventIds.add(event.id); return { event, isEncrypted: false }; } diff --git a/src/transport/nostr-transport-reconnection.test.ts b/src/transport/nostr-transport-reconnection.test.ts index b6cdd5a..10aa519 100644 --- a/src/transport/nostr-transport-reconnection.test.ts +++ b/src/transport/nostr-transport-reconnection.test.ts @@ -16,9 +16,13 @@ import { bytesToHex, hexToBytes } from 'nostr-tools/utils'; import { ApplesauceRelayPool } from '../relay/applesauce-relay-pool.js'; import { McpServer } from '@contextvm/mcp-sdk/server/mcp'; import { z } from 'zod'; -import { EncryptionMode } from '../core/interfaces.js'; -import type { CallToolResult } from '@contextvm/mcp-sdk/types.js'; +import { EncryptionMode, GiftWrapMode } from '../core/interfaces.js'; +import { + type CallToolResult, + isJSONRPCRequest, +} from '@contextvm/mcp-sdk/types.js'; import { DEFAULT_TIMEOUT_MS } from '../core/constants.js'; +import { waitFor } from '../core/utils/test.utils.js'; import { spawnMockRelay, restartMockRelay, @@ -190,6 +194,178 @@ describe.serial('NostrTransport Reconnection', () => { ); }; + test.serial( + 'does not execute completed encrypted tool calls again after cache eviction and relay replay', + async () => { + const requestCount = 12; + const executions = new Map(); + const server = new McpServer({ + name: 'Replay-Safety-Server', + version: '1.0.0', + }); + server.registerTool( + 'count', + { inputSchema: { sequence: z.number() } }, + async ({ sequence }) => { + executions.set(sequence, (executions.get(sequence) ?? 0) + 1); + return { content: [{ type: 'text', text: String(sequence) }] }; + }, + ); + + const pool = new ApplesauceRelayPool([relayUrl], { + reconnectBaseDelayMs: 100, + reconnectMaxDelayMs: 1_000, + }); + let receivedEvents = 0; + let processedEvents = 0; + let eoseCount = 0; + let firstEnvelopeId: string | undefined; + const subscribe = pool.subscribe.bind(pool); + pool.subscribe = (filters, onEvent, onEose) => + subscribe( + filters, + (event) => { + receivedEvents += 1; + firstEnvelopeId ??= event.id; + onEvent(event); + }, + () => { + eoseCount += 1; + onEose?.(); + }, + ); + + const serverTransport = new NostrServerTransport({ + signer: new PrivateKeySigner(serverPrivateKey), + relayHandler: pool, + encryptionMode: EncryptionMode.REQUIRED, + giftWrapMode: GiftWrapMode.PERSISTENT, + maxSessions: 1, + }); + // Exercise real eviction with a small test-only cache, keeping CI fast. + const envelopeCache = serverTransport['seenGiftWrapIds']; + envelopeCache['capacity'] = 4; + const processEvent = + serverTransport['processIncomingEvent'].bind(serverTransport); + serverTransport['processIncomingEvent'] = async ( + event, + ): Promise => { + try { + await processEvent(event); + } finally { + processedEvents += 1; + } + }; + + const client = new Client({ + name: 'Replay-Safety-Client', + version: '1.0.0', + }); + const clientTransport = new NostrClientTransport({ + signer: new PrivateKeySigner(bytesToHex(generateSecretKey())), + relayHandler: new ApplesauceRelayPool([relayUrl], { + reconnectBaseDelayMs: 100, + reconnectMaxDelayMs: 1_000, + }), + serverPubkey: serverPublicKey, + isStateless: true, + encryptionMode: EncryptionMode.REQUIRED, + giftWrapMode: GiftWrapMode.PERSISTENT, + }); + const pendingResponses = new Set(); + const requestEventIds = new Set(); + + try { + await server.connect(serverTransport); + const onmessage = serverTransport.onmessage; + serverTransport.onmessage = (message): void => { + if (isJSONRPCRequest(message)) { + pendingResponses.add(message.id); + requestEventIds.add(String(message.id)); + } + onmessage?.(message); + }; + const send = serverTransport.send.bind(serverTransport); + serverTransport.send = async (message): Promise => { + const requestId = 'id' in message ? message.id : undefined; + try { + await send(message); + } finally { + if (requestId !== undefined) { + pendingResponses.delete(requestId); + } + } + }; + await client.connect(clientTransport); + + for (let sequence = 0; sequence < requestCount; sequence += 1) { + const result = (await client.callTool({ + name: 'count', + arguments: { sequence }, + })) as CallToolResult; + expectTextResult(result, String(sequence)); + } + await waitFor({ + produce: () => + eoseCount > 0 && + processedEvents === receivedEvents && + pendingResponses.size === 0 + ? true + : undefined, + }); + expect([...executions.values()]).toEqual(Array(requestCount).fill(1)); + expect(firstEnvelopeId).toBeDefined(); + expect(envelopeCache.has(firstEnvelopeId!)).toBe(false); + expect( + serverTransport.getInternalStateForTesting().correlationStore + .eventRouteCount, + ).toBe(0); + const { sessionStore } = serverTransport.getInternalStateForTesting(); + const [clientPubkey] = [...sessionStore.getAllSessions()][0]; + sessionStore.getOrCreateSession( + getPublicKey(generateSecretKey()), + false, + ); + expect(sessionStore.hasSession(clientPubkey)).toBe(false); + expect(requestEventIds.size).toBe(requestCount); + for (const eventId of requestEventIds) { + expect(serverTransport['seenEventIds'].has(eventId)).toBe(true); + } + const eventsBeforeRestart = receivedEvents; + const eoseBeforeRestart = eoseCount; + + await restartPrimaryRelay(); + + // EOSE can arrive while decryptions are queued or MCP responses are pending. + await waitFor({ + produce: () => + eoseCount > eoseBeforeRestart && + receivedEvents >= eventsBeforeRestart + requestCount && + processedEvents === receivedEvents && + pendingResponses.size === 0 + ? true + : undefined, + timeoutMs: 15_000, + }); + expect([...executions.values()]).toEqual(Array(requestCount).fill(1)); + + const result = (await client.callTool({ + name: 'count', + arguments: { sequence: requestCount }, + })) as CallToolResult; + expectTextResult(result, String(requestCount)); + expect(executions.get(requestCount)).toBe(1); + expect(executions.size).toBe(requestCount + 1); + } finally { + await client.close(); + await server.close(); + expect(envelopeCache.size).toBe(0); + expect(serverTransport['seenEventIds'].size).toBe(0); + } + }, + 30_000, + ); + test.serial( 'should handle relay restart and continue processing requests', async () => {