fix(transport): prevent duplicate execution after relay reconnects

.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.

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.

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<true>(DEFAULT_LRU_SIZE);
+  /** Bounded envelope cache to avoid repeated decryption. */
+  private readonly seenGiftWrapIds = new LruCache<true>(DEFAULT_LRU_SIZE);
+
+  /** Retain verified inner/plain IDs until close: relays can replay the full session. */
+  private readonly seenEventIds = new Set<string>();
 
   /** 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();

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<true>(CACHE_CAPACITY);
+  const seenEventIds = new Set<string>();
+  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<void> => {
+    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<void>((resolve) => {
+      releaseDecrypt = resolve;
+    });
+    fixture.signer.nip44.decrypt = async (
+      pubkey: string,
+      ciphertext: string,
+    ): Promise<string> => {
+      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<unknown> =>
+        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();
+    },
+  );
+});

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<true>;
+  seenGiftWrapIds: LruCache<true>;
+  seenEventIds: Set<string>;
   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 };
   }