fix(adapters,plugins,edge): a push token replayed to another replica or Worker isolate is refused

apps/e2e/demo-e2e-cloudflare/e2e/helpers/managed-worker.ts

@@ -12,6 +12,10 @@ const WRANGLER_BIN = join(ROOT_DIR, 'node_modules', '.bin', 'wrangler');
 const MCP_URL = 'http://localhost/mcp';
 const PUSH_URL = 'http://localhost/__skilled_openapi/push';
 const BUNDLE_CACHE_KEY = 'frontmcp:bundle:last-good';
+const BUNDLE_CACHE_NAMESPACE_ID = 'e2e-bundle-cache';
+const PUSH_REPLAY_CLASS = 'FrontMcpPushReplay';
+const FIRST_REPLICA = 'replica-1';
+const SECOND_REPLICA = 'replica-2';
 
 interface JsonRpcResponse {
   result?: { isError?: boolean; content?: Array<{ type: string; text?: string }> };
@@ -29,14 +33,20 @@ export interface ManagedWorkerOptions {
   outbound: (request: Request) => Promise<Response>;
 }
 
-/** The fixture Worker, bundled by `wrangler deploy --dry-run` as it would be deployed, and run in workerd. */
+/**
+ * The fixture Worker, bundled by `wrangler deploy --dry-run` as it would be deployed, and run in workerd as two
+ * replicas: two Workers of one bundle, each in its own isolate with its own memory, sharing the KV namespace and the
+ * push replay Durable Object, as the isolates of a deployed Worker do. The tests use the first unless they ask for the
+ * {@link ManagedWorker.secondReplica}.
+ */
 export class ManagedWorker {
   private requestId = 0;
 
   private constructor(
     private readonly miniflare: Miniflare,
     private readonly bundleDir: string,
     private readonly runtimeLog: string[],
+    private readonly replicaName = FIRST_REPLICA,
   ) {}
 
   static async start({ bindings, outbound }: ManagedWorkerOptions): Promise<ManagedWorker> {
@@ -47,14 +57,25 @@ export class ManagedWorker {
     });
     const runtimeLog: string[] = [];
     const collect = (stream: Readable) => stream.on('data', (chunk: Buffer) => runtimeLog.push(chunk.toString()));
-    const miniflare = new Miniflare({
+    const replica = (
+      name: string,
+      pushReplayObject: { className: string; scriptName?: string; useSQLite: boolean },
+    ) => ({
+      name,
       modulesRoot: bundleDir,
-      modules: [{ type: 'ESModule', path: join(bundleDir, 'worker.js') }],
+      modules: [{ type: 'ESModule' as const, path: join(bundleDir, 'worker.js') }],
       compatibilityDate: '2024-09-23',
       compatibilityFlags: ['nodejs_compat'],
-      kvNamespaces: ['BUNDLE_CACHE'],
+      kvNamespaces: { BUNDLE_CACHE: BUNDLE_CACHE_NAMESPACE_ID },
+      durableObjects: { FRONTMCP_PUSH_REPLAY: pushReplayObject },
       bindings,
       outboundService: outbound,
+    });
+    const miniflare = new Miniflare({
+      workers: [
+        replica(FIRST_REPLICA, { className: PUSH_REPLAY_CLASS, useSQLite: true }),
+        replica(SECOND_REPLICA, { className: PUSH_REPLAY_CLASS, scriptName: FIRST_REPLICA, useSQLite: true }),
+      ],
       handleRuntimeStdio: (stdout, stderr) => {
         collect(stdout);
         collect(stderr);
@@ -64,13 +85,18 @@ export class ManagedWorker {
     return new ManagedWorker(miniflare, bundleDir, runtimeLog);
   }
 
+  /** The other replica of the Worker. */
+  secondReplica(): ManagedWorker {
+    return new ManagedWorker(this.miniflare, this.bundleDir, this.runtimeLog, SECOND_REPLICA);
+  }
+
   /** What the Worker printed, for a failing test's message. */
   get log(): string {
     return this.runtimeLog.join('');
   }
 
   async fetch(url: string, init?: RequestInit): Promise<Response> {
-    const worker = await this.miniflare.getWorker();
+    const worker = await this.miniflare.getWorker(this.replicaName);
     return worker.fetch(url, init) as unknown as Promise<Response>;
   }
 
@@ -93,7 +119,7 @@ export class ManagedWorker {
 
   /** Run the Cron Trigger handler, as the platform does on schedule. */
   async scheduled(): Promise<{ outcome: string }> {
-    const worker = await this.miniflare.getWorker();
+    const worker = await this.miniflare.getWorker(this.replicaName);
     return worker.scheduled({ cron: '*/5 * * * *' });
   }
 
@@ -105,7 +131,7 @@ export class ManagedWorker {
 
   /** The version of the last good bundle the Worker kept in KV. */
   async cachedBundleVersion(): Promise<string | undefined> {
-    const namespace = await this.miniflare.getKVNamespace('BUNDLE_CACHE');
+    const namespace = await this.miniflare.getKVNamespace('BUNDLE_CACHE', this.replicaName);
     const cached = await namespace.get(BUNDLE_CACHE_KEY);
     return cached ? (JSON.parse(cached) as { version: string }).version : undefined;
   }

apps/e2e/demo-e2e-cloudflare/e2e/managed-edge-autoupdate.e2e.spec.ts

@@ -1,5 +1,5 @@
 // A managed @frontmcp/edge Worker in workerd: pull over https, Cron refresh (also as an isolate's first event), token
-// rejection, workflows and pushes, older ones refused.
+// rejection, workflows and pushes, older ones refused, and a push token replayed to another replica refused.
 import { base64urlEncode, randomBytes } from '@frontmcp/utils';
 
 import {
@@ -203,6 +203,39 @@ describe('a managed edge Worker on workerd', () => {
       expect(await servedSkills()).toEqual(['invoices']);
       expect(await worker.cachedBundleVersion()).toBe('5.0.0');
     });
+
+    it('is refused, with 409, when its token is replayed to another replica of the Worker', async () => {
+      const replica = worker.secondReplica();
+      const replicaSkills = async () => {
+        const result = await replica.callTool<SearchSkillResponse>('search_skill', { query: 'billing' });
+        return result
+          .json()
+          .skills.map((skill) => `${skill.skillId}@${skill.bundleVersion}`)
+          .sort();
+      };
+      publish('9.0.0', ['invoices', 'reports']);
+      const servedBeforeReplay = await replicaSkills();
+      const token = await issuer.pushToken();
+
+      const first = await worker.push(`Bearer ${token}`, signer.sign(billingBundle('10.0.0', ['refunds'])));
+      const replayed = await replica.push(`Bearer ${token}`, signer.sign(billingBundle('10.0.0', ['refunds'])));
+      const servedAfterReplay = await replicaSkills();
+      const fresh = await replica.push(
+        `Bearer ${await issuer.pushToken()}`,
+        signer.sign(billingBundle('11.0.0', ['refunds'])),
+      );
+
+      expect(servedBeforeReplay).toEqual(['invoices@9.0.0', 'reports@9.0.0']);
+      expect(first.status).toBe(200);
+      expect(replayed.status).toBe(409);
+      expect(await replayed.json()).toEqual({
+        error: 'push_replayed',
+        error_description: 'a push already used this token',
+      });
+      expect(servedAfterReplay).toEqual(servedBeforeReplay);
+      expect(fresh.status).toBe(200);
+      expect(await replicaSkills()).toEqual(['refunds@11.0.0']);
+    });
   });
 });
 

apps/e2e/demo-e2e-cloudflare/fixture-managed/worker.ts

@@ -13,7 +13,7 @@ interface ManagedFixtureBindings {
 
 const bindings = env as unknown as ManagedFixtureBindings;
 
-export default createEdgeMcp({
+const mcp = createEdgeMcp({
   info: { name: 'cf-managed-fixture', version: '1.0.0' },
   apps: [],
   http: { entryPath: '/mcp' },
@@ -30,3 +30,6 @@ export default createEdgeMcp({
     cache: kvBundleCacheFromEnv('BUNDLE_CACHE'),
   },
 });
+
+export default mcp;
+export const FrontMcpPushReplay = mcp.PushReplayDurableObject;

apps/e2e/demo-e2e-cloudflare/fixture-managed/wrangler.toml

@@ -7,5 +7,13 @@ compatibility_flags = ["nodejs_compat"]
 binding = "BUNDLE_CACHE"
 id = "e2e-bundle-cache"
 
+[[durable_objects.bindings]]
+name = "FRONTMCP_PUSH_REPLAY"
+class_name = "FrontMcpPushReplay"
+
+[[migrations]]
+tag = "v1"
+new_sqlite_classes = ["FrontMcpPushReplay"]
+
 [triggers]
 crons = ["*/5 * * * *"]

libs/adapters/src/skills/__tests__/replay-nonce-store.spec.ts

@@ -0,0 +1,88 @@
+import { createMemoryStorage, createStorage } from '@frontmcp/utils';
+
+import { createStorageReplayNonceStore, StorageReplayNonceStore } from '../security/replay-nonce-store';
+
+jest.mock('@frontmcp/utils', () => ({ ...jest.requireActual('@frontmcp/utils'), createStorage: jest.fn() }));
+
+const createStorageMock = jest.mocked(createStorage);
+
+async function connectedMemoryStorage() {
+  const storage = createMemoryStorage({ enableSweeper: false });
+  await storage.connect();
+  return storage;
+}
+
+describe('StorageReplayNonceStore', () => {
+  it('records a key once, with the TTL rounded up to whole seconds, and again once the TTL expired', async () => {
+    const storage = await connectedMemoryStorage();
+    const store = new StorageReplayNonceStore(storage);
+    const startedAt = Date.now();
+    const clock = jest.spyOn(Date, 'now').mockReturnValue(startedAt);
+
+    try {
+      const first = await store.claim('jti-hash', 1_500);
+      const ttlSeconds = await storage.ttl('jti-hash');
+      const replayed = await store.claim('jti-hash', 1_500);
+      clock.mockReturnValue(startedAt + 2_001);
+      const afterExpiry = await store.claim('jti-hash', 1_500);
+
+      expect([first, replayed, afterExpiry]).toEqual([true, false, true]);
+      expect(ttlSeconds).toBe(2);
+    } finally {
+      clock.mockRestore();
+    }
+  });
+
+  it('keeps a key for at least a second when the window ends now', async () => {
+    const storage = await connectedMemoryStorage();
+
+    await new StorageReplayNonceStore(storage).claim('at-the-boundary', 0);
+
+    expect(await storage.ttl('at-the-boundary')).toBe(1);
+  });
+
+  it('lets only one of two concurrent claims of a key win', async () => {
+    const shared = await connectedMemoryStorage();
+    const firstReplica = new StorageReplayNonceStore(shared);
+    const secondReplica = new StorageReplayNonceStore(shared);
+
+    const outcomes = await Promise.all([firstReplica.claim('raced', 60_000), secondReplica.claim('raced', 60_000)]);
+
+    expect(outcomes.sort()).toEqual([false, true]);
+  });
+});
+
+describe('createStorageReplayNonceStore', () => {
+  afterEach(() => {
+    createStorageMock.mockReset();
+  });
+
+  it('connects on the first claim, never falling back to memory, and keys under the namespace', async () => {
+    const storage = await connectedMemoryStorage();
+    createStorageMock.mockResolvedValue(storage);
+    const store = createStorageReplayNonceStore({ type: 'redis', redis: { url: 'redis://replay:6379' } }, 'jti:acme');
+
+    expect(createStorageMock).not.toHaveBeenCalled();
+    const first = await store.claim('nonce-hash', 60_000);
+    const replayed = await store.claim('nonce-hash', 60_000);
+
+    expect([first, replayed]).toEqual([true, false]);
+    expect(createStorageMock).toHaveBeenCalledTimes(1);
+    expect(createStorageMock).toHaveBeenCalledWith({
+      type: 'redis',
+      redis: { url: 'redis://replay:6379' },
+      fallback: 'error',
+    });
+    expect(await storage.keys('*')).toEqual(['jti:acme:nonce-hash']);
+  });
+
+  it('throws while the storage cannot connect, and connects again on the next claim', async () => {
+    const storage = await connectedMemoryStorage();
+    createStorageMock.mockRejectedValueOnce(new Error('ECONNREFUSED')).mockResolvedValue(storage);
+    const store = createStorageReplayNonceStore({ type: 'redis' }, 'jti:acme');
+
+    await expect(store.claim('nonce-hash', 60_000)).rejects.toThrow('ECONNREFUSED');
+    await expect(store.claim('nonce-hash', 60_000)).resolves.toBe(true);
+    expect(createStorageMock).toHaveBeenCalledTimes(2);
+  });
+});

libs/adapters/src/skills/__tests__/saas-pull.source.push.spec.ts

@@ -1,5 +1,8 @@
 // With enableWebhook, the saas source checks a pushed bundle's token as it checks the pull token, plus its iat and jti.
+import { createMemoryStorage, createStorage } from '@frontmcp/utils';
+
 import type { ResolvedBundle } from '../bundle/bundle.types';
+import { StorageReplayNonceStore, type ReplayNonceStore } from '../security/replay-nonce-store';
 import type { SaasSourceOptions } from '../source-options';
 import type { BundleCacheStore } from '../sources';
 import { SaasPullSource, SaasPushRefusedError } from '../sources/saas-pull.source';
@@ -11,6 +14,8 @@ import {
   type SaasTokenIssuer,
 } from './saas-token.fixture';
 
+jest.mock('@frontmcp/utils', () => ({ ...jest.requireActual('@frontmcp/utils'), createStorage: jest.fn() }));
+
 const logger = {
   warn: jest.fn(),
   info: jest.fn(),
@@ -27,8 +32,9 @@ class StubbedSaasSource extends SaasPullSource {
     options: SaasSourceOptions,
     private readonly stub: Stub,
     cache?: BundleCacheStore,
+    pushReplayStore?: ReplayNonceStore,
   ) {
-    super(options, undefined, logger, { cache, disablePolling: true });
+    super(options, undefined, logger, { cache, disablePolling: true, pushReplayStore });
   }
   protected override async httpGet(url: string) {
     return this.stub(url);
@@ -118,6 +124,75 @@ describe('SaasPullSource push checks', () => {
     expect(refused.message).toContain(reason);
   });
 
+  describe('with a replay store shared by several sources', () => {
+    const servingJwks: Stub = async (url) =>
+      url === SAAS_JWKS_URL ? { status: 200, body: JSON.stringify(issuer.jwks) } : { status: 404, body: '' };
+
+    const replicaSharing = (store: ReplayNonceStore) =>
+      new StubbedSaasSource(sourceOptions(true), servingJwks, undefined, store);
+
+    it('refuses, on a second replica, a jti a push to the first replica used', async () => {
+      const storage = createMemoryStorage({ enableSweeper: false });
+      await storage.connect();
+      const shared = new StorageReplayNonceStore(storage);
+      const token = await issuer.token({ jti: 'push-shared-1' });
+      await replicaSharing(shared).verifyPushToken(token);
+
+      const replayed = await replicaSharing(shared)
+        .verifyPushToken(token)
+        .catch((e: unknown) => e);
+
+      expect(replayed).toMatchObject({ refusal: 'replayed', message: expect.stringContaining('nonce replay') });
+    });
+
+    it('refuses the push, as unchecked, when the replay store fails', async () => {
+      const failing = { claim: jest.fn(async () => Promise.reject(new Error('ECONNREFUSED 10.0.0.7:6379'))) };
+
+      const error = await replicaSharing(failing)
+        .verifyPushToken(await issuer.token({ jti: 'push-shared-2' }))
+        .catch((e: unknown) => e);
+
+      expect(error).toBeInstanceOf(SaasPushRefusedError);
+      expect(error).toMatchObject({
+        refusal: 'replay-unchecked',
+        message: expect.stringContaining('the replay store failed: ECONNREFUSED'),
+      });
+    });
+
+    it('keeps the jtis in the storage pushReplayStorage names, under its audience, never falling back to memory', async () => {
+      const storage = createMemoryStorage({ enableSweeper: false });
+      await storage.connect();
+      jest.mocked(createStorage).mockResolvedValue(storage);
+      const pushReplayStorage = { type: 'redis' as const, redis: { url: 'redis://replay.internal:6379' } };
+      const replica = () => new StubbedSaasSource({ ...sourceOptions(true), pushReplayStorage }, servingJwks);
+      const token = await issuer.token({ jti: 'push-shared-3' });
+
+      await replica().verifyPushToken(token);
+      const replayed = await replica()
+        .verifyPushToken(token)
+        .catch((e: unknown) => e);
+
+      expect(replayed).toMatchObject({ refusal: 'replayed' });
+      expect(createStorage).toHaveBeenCalledWith({ ...pushReplayStorage, fallback: 'error' });
+      expect(await storage.keys('*')).toEqual([
+        expect.stringMatching(new RegExp(`^frontmcp:skilled-openapi:push-jti:${SAAS_AUDIENCE}:[0-9a-f]{64}$`)),
+      ]);
+    });
+
+    it('uses an injected pushReplayStore rather than the storage pushReplayStorage names', async () => {
+      const injected = { claim: jest.fn(async () => true) };
+      const options = { ...sourceOptions(true), pushReplayStorage: { type: 'redis' as const } };
+      jest.mocked(createStorage).mockClear();
+
+      await new StubbedSaasSource(options, servingJwks, undefined, injected).verifyPushToken(
+        await issuer.token({ jti: 'push-shared-4' }),
+      );
+
+      expect(injected.claim).toHaveBeenCalledTimes(1);
+      expect(createStorage).not.toHaveBeenCalled();
+    });
+  });
+
   it('refuses a token signed by a key the JWKS does not hold', async () => {
     const stranger = await createSaasTokenIssuer();