diff --git a/apps/e2e/demo-e2e-cloudflare/e2e/helpers/managed-worker.ts b/apps/e2e/demo-e2e-cloudflare/e2e/helpers/managed-worker.ts index 588e38c61..73275eb88 100644 --- a/apps/e2e/demo-e2e-cloudflare/e2e/helpers/managed-worker.ts +++ b/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,7 +33,12 @@ export interface ManagedWorkerOptions { outbound: (request: Request) => Promise; } -/** 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; @@ -37,6 +46,7 @@ export class ManagedWorker { private readonly miniflare: Miniflare, private readonly bundleDir: string, private readonly runtimeLog: string[], + private readonly replicaName = FIRST_REPLICA, ) {} static async start({ bindings, outbound }: ManagedWorkerOptions): Promise { @@ -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 { - const worker = await this.miniflare.getWorker(); + const worker = await this.miniflare.getWorker(this.replicaName); return worker.fetch(url, init) as unknown as Promise; } @@ -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 { - 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; } diff --git a/apps/e2e/demo-e2e-cloudflare/e2e/managed-edge-autoupdate.e2e.spec.ts b/apps/e2e/demo-e2e-cloudflare/e2e/managed-edge-autoupdate.e2e.spec.ts index d98909738..153f68650 100644 --- a/apps/e2e/demo-e2e-cloudflare/e2e/managed-edge-autoupdate.e2e.spec.ts +++ b/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('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']); + }); }); }); diff --git a/apps/e2e/demo-e2e-cloudflare/fixture-managed/worker.ts b/apps/e2e/demo-e2e-cloudflare/fixture-managed/worker.ts index 8962b2ae5..2717f0c9e 100644 --- a/apps/e2e/demo-e2e-cloudflare/fixture-managed/worker.ts +++ b/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; diff --git a/apps/e2e/demo-e2e-cloudflare/fixture-managed/wrangler.toml b/apps/e2e/demo-e2e-cloudflare/fixture-managed/wrangler.toml index 30c4f5d1b..177cd492c 100644 --- a/apps/e2e/demo-e2e-cloudflare/fixture-managed/wrangler.toml +++ b/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 * * * *"] diff --git a/libs/adapters/src/skills/__tests__/replay-nonce-store.spec.ts b/libs/adapters/src/skills/__tests__/replay-nonce-store.spec.ts new file mode 100644 index 000000000..941b257b0 --- /dev/null +++ b/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); + }); +}); diff --git a/libs/adapters/src/skills/__tests__/saas-pull.source.push.spec.ts b/libs/adapters/src/skills/__tests__/saas-pull.source.push.spec.ts index f754199d3..ae7b6eed7 100644 --- a/libs/adapters/src/skills/__tests__/saas-pull.source.push.spec.ts +++ b/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(); diff --git a/libs/adapters/src/skills/__tests__/source-options.spec.ts b/libs/adapters/src/skills/__tests__/source-options.spec.ts index e1b58ed48..a8e8168f8 100644 --- a/libs/adapters/src/skills/__tests__/source-options.spec.ts +++ b/libs/adapters/src/skills/__tests__/source-options.spec.ts @@ -58,6 +58,39 @@ describe('source-options schemas', () => { expect(() => saasSourceSchema.parse({ ...baseSaas, allowRollback: 'yes' })).toThrow(); }); + describe('pushReplayStorage', () => { + const issuesOf = (pushReplayStorage: unknown) => { + const result = saasSourceSchema.safeParse({ ...baseSaas, enableWebhook: true, pushReplayStorage }); + return result.success ? [] : result.error.issues.map((issue) => `${issue.path.join('.')}: ${issue.message}`); + }; + + it('is unset by default, and takes a store every replica shares', () => { + const redis = { type: 'redis', redis: { url: 'redis://replay.internal:6379' }, prefix: 'acme:' }; + + expect(saasSourceSchema.parse(baseSaas).pushReplayStorage).toBeUndefined(); + expect(saasSourceSchema.parse({ ...baseSaas, pushReplayStorage: redis }).pushReplayStorage).toEqual(redis); + expect(issuesOf({ type: 'upstash' })).toEqual([]); + expect(issuesOf({ type: 'vercel-kv' })).toEqual([]); + expect(issuesOf({ type: 'sqlite', sqlite: { path: '/var/lib/acme/replay.sqlite' } })).toEqual([]); + }); + + it.each([['memory'], ['auto'], ['cloudflare-kv'], [undefined]])( + 'refuses type %s, which replicas do not share or cannot write once atomically', + (type) => { + expect(issuesOf({ type })).toEqual([expect.stringMatching(/^pushReplayStorage\.type: must be "redis", /)]); + }, + ); + + it('refuses a fallback to memory, and a value that is not a storage config', () => { + expect(issuesOf({ type: 'redis', fallback: 'memory' })).toEqual([ + 'pushReplayStorage.fallback: cannot be "memory": each replica would then accept the same push', + ]); + expect(issuesOf('redis://replay.internal:6379')).toEqual([ + expect.stringMatching(/^pushReplayStorage: must be a storage config/), + ]); + }); + }); + it('rejects http:// endpoint (must be HTTPS)', () => { expect(() => saasSourceSchema.parse({ ...baseSaas, endpoint: 'http://insecure.example/x' })).toThrow(); }); diff --git a/libs/adapters/src/skills/__tests__/webhook-replay-guard.spec.ts b/libs/adapters/src/skills/__tests__/webhook-replay-guard.spec.ts index 4ca61d9b8..f8c371437 100644 --- a/libs/adapters/src/skills/__tests__/webhook-replay-guard.spec.ts +++ b/libs/adapters/src/skills/__tests__/webhook-replay-guard.spec.ts @@ -1,137 +1,109 @@ +import { createMemoryStorage } from '@frontmcp/utils'; + +import { MemoryReplayNonceStore, StorageReplayNonceStore } from '../security/replay-nonce-store'; import { WebhookReplayGuard } from '../security/webhook-replay-guard'; +async function connectedMemoryStorage() { + const storage = createMemoryStorage({ enableSweeper: false }); + await storage.connect(); + return storage; +} + describe('WebhookReplayGuard', () => { - it('accepts a fresh nonce inside the window', () => { + it('accepts a fresh nonce inside the window', async () => { const guard = new WebhookReplayGuard({ windowMs: 60_000 }); guard.setNowProvider(() => 1_000_000); - const r = guard.check({ timestampMs: 1_000_000, nonce: 'abc12345' }); + const r = await guard.check({ timestampMs: 1_000_000, nonce: 'abc12345' }); expect(r.ok).toBe(true); }); - it('rejects nonce replay within the window', () => { + it('rejects nonce replay within the window', async () => { const guard = new WebhookReplayGuard({ windowMs: 60_000 }); guard.setNowProvider(() => 1_000_000); - const ok1 = guard.check({ timestampMs: 1_000_000, nonce: 'replay123' }); - const ok2 = guard.check({ timestampMs: 1_000_000, nonce: 'replay123' }); + const ok1 = await guard.check({ timestampMs: 1_000_000, nonce: 'replay123' }); + const ok2 = await guard.check({ timestampMs: 1_000_000, nonce: 'replay123' }); expect(ok1.ok).toBe(true); expect(ok2.ok).toBe(false); expect(ok2.reason).toMatch(/replay/); }); - it('names each rejection with a code', () => { + it('names each rejection with a code', async () => { const guard = new WebhookReplayGuard({ windowMs: 1_000 }); guard.setNowProvider(() => 1_000_000); - guard.check({ timestampMs: 1_000_000, nonce: 'seen-nonce' }); + await guard.check({ timestampMs: 1_000_000, nonce: 'seen-nonce' }); const codes = [ - guard.check({ timestampMs: Number.NaN, nonce: 'nonce-0001' }).code, - guard.check({ timestampMs: 1_000_000, nonce: 'short' }).code, - guard.check({ timestampMs: 1_000_000 - 60_000, nonce: 'nonce-0002' }).code, - guard.check({ timestampMs: 1_000_000, nonce: 'seen-nonce' }).code, + (await guard.check({ timestampMs: Number.NaN, nonce: 'nonce-0001' })).code, + (await guard.check({ timestampMs: 1_000_000, nonce: 'short' })).code, + (await guard.check({ timestampMs: 1_000_000 - 60_000, nonce: 'nonce-0002' })).code, + (await guard.check({ timestampMs: 1_000_000, nonce: 'seen-nonce' })).code, ]; expect(codes).toEqual(['invalid_timestamp', 'invalid_nonce', 'outside_window', 'nonce_replay']); }); - it('rejects timestamps outside the window', () => { + it('rejects timestamps outside the window', async () => { const guard = new WebhookReplayGuard({ windowMs: 1_000 }); guard.setNowProvider(() => 1_000_000); - const r = guard.check({ timestampMs: 1_000_000 - 60_000, nonce: 'oldnonce1' }); + const r = await guard.check({ timestampMs: 1_000_000 - 60_000, nonce: 'oldnonce1' }); expect(r.ok).toBe(false); expect(r.reason).toMatch(/outside/); }); - it('rejects non-finite timestamps (NaN, +Inf, -Inf)', () => { + it('rejects non-finite timestamps (NaN, +Inf, -Inf)', async () => { const guard = new WebhookReplayGuard(); - expect(guard.check({ timestampMs: Number.NaN, nonce: 'abcdefgh' }).ok).toBe(false); - expect(guard.check({ timestampMs: Number.POSITIVE_INFINITY, nonce: 'abcdefgh' }).ok).toBe(false); - expect(guard.check({ timestampMs: Number.NEGATIVE_INFINITY, nonce: 'abcdefgh' }).ok).toBe(false); + expect((await guard.check({ timestampMs: Number.NaN, nonce: 'abcdefgh' })).ok).toBe(false); + expect((await guard.check({ timestampMs: Number.POSITIVE_INFINITY, nonce: 'abcdefgh' })).ok).toBe(false); + expect((await guard.check({ timestampMs: Number.NEGATIVE_INFINITY, nonce: 'abcdefgh' })).ok).toBe(false); }); - it('rejects too-short nonces', () => { + it('rejects too-short nonces', async () => { const guard = new WebhookReplayGuard(); guard.setNowProvider(() => 1); - const r = guard.check({ timestampMs: 1, nonce: 'tiny' }); + const r = await guard.check({ timestampMs: 1, nonce: 'tiny' }); expect(r.ok).toBe(false); }); - it('LRU drops the entry with the smallest expiresAt when capacity is reached', () => { - // All four entries share the same timestamp (and therefore expiresAt), so - // the eviction loop drops whichever was hit first by the iterator — - // 'aaaaaaaa'. The remaining three plus the newly-inserted 'dddddddd' - // saturate the 3-slot capacity. Verify the survivor set explicitly so the - // test fails if eviction order regresses, instead of only asserting size. - const guard = new WebhookReplayGuard({ capacity: 3 }); - const now = 1; - guard.setNowProvider(() => now); - guard.check({ timestampMs: now, nonce: 'aaaaaaaa' }); - guard.check({ timestampMs: now, nonce: 'bbbbbbbb' }); - guard.check({ timestampMs: now, nonce: 'cccccccc' }); - guard.check({ timestampMs: now, nonce: 'dddddddd' }); - expect(guard.size).toBe(3); - // 'aaaaaaaa' was evicted — re-using it now must succeed. - expect(guard.check({ timestampMs: now, nonce: 'aaaaaaaa' }).ok).toBe(true); - // The most-recent insert remains tracked — replay must be rejected. - // Re-checking 'aaaaaaaa' above re-inserted it (filling capacity again), - // so checking 'dddddddd' here may itself trigger another eviction; we - // just need to confirm 'dddddddd' is still treated as a known nonce. - const replay = guard.check({ timestampMs: now, nonce: 'dddddddd' }); - expect(replay.ok).toBe(false); - }); - - it('prune drops expired entries', () => { - const guard = new WebhookReplayGuard({ windowMs: 1_000 }); - guard.setNowProvider(() => 1_000); - guard.check({ timestampMs: 1_000, nonce: 'expirable' }); - guard.setNowProvider(() => 1_000_000); - expect(guard.prune()).toBe(1); - expect(guard.size).toBe(0); - }); - - it('default options work without overrides', () => { + it('default options work without overrides', async () => { const guard = new WebhookReplayGuard(); guard.setNowProvider(() => 1_000_000); - expect(guard.check({ timestampMs: 1_000_000, nonce: 'defaultok' }).ok).toBe(true); + expect((await guard.check({ timestampMs: 1_000_000, nonce: 'defaultok' })).ok).toBe(true); }); - it('size getter reflects current map state and prune does not over-evict', () => { - const guard = new WebhookReplayGuard({ windowMs: 10_000 }); - let now = 100; - guard.setNowProvider(() => now); - guard.check({ timestampMs: now, nonce: 'fresh111' }); - guard.check({ timestampMs: now, nonce: 'fresh222' }); - expect(guard.size).toBe(2); - // Bump time within the window — prune should not drop anything. - now = 105; - expect(guard.prune()).toBe(0); - expect(guard.size).toBe(2); - }); - - it('rejects nonce reuse even with a different in-window timestamp', () => { - // Class contract: "the same nonce can't be replayed within the window". - // Keying on (ts|nonce) would let an attacker reuse the nonce by varying - // the timestamp — fix is to dedupe by nonce alone. + it('rejects nonce reuse even with a different in-window timestamp', async () => { const guard = new WebhookReplayGuard({ windowMs: 60_000 }); guard.setNowProvider(() => 1_000); - expect(guard.check({ timestampMs: 1_000, nonce: 'aaaa1111' }).ok).toBe(true); - expect(guard.check({ timestampMs: 1_001, nonce: 'aaaa1111' }).ok).toBe(false); + expect((await guard.check({ timestampMs: 1_000, nonce: 'aaaa1111' })).ok).toBe(true); + expect((await guard.check({ timestampMs: 1_001, nonce: 'aaaa1111' })).ok).toBe(false); }); - it('rejects nonce replay at the freshness-window boundary', () => { - // Regression: a nonce whose expiresAt equals `now` must still be treated - // as tracked, otherwise an attacker can replay it at the exact boundary. + it('rejects nonce replay at the freshness-window boundary', async () => { const guard = new WebhookReplayGuard({ windowMs: 1_000 }); let now = 1_000; guard.setNowProvider(() => now); - expect(guard.check({ timestampMs: 1_000, nonce: 'boundary' }).ok).toBe(true); - // Advance to exactly expiresAt (1_000 + 1_000 = 2_000). The nonce must - // still be considered tracked at this tick. + expect((await guard.check({ timestampMs: 1_000, nonce: 'boundary' })).ok).toBe(true); now = 2_000; - const r = guard.check({ timestampMs: 2_000, nonce: 'boundary' }); + const r = await guard.check({ timestampMs: 2_000, nonce: 'boundary' }); expect(r.ok).toBe(false); expect(r.reason).toMatch(/replay/); }); + it('keeps a nonce until its timestamp leaves the window, so a token issued ahead of the clock cannot be replayed', async () => { + const claims: Array<{ key: string; ttlMs: number }> = []; + const store = { + claim: jest.fn(async (key: string, ttlMs: number) => { + claims.push({ key, ttlMs }); + return true; + }), + }; + const guard = new WebhookReplayGuard({ windowMs: 60_000, store }); + guard.setNowProvider(() => 100_000); + + await guard.check({ timestampMs: 130_000, nonce: 'ahead-of-clock' }); + + expect(claims).toEqual([{ key: expect.stringMatching(/^[0-9a-f]{64}$/), ttlMs: 90_000 }]); + }); + it('constructor rejects invalid windowMs and capacity', () => { expect(() => new WebhookReplayGuard({ windowMs: 0 })).toThrow(RangeError); expect(() => new WebhookReplayGuard({ windowMs: -1 })).toThrow(RangeError); @@ -141,6 +113,46 @@ describe('WebhookReplayGuard', () => { expect(() => new WebhookReplayGuard({ capacity: -3 })).toThrow(RangeError); }); + describe('over a shared store', () => { + it('refuses, in a second guard sharing the store, a nonce the first one accepted', async () => { + const shared = new StorageReplayNonceStore(await connectedMemoryStorage()); + const firstReplica = new WebhookReplayGuard({ store: shared }); + const secondReplica = new WebhookReplayGuard({ store: shared }); + + const first = await firstReplica.check({ timestampMs: Date.now(), nonce: 'shared-nonce' }); + const replayed = await secondReplica.check({ timestampMs: Date.now(), nonce: 'shared-nonce' }); + + expect(first.ok).toBe(true); + expect(replayed).toMatchObject({ ok: false, code: 'nonce_replay' }); + }); + + it('accepts in each guard the nonces it alone saw, when the guards keep them in memory', async () => { + const firstReplica = new WebhookReplayGuard(); + const secondReplica = new WebhookReplayGuard(); + + await firstReplica.check({ timestampMs: Date.now(), nonce: 'per-process' }); + + expect((await secondReplica.check({ timestampMs: Date.now(), nonce: 'per-process' })).ok).toBe(true); + }); + + it('refuses the nonce when the store fails, and accepts it once the store is back', async () => { + const storage = await connectedMemoryStorage(); + const failingSet = jest.spyOn(storage, 'set').mockRejectedValueOnce(new Error('connection refused')); + const guard = new WebhookReplayGuard({ store: new StorageReplayNonceStore(storage) }); + + const whileDown = await guard.check({ timestampMs: Date.now(), nonce: 'store-down' }); + const onceBack = await guard.check({ timestampMs: Date.now(), nonce: 'store-down' }); + + expect(whileDown).toEqual({ + ok: false, + code: 'store_unavailable', + reason: 'the replay store failed: connection refused', + }); + expect(onceBack.ok).toBe(true); + failingSet.mockRestore(); + }); + }); + describe('telemetry', () => { function makeFakeTelemetry(): { telemetry: { @@ -163,11 +175,11 @@ describe('WebhookReplayGuard', () => { }; } - it('records checks_total{status=ok} on accepted nonce and skips the rejects counter', () => { + it('records checks_total{status=ok} on accepted nonce and skips the rejects counter', async () => { const { telemetry, counterCalls } = makeFakeTelemetry(); const guard = new WebhookReplayGuard({ windowMs: 60_000, telemetry }); guard.setNowProvider(() => 1_000_000); - guard.check({ timestampMs: 1_000_000, nonce: 'abcdefgh' }); + await guard.check({ timestampMs: 1_000_000, nonce: 'abcdefgh' }); const rejects = counterCalls.filter((c) => c.name === 'frontmcp_skills_replay_rejects_total'); const checks = counterCalls.filter((c) => c.name === 'frontmcp_skills_replay_checks_total'); expect(rejects).toHaveLength(0); @@ -175,64 +187,94 @@ describe('WebhookReplayGuard', () => { expect(checks[0].attributes['status']).toBe('ok'); }); - it('increments with reason=invalid_timestamp on non-finite timestamp', () => { - const { telemetry, counterCalls } = makeFakeTelemetry(); - const guard = new WebhookReplayGuard({ telemetry }); - guard.check({ timestampMs: Number.NaN, nonce: 'abcdefgh' }); - expect(counterCalls[0].name).toBe('frontmcp_skills_replay_rejects_total'); - expect(counterCalls[0].attributes['reason']).toBe('invalid_timestamp'); - }); - - it('increments with reason=invalid_nonce on too-short nonce', () => { - const { telemetry, counterCalls } = makeFakeTelemetry(); - const guard = new WebhookReplayGuard({ telemetry }); - guard.setNowProvider(() => 1); - guard.check({ timestampMs: 1, nonce: 'tiny' }); - expect(counterCalls[0].attributes['reason']).toBe('invalid_nonce'); - }); - - it('increments with reason=outside_window on stale timestamp', () => { + it.each([ + ['invalid_timestamp', { timestampMs: Number.NaN, nonce: 'abcdefgh' }], + ['invalid_nonce', { timestampMs: 1_000_000, nonce: 'tiny' }], + ['outside_window', { timestampMs: 1_000, nonce: 'abcdefgh' }], + ])('increments with reason=%s', async (reason, check) => { const { telemetry, counterCalls } = makeFakeTelemetry(); const guard = new WebhookReplayGuard({ windowMs: 1_000, telemetry }); guard.setNowProvider(() => 1_000_000); - guard.check({ timestampMs: 1_000, nonce: 'abcdefgh' }); - expect(counterCalls[0].attributes['reason']).toBe('outside_window'); + await guard.check(check); + expect(counterCalls[0].name).toBe('frontmcp_skills_replay_rejects_total'); + expect(counterCalls[0].attributes['reason']).toBe(reason); }); - it('increments with reason=nonce_replay when nonce is reused', () => { + it('increments with reason=nonce_replay when nonce is reused', async () => { const { telemetry, counterCalls } = makeFakeTelemetry(); const guard = new WebhookReplayGuard({ windowMs: 60_000, telemetry }); guard.setNowProvider(() => 1_000_000); - guard.check({ timestampMs: 1_000_000, nonce: 'replayed' }); - guard.check({ timestampMs: 1_000_000, nonce: 'replayed' }); + await guard.check({ timestampMs: 1_000_000, nonce: 'replayed' }); + await guard.check({ timestampMs: 1_000_000, nonce: 'replayed' }); const replayCalls = counterCalls.filter((c) => c.attributes['reason'] === 'nonce_replay'); expect(replayCalls).toHaveLength(1); }); - it('omitting telemetry keeps the guard dependency-free', () => { + it('increments with reason=store_unavailable when the store fails', async () => { + const { telemetry, counterCalls } = makeFakeTelemetry(); + const store = { claim: jest.fn(async () => Promise.reject(new Error('down'))) }; + const guard = new WebhookReplayGuard({ telemetry, store }); + await guard.check({ timestampMs: Date.now(), nonce: 'abcdefgh' }); + expect(counterCalls[0].attributes['reason']).toBe('store_unavailable'); + }); + + it('omitting telemetry keeps the guard dependency-free', async () => { const guard = new WebhookReplayGuard(); guard.setNowProvider(() => 1); - expect(() => guard.check({ timestampMs: Number.NaN, nonce: 'aaaaaaaa' })).not.toThrow(); + await expect(guard.check({ timestampMs: Number.NaN, nonce: 'aaaaaaaa' })).resolves.toMatchObject({ ok: false }); + }); + }); + + describe('its default in-memory store', () => { + it('drops the entry with the smallest expiry when capacity is reached', async () => { + const guard = new WebhookReplayGuard({ capacity: 3 }); + guard.setNowProvider(() => 1); + for (const nonce of ['aaaaaaaa', 'bbbbbbbb', 'cccccccc', 'dddddddd']) { + await guard.check({ timestampMs: 1, nonce }); + } + expect((await guard.check({ timestampMs: 1, nonce: 'aaaaaaaa' })).ok).toBe(true); + expect((await guard.check({ timestampMs: 1, nonce: 'dddddddd' })).ok).toBe(false); + }); + + it('drops the entry with the smallest expiry, not the oldest insertion', async () => { + const guard = new WebhookReplayGuard({ capacity: 2, windowMs: 10_000 }); + guard.setNowProvider(() => 5_000); + await guard.check({ timestampMs: 1_000, nonce: 'olderexp' }); + await guard.check({ timestampMs: 5_000, nonce: 'newerexp' }); + await guard.check({ timestampMs: 5_000, nonce: 'thirdval' }); + expect((await guard.check({ timestampMs: 5_500, nonce: 'newerexp' })).ok).toBe(false); + expect((await guard.check({ timestampMs: 1_500, nonce: 'olderexp' })).ok).toBe(true); }); }); +}); + +describe('MemoryReplayNonceStore', () => { + it('prune drops expired entries only', async () => { + let now = 1_000; + const store = new MemoryReplayNonceStore({ now: () => now }); + await store.claim('expiring', 1_000); + await store.claim('lasting', 10_000); + + now = 1_500; + expect(store.prune()).toBe(0); + now = 2_001; + expect(store.prune()).toBe(1); + expect(store.size).toBe(1); + }); + + it('records a key again once it expired', async () => { + let now = 0; + const store = new MemoryReplayNonceStore({ now: () => now }); + + const first = await store.claim('key', 1_000); + const replayed = await store.claim('key', 1_000); + now = 1_001; + const afterExpiry = await store.claim('key', 1_000); + + expect([first, replayed, afterExpiry]).toEqual([true, false, true]); + }); - it('eviction drops the entry with the smallest expiresAt, not the oldest insertion', () => { - // Mix in a stale-but-in-window timestamp so insertion order != expiry order. - const guard = new WebhookReplayGuard({ capacity: 2, windowMs: 10_000 }); - guard.setNowProvider(() => 5_000); - // Insert an entry with the EARLIEST expiry (timestampMs much older but - // still within window — expiresAt = 1_000 + 10_000 = 11_000). - guard.check({ timestampMs: 1_000, nonce: 'olderexp' }); - // Insert a fresher entry (expiresAt = 5_000 + 10_000 = 15_000). - guard.check({ timestampMs: 5_000, nonce: 'newerexp' }); - // Hit capacity — should evict the entry with the smallest expiresAt, - // i.e. 'olderexp', leaving 'newerexp' alive. - guard.check({ timestampMs: 5_000, nonce: 'thirdval' }); - // Verify the survivor first: 'newerexp' is still tracked, so re-use is - // rejected. (Doing this before re-checking 'olderexp' avoids re-triggering - // capacity eviction, which would then drop one of the two survivors.) - expect(guard.check({ timestampMs: 5_500, nonce: 'newerexp' }).ok).toBe(false); - // 'olderexp' was evicted, so re-using it now should succeed. - expect(guard.check({ timestampMs: 1_500, nonce: 'olderexp' }).ok).toBe(true); + it('rejects an invalid capacity', () => { + expect(() => new MemoryReplayNonceStore({ capacity: 0 })).toThrow(RangeError); }); }); diff --git a/libs/adapters/src/skills/index.ts b/libs/adapters/src/skills/index.ts index 5a1052448..b17c71bf6 100644 --- a/libs/adapters/src/skills/index.ts +++ b/libs/adapters/src/skills/index.ts @@ -102,6 +102,13 @@ export { type SignatureVerifyTelemetry, } from './security/bundle-signature'; export { BundlePushJwtVerifier, type BundlePushJwtVerifierOptions } from './security/jwt-verifier'; +export { + createStorageReplayNonceStore, + MemoryReplayNonceStore, + StorageReplayNonceStore, + type MemoryReplayNonceStoreOptions, + type ReplayNonceStore, +} from './security/replay-nonce-store'; export { WebhookReplayGuard, type ReplayCheckResult, diff --git a/libs/adapters/src/skills/security/replay-nonce-store.ts b/libs/adapters/src/skills/security/replay-nonce-store.ts new file mode 100644 index 000000000..341750262 --- /dev/null +++ b/libs/adapters/src/skills/security/replay-nonce-store.ts @@ -0,0 +1,124 @@ +// file: libs/adapters/src/skills/security/replay-nonce-store.ts +// +// Where the push replay guard records the nonces (push token `jti`s) it accepted. + +import { createStorage, randomUUID, type StorageAdapter, type StorageConfig } from '@frontmcp/utils'; + +/** + * Records each nonce the replay guard accepts. Every instance that serves the server must share one store, or a push + * replayed to another replica or Worker isolate is accepted again. + */ +export interface ReplayNonceStore { + /** + * Record `key` as used for `ttlMs`. Resolves true when this call recorded it, false when it is already recorded and + * not expired. Throws when the store can't be read or written: the guard then refuses the push. + */ + claim(key: string, ttlMs: number): Promise; +} + +export interface MemoryReplayNonceStoreOptions { + /** Max nonces tracked at once (default 5_000); at capacity, the one that expires first is dropped. */ + capacity?: number; + /** The time source, for tests. */ + now?: () => number; +} + +const DEFAULT_CAPACITY = 5_000; + +/** A {@link ReplayNonceStore} in this process's memory: enough for one process, never for replicas or isolates. */ +export class MemoryReplayNonceStore implements ReplayNonceStore { + private readonly capacity: number; + private readonly now: () => number; + private readonly expiries = new Map(); + + constructor(options: MemoryReplayNonceStoreOptions = {}) { + const capacity = options.capacity ?? DEFAULT_CAPACITY; + if (!Number.isInteger(capacity) || capacity <= 0) { + throw new RangeError('MemoryReplayNonceStore: capacity must be a positive integer'); + } + this.capacity = capacity; + this.now = options.now ?? Date.now; + } + + async claim(key: string, ttlMs: number): Promise { + const now = this.now(); + const expiresAt = this.expiries.get(key); + // A nonce whose expiry is now is still tracked, so it can't be replayed at the window's last tick. + if (expiresAt !== undefined && expiresAt >= now) return false; + if (this.expiries.size >= this.capacity) this.prune(); + if (this.expiries.size >= this.capacity) this.evictFirstToExpire(); + this.expiries.set(key, now + ttlMs); + return true; + } + + /** Drop the expired nonces; returns how many were dropped. */ + prune(): number { + const now = this.now(); + let removed = 0; + for (const [key, expiresAt] of this.expiries) { + if (expiresAt < now) { + this.expiries.delete(key); + removed++; + } + } + return removed; + } + + /** How many nonces are tracked. */ + get size(): number { + return this.expiries.size; + } + + /** Insertion order says nothing about expiry (a token's `iat` can be older or newer than now), so search. */ + private evictFirstToExpire(): void { + let victimKey: string | undefined; + let victimExpiry = Number.POSITIVE_INFINITY; + for (const [candidateKey, candidateExpiry] of this.expiries) { + if (candidateExpiry < victimExpiry) { + victimKey = candidateKey; + victimExpiry = candidateExpiry; + } + } + if (victimKey !== undefined) this.expiries.delete(victimKey); + } +} + +/** + * A {@link ReplayNonceStore} on a `@frontmcp/utils` storage that every replica reaches (Redis, Upstash, Vercel KV, + * SQLite). A nonce is written once, with the TTL, only when absent; the write that stored its own claim id won. + */ +export class StorageReplayNonceStore implements ReplayNonceStore { + constructor(private readonly storage: StorageAdapter) {} + + async claim(key: string, ttlMs: number): Promise { + const claimId = randomUUID(); + const ttlSeconds = Math.max(1, Math.ceil(ttlMs / 1000)); + await this.storage.set(key, claimId, { ttlSeconds, ifNotExists: true }); + return (await this.storage.get(key)) === claimId; + } +} + +/** + * A {@link StorageReplayNonceStore} on the storage `config` describes, under `namespace`. It connects on the first + * claim, and again on the next one after a failed connection; it never falls back to memory. + */ +export function createStorageReplayNonceStore(config: StorageConfig, namespace: string): ReplayNonceStore { + let connecting: Promise | undefined; + const connect = async (): Promise => { + const root = await createStorage({ ...config, fallback: 'error' }); + return new StorageReplayNonceStore(root.namespace(namespace)); + }; + return { + async claim(key: string, ttlMs: number): Promise { + connecting ??= connect(); + let store: ReplayNonceStore; + try { + store = await connecting; + } catch (error) { + connecting = undefined; + throw error; + } + return store.claim(key, ttlMs); + }, + }; +} diff --git a/libs/adapters/src/skills/security/webhook-replay-guard.ts b/libs/adapters/src/skills/security/webhook-replay-guard.ts index a6ca84cfe..ca73156aa 100644 --- a/libs/adapters/src/skills/security/webhook-replay-guard.ts +++ b/libs/adapters/src/skills/security/webhook-replay-guard.ts @@ -4,8 +4,15 @@ import { sha256Hex } from '@frontmcp/utils'; +import { MemoryReplayNonceStore, type ReplayNonceStore } from './replay-nonce-store'; + /** Why a check refused: also the `reason` label of `frontmcp_skills_replay_rejects_total`. */ -export type ReplayRejectCode = 'invalid_timestamp' | 'invalid_nonce' | 'outside_window' | 'nonce_replay'; +export type ReplayRejectCode = + | 'invalid_timestamp' + | 'invalid_nonce' + | 'outside_window' + | 'nonce_replay' + | 'store_unavailable'; export interface ReplayCheckResult { ok: boolean; @@ -28,8 +35,13 @@ export interface ReplayGuardTelemetry { export interface WebhookReplayGuardOptions { /** Allowed clock skew window in ms (default 300_000 = 5min). */ windowMs?: number; - /** Max nonces tracked simultaneously (default 5_000). */ + /** Max nonces the default in-memory store tracks simultaneously (default 5_000). */ capacity?: number; + /** + * Where the accepted nonces are recorded. Default: this process's memory, which replicas and Worker isolates don't + * share. Give every instance that serves the server the same store. + */ + store?: ReplayNonceStore; /** * Optional telemetry hook. When provided, every `check()` rejection is * recorded against `frontmcp_skills_replay_rejects_total` with a low-cardinality @@ -39,27 +51,24 @@ export interface WebhookReplayGuardOptions { } const DEFAULT_WINDOW_MS = 5 * 60 * 1000; -const DEFAULT_CAPACITY = 5_000; export class WebhookReplayGuard { private readonly windowMs: number; - private readonly capacity: number; - private readonly seen = new Map(); + private readonly store: ReplayNonceStore; private now: () => number = Date.now; private readonly rejectsCounter: ReplayGuardCounter | undefined; private readonly checksCounter: ReplayGuardCounter | undefined; constructor(options: WebhookReplayGuardOptions = {}) { const windowMs = options.windowMs ?? DEFAULT_WINDOW_MS; - const capacity = options.capacity ?? DEFAULT_CAPACITY; if (!Number.isFinite(windowMs) || windowMs <= 0) { throw new RangeError('WebhookReplayGuard: windowMs must be a finite number > 0'); } - if (!Number.isInteger(capacity) || capacity <= 0) { + if (options.capacity !== undefined && (!Number.isInteger(options.capacity) || options.capacity <= 0)) { throw new RangeError('WebhookReplayGuard: capacity must be a positive integer'); } this.windowMs = windowMs; - this.capacity = capacity; + this.store = options.store ?? new MemoryReplayNonceStore({ capacity: options.capacity, now: () => this.now() }); // Two counters mirror the bundle-pulls / signature-verifications // pattern: the failure counter has a `reason` label; a `_checks_total` // counter tracks every call so operators can compute reject-rate from @@ -80,11 +89,10 @@ export class WebhookReplayGuard { } /** - * Check that a (timestamp, nonce) pair is fresh and unseen. Returns ok=false - * if the timestamp is outside the freshness window or the nonce has already - * been observed within that window. + * Check that a (timestamp, nonce) pair is fresh and unseen, and record the nonce. Returns ok=false if the timestamp + * is outside the freshness window, the nonce was already recorded within that window, or the store failed. */ - check(args: { timestampMs: number; nonce: string }): ReplayCheckResult { + async check(args: { timestampMs: number; nonce: string }): Promise { const { timestampMs, nonce } = args; const reject = (code: ReplayRejectCode, reason: string): ReplayCheckResult => { this.rejectsCounter?.inc(1, { reason: code }); @@ -103,60 +111,18 @@ export class WebhookReplayGuard { return reject('outside_window', `timestamp outside ${this.windowMs}ms window`); } - // Dedupe by nonce, NOT by (timestamp, nonce). The class contract is "the - // same nonce can't be replayed within the window"; keying on the tuple - // would let an attacker reuse a nonce by varying the in-window timestamp - // header (relevant when timestamps aren't independently bound by the - // outer JWT signature). - const key = sha256Hex(nonce); - const expiresAt = this.seen.get(key); - // Boundary handling: a nonce is considered fresh while expiresAt >= now. - // Using strict `>` here paired with `<=` in prune() left a one-tick window - // (abs(now - timestampMs) === windowMs) where a nonce could be accepted - // and then immediately replayed. Treat the boundary as still-tracked. - if (expiresAt !== undefined && expiresAt >= now) { - return reject('nonce_replay', 'nonce replay detected within freshness window'); + // Dedupe by nonce, NOT by (timestamp, nonce): keying on the tuple would let an attacker reuse a nonce by varying + // the in-window timestamp. The nonce stays recorded until its timestamp leaves the window. + let claimed: boolean; + try { + claimed = await this.store.claim(sha256Hex(nonce), timestampMs + this.windowMs - now); + } catch (error) { + return reject('store_unavailable', `the replay store failed: ${(error as Error).message}`); } - - // Capacity eviction: drop expired entries first; if still at capacity, - // evict the entry with the SMALLEST `expiresAt` — Map insertion order - // does not reliably correlate with expiry because the supplied - // `timestampMs` can be older or newer than `now`, so insertion-order - // eviction can drop a still-fresh nonce while leaving expired ones live. - if (this.seen.size >= this.capacity) { - this.prune(); - } - if (this.seen.size >= this.capacity) { - let victimKey: string | undefined; - let victimExpiry = Number.POSITIVE_INFINITY; - for (const [candidateKey, candidateExpiry] of this.seen) { - if (candidateExpiry < victimExpiry) { - victimKey = candidateKey; - victimExpiry = candidateExpiry; - } - } - if (victimKey !== undefined) this.seen.delete(victimKey); + if (!claimed) { + return reject('nonce_replay', 'nonce replay detected within freshness window'); } - this.seen.set(key, timestampMs + this.windowMs); this.checksCounter?.inc(1, { status: 'ok' }); return { ok: true }; } - - /** Test/maintenance helper: drop expired entries proactively. */ - prune(): number { - const now = this.now(); - let removed = 0; - for (const [k, expiresAt] of this.seen) { - if (expiresAt < now) { - this.seen.delete(k); - removed++; - } - } - return removed; - } - - /** Inspect: number of nonces currently tracked. */ - get size(): number { - return this.seen.size; - } } diff --git a/libs/adapters/src/skills/source-options.ts b/libs/adapters/src/skills/source-options.ts index ea73112b8..e7493a283 100644 --- a/libs/adapters/src/skills/source-options.ts +++ b/libs/adapters/src/skills/source-options.ts @@ -5,6 +5,7 @@ // options (outbound/SSRF, credentials) re-import what they need from here. import { z } from '@frontmcp/lazy-zod'; +import type { StorageConfig, StorageType } from '@frontmcp/utils'; // SaaS endpoints carry pinned bearer tokens and JWKS material; both must be // HTTPS so the JWT and verification keys are not exposed in plaintext. @@ -22,6 +23,30 @@ const httpsUrl = z { message: 'must use https://' }, ); +/** The storage types every replica reaches and can write a key to once, atomically: a replay store needs both. */ +const SHARED_REPLAY_STORAGE_TYPES: readonly StorageType[] = ['redis', 'upstash', 'vercel-kv', 'sqlite']; + +const pushReplayStorageSchema = z + .custom((value) => typeof value === 'object' && value !== null && !Array.isArray(value), { + message: 'must be a storage config, such as { type: "redis", redis: { url } }', + }) + .superRefine((storage, ctx) => { + if (!storage.type || !SHARED_REPLAY_STORAGE_TYPES.includes(storage.type)) { + ctx.addIssue({ + code: 'custom', + path: ['type'], + message: `must be ${SHARED_REPLAY_STORAGE_TYPES.map((type) => `"${type}"`).join(', ')}: a store every replica shares, with an atomic write-once`, + }); + } + if (storage.fallback === 'memory') { + ctx.addIssue({ + code: 'custom', + path: ['fallback'], + message: 'cannot be "memory": each replica would then accept the same push', + }); + } + }); + export const staticSourceSchema = z.object({ type: z.literal('static'), /** Path to the bundle directory (containing spec.yaml + overlay.yaml) or a single overlay file. */ @@ -77,6 +102,12 @@ export const saasSourceSchema = z.object({ * a bundle is refused, so a validly signed old bundle can't roll the server back. */ allowRollback: z.boolean().optional(), + /** + * Where the replay guard records the push tokens' `jti`s, through `createStorage()` from `@frontmcp/utils`: a store + * every replica shares (`redis`, `upstash`, `vercel-kv`, or `sqlite` for processes on one machine). Default: this + * process's memory, enough for a single process. When the store can't be reached, pushes are refused. + */ + pushReplayStorage: pushReplayStorageSchema.optional(), /** JWKS URL for verifying SaaS-issued tokens (fetched without credentials, redirects refused). */ jwksUrl: httpsUrl, /** Expected issuer (`iss`) claim of `authToken`. */ diff --git a/libs/adapters/src/skills/sources/saas-pull.source.ts b/libs/adapters/src/skills/sources/saas-pull.source.ts index e89cb5cad..de7ee6314 100644 --- a/libs/adapters/src/skills/sources/saas-pull.source.ts +++ b/libs/adapters/src/skills/sources/saas-pull.source.ts @@ -9,7 +9,8 @@ import { dirname, ensureDir, isRedirectResponse, pathResolve, readFile, writeFil import type { ResolvedBundle } from '../bundle/bundle.types'; import { OverlayParseError, parseOverlay } from '../bundle/overlay-parser'; -import { WebhookReplayGuard } from '../security/webhook-replay-guard'; +import { createStorageReplayNonceStore, type ReplayNonceStore } from '../security/replay-nonce-store'; +import { WebhookReplayGuard, type ReplayRejectCode } from '../security/webhook-replay-guard'; import type { SaasSourceOptions } from '../source-options'; import type { BundlePushReceiver, @@ -38,12 +39,13 @@ export class SaasPullTokenRejectedError extends Error { } /** - * The token of a pushed bundle was refused: it fails the checks the pull token passes (`invalid-token`), or a push - * already used it (`replayed`). The pushed bundle is not applied. + * The token of a pushed bundle was refused: it fails the checks the pull token passes (`invalid-token`), a push + * already used it (`replayed`), or the replay store could not record its `jti` (`replay-unchecked`). The pushed + * bundle is not applied. */ export class SaasPushRefusedError extends Error { constructor( - readonly refusal: 'invalid-token' | 'replayed', + readonly refusal: 'invalid-token' | 'replayed' | 'replay-unchecked', reason: string, ) { super(`[saas-source] push refused: ${reason}`); @@ -53,6 +55,15 @@ export class SaasPushRefusedError extends Error { type TokenVerdict = { ok: true; claims: Record } | { ok: false; reason: string }; +/** The key prefix of the push token `jti`s a shared replay store keeps, followed by the source's audience. */ +const PUSH_JTI_NAMESPACE = 'frontmcp:skilled-openapi:push-jti'; + +function pushRefusalFor(code: ReplayRejectCode | undefined): SaasPushRefusedError['refusal'] { + if (code === 'nonce_replay') return 'replayed'; + if (code === 'store_unavailable') return 'replay-unchecked'; + return 'invalid-token'; +} + /** Why a pull failed, with the first issues the bundle checks found when the bundle was the problem. */ function describePullFailure(error: unknown): string { const message = error instanceof Error ? error.message : String(error); @@ -136,7 +147,15 @@ export class SaasPullSource implements SkillBundleSource, BundlePushReceiver { private readonly deps: BundleSourceDeps = {}, ) { this.id = `saas:${this.options.endpoint}`; - if (options.enableWebhook) this.replayGuard = new WebhookReplayGuard(); + if (options.enableWebhook) this.replayGuard = new WebhookReplayGuard({ store: this.pushReplayStore() }); + } + + /** The store an injected `pushReplayStore` or the `pushReplayStorage` option names; none keeps the jtis in memory. */ + private pushReplayStore(): ReplayNonceStore | undefined { + if (this.deps.pushReplayStore) return this.deps.pushReplayStore; + const { pushReplayStorage, expectedAudience } = this.options; + if (!pushReplayStorage) return undefined; + return createStorageReplayNonceStore(pushReplayStorage, `${PUSH_JTI_NAMESPACE}:${expectedAudience}`); } /** This source, when `enableWebhook` is on: it then takes bundles pushed to the server. */ @@ -151,8 +170,9 @@ export class SaasPullSource implements SkillBundleSource, BundlePushReceiver { /** * Check the token of a pushed bundle: by the rules the pull token passes (a key in `jwksUrl`, `iss`, `aud`, - * `resource`, expiry), then by the replay guard, so its `iat` must be within five minutes and its `jti` unused. - * Throws {@link SaasPushRefusedError}, or a plain Error when the JWKS can't be read. + * `resource`, expiry), then by the replay guard, so its `iat` must be within five minutes and its `jti` unused by + * every instance sharing the replay store. Throws {@link SaasPushRefusedError}, or a plain Error when the JWKS can't + * be read. */ async verifyPushToken(token: string): Promise { if (!this.replayGuard) { @@ -167,10 +187,9 @@ export class SaasPullSource implements SkillBundleSource, BundlePushReceiver { if (typeof iat !== 'number' || typeof jti !== 'string') { throw new SaasPushRefusedError('invalid-token', 'a push token needs "iat" and "jti" claims'); } - const replay = this.replayGuard.check({ timestampMs: iat * 1000, nonce: jti }); + const replay = await this.replayGuard.check({ timestampMs: iat * 1000, nonce: jti }); if (!replay.ok) { - const refusal = replay.code === 'nonce_replay' ? 'replayed' : 'invalid-token'; - throw new SaasPushRefusedError(refusal, `its "jti" or "iat" claim: ${replay.reason}`); + throw new SaasPushRefusedError(pushRefusalFor(replay.code), `its "jti" or "iat" claim: ${replay.reason}`); } } diff --git a/libs/adapters/src/skills/sources/skill-bundle-source.interface.ts b/libs/adapters/src/skills/sources/skill-bundle-source.interface.ts index 67027e668..6647aca7f 100644 --- a/libs/adapters/src/skills/sources/skill-bundle-source.interface.ts +++ b/libs/adapters/src/skills/sources/skill-bundle-source.interface.ts @@ -1,6 +1,7 @@ // file: plugins/plugin-skilled-openapi/src/sources/skill-bundle-source.interface.ts import type { ResolvedBundle } from '../bundle/bundle.types'; +import type { ReplayNonceStore } from '../security/replay-nonce-store'; /** * Pluggable source of skill bundles. Static, npm-installed, and SaaS-pull @@ -95,7 +96,9 @@ export interface BundleCacheStore { * - `cache` replaces the on-disk last-good cache (no filesystem on a Worker), * - `disablePolling` turns off the internal `setInterval` loop (no background * execution on a Worker; a Cron Trigger / DO alarm drives {@link - * SkillBundleSource.refresh} instead). + * SkillBundleSource.refresh} instead), + * - `pushReplayStore` records the pushed tokens' `jti`s where every instance + * reaches them (a Durable Object on a Worker). */ export interface BundleSourceDeps { /** @@ -109,4 +112,9 @@ export interface BundleSourceDeps { * {@link SkillBundleSource.refresh} instead. */ disablePolling?: boolean; + /** + * Where the push replay guard records the `jti`s it accepted, shared by every instance that serves the server. + * Replaces the `saas` source's `pushReplayStorage` option and its in-memory default. + */ + pushReplayStore?: ReplayNonceStore; } diff --git a/libs/edge/README.md b/libs/edge/README.md index 4ed151ab3..332c19167 100644 --- a/libs/edge/README.md +++ b/libs/edge/README.md @@ -138,6 +138,7 @@ crons = ["*/5 * * * *"] | `expectedAudience`, `jwksUrl`, `expectedIssuer` | How the bundle server's tokens are verified | | `cache` | Keeps the last bundle applied: `kvBundleCacheFromEnv('BINDING')`, or a store of your own | | `enableWebhook` | Also take bundles pushed to `POST /__skilled_openapi/push`, as below | +| `pushReplay` | Where pushed tokens' `jti`s are kept: `{ binding }` (default `FRONTMCP_PUSH_REPLAY`), as below | | `allowRollback` | Apply a pulled or pushed bundle older than the one being served, as below. Default `false` | | `requireSignature`, `trustedKeys`, `dev`, `credentials`, `outbound` | Passed to the Skilled OpenAPI plugin. `dev` skips signature checks: never use it in production | | `pollIntervalMs`, `bundleCacheDir` | Ignored on a Worker, which has no timers or file system | @@ -172,9 +173,37 @@ Content-Type: application/json | `409` | `{ "error": "push_replayed" }`: a push already used this token | | `409` | `{ "error": "bundle_outdated" }`: the bundle is older than the one being served, which stays | | `422` | `{ "error": "bundle_not_applied" }`: the signature or the bundle's checks failed; the served bundle stays | -| `503` | `{ "error": "temporarily_unavailable" }`: the JWKS could not be read | +| `503` | `{ "error": "temporarily_unavailable" }`: the JWKS could not be read, or the `jti` could not be recorded | -The endpoint is a hookable flow, `skilled-openapi:bundle-push`, behind the same IP filter and rate limit as the server's other HTTP flows. The `jti`s it has seen are kept in memory, per isolate, so two isolates of a Worker don't share them: keep push tokens short-lived. +The endpoint is a hookable flow, `skilled-openapi:bundle-push`, behind the same IP filter and rate limit as the server's other HTTP flows. + +The `jti`s are kept in a Durable Object that every isolate and replica of the Worker reaches, so a push replayed to another isolate is refused too. (KV can't do this: it is eventually consistent, so two isolates could both read a `jti` as unused.) Export the class and bind it: + +```ts +// src/worker.ts +const mcp = createEdgeMcp({ + // ... + managed: { /* ... */ enableWebhook: true }, // the binding defaults to FRONTMCP_PUSH_REPLAY; set pushReplay: { binding } to change it +}); + +export default mcp; +export const FrontMcpPushReplay = mcp.PushReplayDurableObject; +``` + +```toml +# wrangler.toml +[[durable_objects.bindings]] +name = "FRONTMCP_PUSH_REPLAY" +class_name = "FrontMcpPushReplay" + +[[migrations]] +tag = "v1" # or the next tag, when wrangler.toml already has migrations +new_sqlite_classes = ["FrontMcpPushReplay"] +``` + +- **One object per audience** keeps the `jti`s, named after `expectedAudience`. Each `jti` is kept until its token's `iat` is five minutes old, and dropped after that. +- **Without the binding**, every push answers `503 temporarily_unavailable`, and the log names the missing binding. Pulls and Cron runs are not affected. Up to 1.11.0, the `jti`s were kept in each isolate's memory, which isolates don't share. +- **On Deno or Bun**, set `pushReplay` to a function of `env` that returns a store, `{ claim(key, ttlMs) }`: it resolves `true` when it recorded `key`, `false` when `key` is already recorded, and throws when it can't tell. A push is refused when it throws. ## API @@ -196,9 +225,10 @@ It returns: | ------------------------------- | ----------------------------------------------------------------------- | | `fetch(request, env?, ctx?)` | The Worker's request handler | | `SessionDurableObject` | The Durable Object class for `sessions`. Export it only when you use it | +| `PushReplayDurableObject` | The Durable Object class for `managed.enableWebhook`, as above | | `scheduled(event?, env?, ctx?)` | With `managed` only: the Cron Trigger handler, which pulls the bundle | -Other exports: `kvBundleCacheFromEnv`, `createKvBundleCache`, `kvSkillIndexCacheFromEnv`, `createKvSkillIndexCache`, `buildManagedOpenApiPluginOptions`, and the `EdgeMcp`, `EdgeMcpConfig` and `ManagedEdgeOptions` types. +Other exports: `kvBundleCacheFromEnv`, `createKvBundleCache`, `kvSkillIndexCacheFromEnv`, `createKvSkillIndexCache`, `buildManagedOpenApiPluginOptions`, and the `EdgeMcp`, `EdgeMcpConfig`, `ManagedEdgeOptions`, `EdgePushReplayStore` and `EdgePushReplayStoreFactory` types. ## Docs diff --git a/libs/edge/src/__tests__/index-wiring.spec.ts b/libs/edge/src/__tests__/index-wiring.spec.ts index fd57cda9b..d31c71d79 100644 --- a/libs/edge/src/__tests__/index-wiring.spec.ts +++ b/libs/edge/src/__tests__/index-wiring.spec.ts @@ -20,6 +20,7 @@ import { kvSkillIndexCacheFromEnv, type EdgeMcpConfig, } from '../index'; +import { PushReplayDurableObject } from '../push-replay'; // ---- mocks ----------------------------------------------------------------- @@ -466,6 +467,51 @@ describe('createEdgeMcp — managed mode', () => { expect(controllerCache).toBe(store); }); + /** The push replay store the controller hands the plugin, after the first request built the scope with `env`. */ + async function controllerPushReplayStore(managed: Record, env: unknown) { + let store: { claim(key: string, ttlMs: number): Promise } | undefined; + createForGraph.mockImplementation(async (cfg: { providers?: Array<{ provide: symbol; useValue: unknown }> }) => { + const provider = (cfg.providers ?? []).find((p) => typeof p.provide === 'symbol'); + store = (provider?.useValue as { pushReplayStore?: typeof store }).pushReplayStore; + return builtInstance(makeScope()); + }); + const edge = createEdgeMcp({ ...BASE, managed: { ...MANAGED, ...managed } } as unknown as EdgeMcpConfig); + await edge.fetch(new Request('https://w/'), env); + return store; + } + + it('hands the plugin a push replay store on the Durable Object bound in env, named after the audience', async () => { + const fetchClaim = jest.fn(async () => Response.json({ claimed: true })); + const namespace = { + idFromName: jest.fn((name: string) => `id:${name}`), + get: jest.fn(() => ({ fetch: fetchClaim })), + }; + + const store = await controllerPushReplayStore({ enableWebhook: true }, { FRONTMCP_PUSH_REPLAY: namespace }); + + await expect(store?.claim('jti-hash', 1_000)).resolves.toBe(true); + expect(namespace.idFromName).toHaveBeenCalledWith('push-jti:acme:prod'); + expect(namespace.get).toHaveBeenCalledWith('id:push-jti:acme:prod'); + }); + + it('hands the plugin a store that refuses every push when the Durable Object is not bound', async () => { + const store = await controllerPushReplayStore({ enableWebhook: true, pushReplay: { binding: 'MY_REPLAY' } }, {}); + + await expect(store?.claim('jti-hash', 1_000)).rejects.toThrow( + 'the Durable Object binding "MY_REPLAY" is not bound', + ); + }); + + it('hands the plugin no push replay store without enableWebhook', async () => { + const store = await controllerPushReplayStore({}, { FRONTMCP_PUSH_REPLAY: {} }); + + expect(store).toBeUndefined(); + }); + + it('exposes the push replay Durable Object class on the module', () => { + expect(createEdgeMcp(BASE).PushReplayDurableObject).toBe(PushReplayDurableObject); + }); + it('passes a plain managed.cache store through unchanged', async () => { const store = { read: jest.fn(), write: jest.fn() }; let controllerCache: unknown; diff --git a/libs/edge/src/__tests__/push-replay.spec.ts b/libs/edge/src/__tests__/push-replay.spec.ts new file mode 100644 index 000000000..c1a5ac79e --- /dev/null +++ b/libs/edge/src/__tests__/push-replay.spec.ts @@ -0,0 +1,198 @@ +import { + createDurableObjectPushReplayStore, + DEFAULT_PUSH_REPLAY_BINDING, + PushReplayDurableObject, + resolvePushReplayStore, +} from '../push-replay'; + +/** A Durable Object's `state.storage`, as far as the replay object uses it; `delete` takes 128 keys at most, as on Cloudflare. */ +class FakeDurableObjectStorage { + readonly entries = new Map(); + + async list(): Promise> { + return new Map(this.entries); + } + + async put(key: string, value: unknown): Promise { + this.entries.set(key, value); + } + + async delete(keys: string[]): Promise { + if (keys.length > 128) throw new Error('delete() takes at most 128 keys'); + return keys.filter((key) => this.entries.delete(key)).length; + } +} + +/** A Durable Object namespace: one object per name, as on Cloudflare, whichever isolate asks. */ +function fakeNamespace(): { + idFromName(name: string): unknown; + get(id: unknown): { fetch(request: Request): Promise }; + storageOf(name: string): FakeDurableObjectStorage | undefined; +} { + const objects = new Map(); + return { + idFromName: (name) => `id:${name}`, + get(id) { + const key = String(id); + let entry = objects.get(key); + if (!entry) { + const storage = new FakeDurableObjectStorage(); + entry = { replayObject: new PushReplayDurableObject({ storage }), storage }; + objects.set(key, entry); + } + const { replayObject } = entry; + return { fetch: (request) => replayObject.fetch(request) }; + }, + storageOf: (name) => objects.get(`id:${name}`)?.storage, + }; +} + +const claimRequest = (body: unknown) => + new Request('https://push-replay.frontmcp.internal/claim', { method: 'POST', body: JSON.stringify(body) }); + +describe('PushReplayDurableObject', () => { + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('records a key once and refuses it while it is unexpired', async () => { + const replayObject = new PushReplayDurableObject({ storage: new FakeDurableObjectStorage() }); + + const first = await replayObject.fetch(claimRequest({ key: 'jti-hash', ttlMs: 60_000 })); + const replayed = await replayObject.fetch(claimRequest({ key: 'jti-hash', ttlMs: 60_000 })); + + expect(await first.json()).toEqual({ claimed: true }); + expect(await replayed.json()).toEqual({ claimed: false }); + }); + + it('records a key again once its TTL passed, and drops the expired keys as it claims', async () => { + const storage = new FakeDurableObjectStorage(); + const replayObject = new PushReplayDurableObject({ storage }); + const clock = jest.spyOn(Date, 'now').mockReturnValue(1_000_000); + await replayObject.fetch(claimRequest({ key: 'short-lived', ttlMs: 1_000 })); + await replayObject.fetch(claimRequest({ key: 'long-lived', ttlMs: 600_000 })); + + clock.mockReturnValue(1_000_000 + 1_000); + const atExpiry = await replayObject.fetch(claimRequest({ key: 'short-lived', ttlMs: 1_000 })); + clock.mockReturnValue(1_000_000 + 1_001); + const pastExpiry = await replayObject.fetch(claimRequest({ key: 'short-lived', ttlMs: 1_000 })); + + expect(await atExpiry.json()).toEqual({ claimed: false }); + expect(await pastExpiry.json()).toEqual({ claimed: true }); + expect([...storage.entries.keys()].sort()).toEqual(['long-lived', 'short-lived']); + expect(storage.entries.get('short-lived')).toBe(1_000_000 + 1_001 + 1_000); + }); + + it('drops more expired keys than one delete takes', async () => { + const storage = new FakeDurableObjectStorage(); + for (let index = 0; index < 300; index++) storage.entries.set(`expired-${index}`, 1); + const replayObject = new PushReplayDurableObject({ storage }); + + const response = await replayObject.fetch(claimRequest({ key: 'fresh', ttlMs: 60_000 })); + + expect(await response.json()).toEqual({ claimed: true }); + expect([...storage.entries.keys()]).toEqual(['fresh']); + }); + + it.each([ + ['no key', { ttlMs: 1_000 }], + ['an empty key', { key: '', ttlMs: 1_000 }], + ['a negative TTL', { key: 'jti-hash', ttlMs: -1 }], + ['a TTL that is not a number', { key: 'jti-hash', ttlMs: '60000' }], + ])('answers 400 to a claim with %s', async (_label, body) => { + const storage = new FakeDurableObjectStorage(); + const replayObject = new PushReplayDurableObject({ storage }); + + const response = await replayObject.fetch(claimRequest(body)); + + expect(response.status).toBe(400); + expect(await response.json()).toEqual({ error: 'invalid_claim' }); + expect(storage.entries.size).toBe(0); + }); +}); + +describe('createDurableObjectPushReplayStore', () => { + it('refuses, from a second isolate, a key the first isolate claimed: both reach the same object', async () => { + const namespace = fakeNamespace(); + const firstIsolate = createDurableObjectPushReplayStore(namespace, 'push-jti:acme'); + const secondIsolate = createDurableObjectPushReplayStore(namespace, 'push-jti:acme'); + + const first = await firstIsolate.claim('jti-hash', 60_000); + const replayed = await secondIsolate.claim('jti-hash', 60_000); + + expect([first, replayed]).toEqual([true, false]); + expect(namespace.storageOf('push-jti:acme')?.entries.has('jti-hash')).toBe(true); + }); + + it('keeps the keys of each name apart', async () => { + const namespace = fakeNamespace(); + + await createDurableObjectPushReplayStore(namespace, 'push-jti:acme').claim('jti-hash', 60_000); + + await expect( + createDurableObjectPushReplayStore(namespace, 'push-jti:globex').claim('jti-hash', 60_000), + ).resolves.toBe(true); + }); + + it.each([ + ['an error status', () => Response.json({ error: 'internal' }, { status: 500 })], + ['an answer without "claimed"', () => Response.json({ ok: true })], + ['an answer that is not JSON', () => new Response('', { status: 200 })], + ])('throws, so the push is refused, when the object answers %s', async (_label, answer) => { + const namespace = { idFromName: () => 'id', get: () => ({ fetch: async () => answer() }) }; + + await expect( + createDurableObjectPushReplayStore(namespace, 'push-jti:acme').claim('jti-hash', 1_000), + ).rejects.toThrow(/^the push replay Durable Object answered \d+$/); + }); + + it('throws when the object cannot be reached', async () => { + const namespace = { + idFromName: () => 'id', + get: () => ({ fetch: async () => Promise.reject(new Error('Durable Object reset')) }), + }; + + await expect( + createDurableObjectPushReplayStore(namespace, 'push-jti:acme').claim('jti-hash', 1_000), + ).rejects.toThrow('Durable Object reset'); + }); +}); + +describe('resolvePushReplayStore', () => { + it(`uses the Durable Object bound as ${DEFAULT_PUSH_REPLAY_BINDING} by default, or under the binding named`, async () => { + const defaultNamespace = fakeNamespace(); + const namedNamespace = fakeNamespace(); + const env = { [DEFAULT_PUSH_REPLAY_BINDING]: defaultNamespace, REPLAY: namedNamespace }; + + await resolvePushReplayStore(undefined, env, 'push-jti:acme').claim('by-default', 1_000); + await resolvePushReplayStore({ binding: 'REPLAY' }, env, 'push-jti:acme').claim('by-name', 1_000); + + expect([...(defaultNamespace.storageOf('push-jti:acme')?.entries.keys() ?? [])]).toEqual(['by-default']); + expect([...(namedNamespace.storageOf('push-jti:acme')?.entries.keys() ?? [])]).toEqual(['by-name']); + }); + + it.each([ + ['no env', undefined], + ['no binding', {}], + ['a binding that is not a Durable Object namespace', { [DEFAULT_PUSH_REPLAY_BINDING]: { get: () => undefined } }], + ])('returns a store that refuses every claim when there is %s', async (_label, env) => { + const store = resolvePushReplayStore({}, env, 'push-jti:acme'); + + await expect(store.claim('jti-hash', 1_000)).rejects.toThrow( + `the Durable Object binding "${DEFAULT_PUSH_REPLAY_BINDING}" is not bound: export mcp.PushReplayDurableObject and bind it in wrangler.toml`, + ); + }); + + it('uses the store a function of env returns, and refuses every claim when it returns none', async () => { + const custom = { claim: jest.fn(async () => true) }; + const env = { CUSTOM: true }; + + const fromFunction = resolvePushReplayStore((received) => (received === env ? custom : undefined), env, 'name'); + const fromNothing = resolvePushReplayStore(() => undefined, env, 'name'); + + expect(fromFunction).toBe(custom); + await expect(fromNothing.claim('jti-hash', 1_000)).rejects.toThrow( + 'the managed.pushReplay function returned no store', + ); + }); +}); diff --git a/libs/edge/src/index.ts b/libs/edge/src/index.ts index 25d4d12d3..b04cf39f8 100644 --- a/libs/edge/src/index.ts +++ b/libs/edge/src/index.ts @@ -50,6 +50,7 @@ import { import type { EdgeBundleCacheFactory, EdgeBundleCacheStore } from './kv-cache'; import { buildManagedOpenApiPluginOptions, type ManagedEdgeOptions } from './managed'; +import { PushReplayDurableObject, resolvePushReplayStore, type EdgePushReplayStore } from './push-replay'; import { createEdgeSessionDurableObject, createEdgeSessionRouter } from './session-host'; import { kvSkillIndexCacheFromEnv, type EdgeSkillIndexCacheFactory } from './skill-index-cache'; import { logStartFailure } from './start-failure'; @@ -69,6 +70,7 @@ export { type EdgeKvNamespace, type KvBundleCacheOptions, } from './kv-cache'; +export type { EdgePushReplayStore, EdgePushReplayStoreFactory } from './push-replay'; export { createKvSkillIndexCache, kvSkillIndexCacheFromEnv, @@ -95,17 +97,18 @@ interface BundleScheduler { } /** - * Carries the KV cache + `disablePolling` flag into the skilled-openapi plugin - * and captures the plugin's scheduler so {@link EdgeMcp.scheduled} can run it. A - * Worker has no background timers, so polling is disabled and a Cron Trigger / - * Durable Object alarm drives refresh instead. + * Carries the KV cache, the push replay store and the `disablePolling` flag into + * the skilled-openapi plugin and captures the plugin's scheduler so {@link + * EdgeMcp.scheduled} can run it. A Worker has no background timers, so polling + * is disabled and a Cron Trigger / Durable Object alarm drives refresh instead. */ class EdgeRefreshController { private scheduler?: BundleScheduler; + readonly disablePolling = true; constructor( readonly cache?: EdgeBundleCacheStore, - readonly disablePolling = true, + readonly pushReplayStore?: EdgePushReplayStore, ) {} attach(scheduler: BundleScheduler): void { @@ -269,6 +272,12 @@ export interface EdgeMcp { * ``` */ SessionDurableObject: new (state: unknown, env: unknown) => { fetch(request: Request): Promise }; + /** + * The **Durable Object class** that keeps the push tokens' `jti`s, with + * `managed.enableWebhook`. Re-export it from the worker entry and bind it under + * `managed.pushReplay.binding` (default `FRONTMCP_PUSH_REPLAY`). + */ + PushReplayDurableObject: typeof PushReplayDurableObject; } /** @@ -315,7 +324,12 @@ export function createEdgeMcp(config: EdgeMcpConfig): EdgeMcp { let frontmcpConfig = base; if (managed) { - controller = new EdgeRefreshController(resolveCache(managed.cache, env)); + controller = new EdgeRefreshController( + resolveCache(managed.cache, env), + managed.enableWebhook + ? resolvePushReplayStore(managed.pushReplay, env, `push-jti:${managed.expectedAudience}`) + : undefined, + ); // Wire the skilled-openapi plugin (SaaS source) so the bundle is pulled // on boot and auto-refreshed. Lazy-imported so non-managed edges never // evaluate it; it is a dependency, so wrangler always resolves the literal specifier (#769). @@ -411,6 +425,7 @@ export function createEdgeMcp(config: EdgeMcpConfig): EdgeMcp { // The DO builds its own session-local scope (in its isolate) via buildScope, // and bridges `env`→`process.env` the same way the worker does. SessionDurableObject: createEdgeSessionDurableObject(buildScope, bridgeEnvToProcessEnv), + PushReplayDurableObject, }; if (config.managed) { diff --git a/libs/edge/src/managed.ts b/libs/edge/src/managed.ts index 3fde3bf59..7e9459c49 100644 --- a/libs/edge/src/managed.ts +++ b/libs/edge/src/managed.ts @@ -11,6 +11,7 @@ * cache fallback). */ import type { EdgeBundleCacheFactory, EdgeBundleCacheStore } from './kv-cache'; +import type { EdgePushReplayStoreFactory } from './push-replay'; export interface ManagedEdgeOptions { /** SaaS pull endpoint serving the signed bundle (HTTPS). */ @@ -35,6 +36,12 @@ export interface ManagedEdgeOptions { pollIntervalMs?: number; /** Also take bundles the bundle server pushes, at `POST /__skilled_openapi/push`; see the README for the request. */ enableWebhook?: boolean; + /** + * Where the push replay guard records the push tokens' `jti`s, for every isolate and replica of the Worker. Default + * `{ binding: 'FRONTMCP_PUSH_REPLAY' }`: the Durable Object class `mcp.PushReplayDurableObject`, bound under that + * name. A function of `env` can return another store (on Deno or Bun). With no store, pushes are refused. + */ + pushReplay?: { binding?: string } | EdgePushReplayStoreFactory; /** Apply a pulled or pushed bundle older than the one being served. Default false: bundles only move forward. */ allowRollback?: boolean; /** Require a signed bundle. Default true. */ diff --git a/libs/edge/src/push-replay.ts b/libs/edge/src/push-replay.ts new file mode 100644 index 000000000..0fc68e6a4 --- /dev/null +++ b/libs/edge/src/push-replay.ts @@ -0,0 +1,144 @@ +/** + * The push replay guard's store on Cloudflare: a Durable Object. + * + * With `managed.enableWebhook`, a push token's `jti` may be used once. The isolates and replicas of a Worker share no + * memory, and KV is eventually consistent (two isolates can both read a `jti` as unused), so every isolate records the + * `jti`s in one Durable Object instance, which reads and writes them atomically. + */ + +/** Minimal structural view of a Cloudflare `DurableObjectNamespace`. */ +interface DurableObjectNamespaceLike { + idFromName(name: string): unknown; + get(id: unknown): { fetch(request: Request): Promise }; +} + +/** The parts of a Durable Object's `state.storage` the replay object uses. */ +interface PushReplayStorageLike { + list(): Promise>; + put(key: string, value: unknown): Promise; + delete(keys: string[]): Promise; +} + +/** Records each push token's `jti` once. Structurally the `ReplayNonceStore` of `@frontmcp/adapters/skills`. */ +export interface EdgePushReplayStore { + /** Resolves true when this call recorded `key`, false when it is recorded and not expired; throws on a failure. */ + claim(key: string, ttlMs: number): Promise; +} + +/** Builds the push replay store from the Worker's `env`, where Cloudflare bindings live. */ +export type EdgePushReplayStoreFactory = (env: unknown) => EdgePushReplayStore | undefined; + +interface ReplayClaim { + key: string; + ttlMs: number; +} + +/** Default Durable Object binding of the push replay guard. */ +export const DEFAULT_PUSH_REPLAY_BINDING = 'FRONTMCP_PUSH_REPLAY'; + +const CLAIM_URL = 'https://push-replay.frontmcp.internal/claim'; + +/** The most keys a Durable Object storage `delete()` takes at once. */ +const MAX_KEYS_PER_DELETE = 128; + +function readClaim(body: unknown): ReplayClaim | undefined { + const { key, ttlMs } = (body ?? {}) as Partial; + if (typeof key !== 'string' || key.length === 0) return undefined; + if (typeof ttlMs !== 'number' || !Number.isFinite(ttlMs) || ttlMs < 0) return undefined; + return { key, ttlMs }; +} + +function isUnexpired(expiresAt: unknown, now: number): boolean { + return typeof expiresAt === 'number' && expiresAt >= now; +} + +/** + * The Durable Object class that keeps the push tokens' `jti`s. Export it from the Worker entry and bind it under + * `managed.pushReplay.binding` (default `FRONTMCP_PUSH_REPLAY`). Each claim also drops the expired `jti`s. + */ +export class PushReplayDurableObject { + private readonly storage: PushReplayStorageLike; + + constructor(state: { storage: PushReplayStorageLike }) { + this.storage = state.storage; + } + + async fetch(request: Request): Promise { + const claim = readClaim(await request.json().catch(() => undefined)); + if (!claim) return Response.json({ error: 'invalid_claim' }, { status: 400 }); + return Response.json({ claimed: await this.claim(claim) }); + } + + private async claim({ key, ttlMs }: ReplayClaim): Promise { + const now = Date.now(); + // Only storage calls are awaited from here on, so the object's input gate makes the read and the write atomic. + const expiries = await this.storage.list(); + const expiredKeys = [...expiries].filter(([, expiresAt]) => !isUnexpired(expiresAt, now)).map(([name]) => name); + for (let start = 0; start < expiredKeys.length; start += MAX_KEYS_PER_DELETE) { + await this.storage.delete(expiredKeys.slice(start, start + MAX_KEYS_PER_DELETE)); + } + if (isUnexpired(expiries.get(key), now)) return false; + await this.storage.put(key, now + ttlMs); + return true; + } +} + +/** A push replay store on the {@link PushReplayDurableObject} instance named `name` in `namespace`. */ +export function createDurableObjectPushReplayStore( + namespace: DurableObjectNamespaceLike, + name: string, +): EdgePushReplayStore { + return { + async claim(key: string, ttlMs: number): Promise { + const replayObject = namespace.get(namespace.idFromName(name)); + const response = await replayObject.fetch( + new Request(CLAIM_URL, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ key, ttlMs }), + }), + ); + const answer = (await response.json().catch(() => undefined)) as { claimed?: unknown } | undefined; + if (!response.ok || typeof answer?.claimed !== 'boolean') { + throw new Error(`the push replay Durable Object answered ${response.status}`); + } + return answer.claimed; + }, + }; +} + +function isDurableObjectNamespace(value: unknown): value is DurableObjectNamespaceLike { + const candidate = value as Partial | undefined; + return typeof candidate?.idFromName === 'function' && typeof candidate.get === 'function'; +} + +/** A store that refuses every push, so a missing store is never mistaken for an unused `jti`. */ +function unavailablePushReplayStore(reason: string): EdgePushReplayStore { + return { + async claim(): Promise { + throw new Error(reason); + }, + }; +} + +/** + * The push replay store `pushReplay` names, resolved against the Worker's `env`: the Durable Object bound under its + * `binding`, or what its function returns. With neither, every push is refused. + */ +export function resolvePushReplayStore( + pushReplay: { binding?: string } | EdgePushReplayStoreFactory | undefined, + env: unknown, + name: string, +): EdgePushReplayStore { + if (typeof pushReplay === 'function') { + return pushReplay(env) ?? unavailablePushReplayStore('the managed.pushReplay function returned no store'); + } + const binding = pushReplay?.binding ?? DEFAULT_PUSH_REPLAY_BINDING; + const namespace = (env as Record | undefined)?.[binding]; + if (!isDurableObjectNamespace(namespace)) { + return unavailablePushReplayStore( + `the Durable Object binding "${binding}" is not bound: export mcp.PushReplayDurableObject and bind it in wrangler.toml`, + ); + } + return createDurableObjectPushReplayStore(namespace, name); +} diff --git a/libs/skills/catalog/frontmcp-deployment/references/deploy-to-cloudflare-skills-only.md b/libs/skills/catalog/frontmcp-deployment/references/deploy-to-cloudflare-skills-only.md index c396d1990..3b442b3e1 100644 --- a/libs/skills/catalog/frontmcp-deployment/references/deploy-to-cloudflare-skills-only.md +++ b/libs/skills/catalog/frontmcp-deployment/references/deploy-to-cloudflare-skills-only.md @@ -184,17 +184,41 @@ HTTP flow). The request: | `409` | `push_replayed`: a push already used this token's `jti` | | `409` | `bundle_outdated`: older than the bundle being served, which stays | | `422` | `bundle_not_applied`: the signature or the bundle's checks failed; the served bundle stays | -| `503` | `temporarily_unavailable`: the JWKS could not be read | +| `503` | `temporarily_unavailable`: the JWKS could not be read, or the `jti` could not be recorded | A pushed bundle is applied after the first pull and after bundles already -queued, so a push and a Cron refresh never apply at once. The replay guard keeps -the `jti`s it has seen in memory, per isolate: two isolates of the same Worker -don't share them, so keep push tokens short-lived. +queued, so a push and a Cron refresh never apply at once. + +The replay guard keeps the `jti`s in a Durable Object that every isolate and +replica of the Worker reaches, so a push replayed to another isolate is refused +with `409 push_replayed` too. KV can't guard replays: it is eventually +consistent, so two isolates could both read a `jti` as unused. Export the class +and bind it (default binding `FRONTMCP_PUSH_REPLAY`; `managed.pushReplay: +{ binding }` changes it): + +```ts +const mcp = createEdgeMcp({ /* ..., */ managed: { /* ..., */ enableWebhook: true } }); +export default mcp; +export const FrontMcpPushReplay = mcp.PushReplayDurableObject; +``` + +Each `jti` is kept, in one object per `expectedAudience`, until its token's +`iat` is five minutes old. Without the binding, every push answers `503 +temporarily_unavailable` and the log names the missing binding; pulls and Cron +runs are not affected. Up to 1.11.0 the `jti`s were kept in each isolate's +memory. On Deno or Bun, `managed.pushReplay` can be a function of `env` that +returns a store `{ claim(key, ttlMs) }` (`true` when it recorded `key`, `false` +when it is already recorded, throws when it can't tell). + +On Node, the plugin keeps the `jti`s in the process's memory by default; with +several replicas, set the `saas` source's `pushReplayStorage` to a shared store +(`{ type: 'redis', redis: { url } }`, or `upstash`, `vercel-kv`, `sqlite`). This path is bundled by **wrangler** (not `frontmcp build`), so you maintain `wrangler.toml` yourself — it needs a `[[kv_namespaces]] binding = "BUNDLE_CACHE"` and a `[triggers] crons = [...]` (the `managed.pollIntervalMs` option is -ignored on edge — Workers have no background timers; the Cron drives refresh): +ignored on edge — Workers have no background timers; the Cron drives refresh), +and with `enableWebhook` the push replay Durable Object and its migration: ```toml name = "my-worker" @@ -206,6 +230,15 @@ compatibility_flags = ["nodejs_compat"] binding = "BUNDLE_CACHE" id = "" +# with managed.enableWebhook +[[durable_objects.bindings]] +name = "FRONTMCP_PUSH_REPLAY" +class_name = "FrontMcpPushReplay" + +[[migrations]] +tag = "v1" +new_sqlite_classes = ["FrontMcpPushReplay"] + [triggers] crons = ["*/5 * * * *"] ``` diff --git a/plugins/plugin-skilled-openapi/README.md b/plugins/plugin-skilled-openapi/README.md index 900b48ff2..211257ddb 100644 --- a/plugins/plugin-skilled-openapi/README.md +++ b/plugins/plugin-skilled-openapi/README.md @@ -335,12 +335,12 @@ for a bad one. ### Sources -| `type` | Fields | What it does | -| -------- | ------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `inline` | `content` | The bundle object itself. For tests, bundles built in code, and runtimes without a file system. | -| `static` | `path`, `watch?` | A JSON or YAML bundle file, resolved from the working directory. With `watch: true`, the plugin reloads it when it changes. | -| `npm` | `packageName`, `exportName?`, `verifyProvenance?` | Imports a package and uses one of its exports. Provenance checks aren't implemented yet, so the default `verifyProvenance: true` refuses to load it; set `false` only for a package you pin and trust. | -| `saas` | `endpoint`, `authToken`, `expectedAudience`, `jwksUrl`, `expectedIssuer`, `pollIntervalMs?`, `enableWebhook?`, `allowRollback?` | Pulls the bundle over HTTPS at startup and every `pollIntervalMs` (default 5 minutes), after checking `authToken` against `jwksUrl`, `expectedIssuer` and `expectedAudience`. With `enableWebhook: true`, also takes pushed bundles (below). Applies no bundle older than the one it serves, unless `allowRollback: true` (below). | +| `type` | Fields | What it does | +| -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `inline` | `content` | The bundle object itself. For tests, bundles built in code, and runtimes without a file system. | +| `static` | `path`, `watch?` | A JSON or YAML bundle file, resolved from the working directory. With `watch: true`, the plugin reloads it when it changes. | +| `npm` | `packageName`, `exportName?`, `verifyProvenance?` | Imports a package and uses one of its exports. Provenance checks aren't implemented yet, so the default `verifyProvenance: true` refuses to load it; set `false` only for a package you pin and trust. | +| `saas` | `endpoint`, `authToken`, `expectedAudience`, `jwksUrl`, `expectedIssuer`, `pollIntervalMs?`, `enableWebhook?`, `pushReplayStorage?`, `allowRollback?` | Pulls the bundle over HTTPS at startup and every `pollIntervalMs` (default 5 minutes), after checking `authToken` against `jwksUrl`, `expectedIssuer` and `expectedAudience`. With `enableWebhook: true`, also takes pushed bundles (below). Applies no bundle older than the one it serves, unless `allowRollback: true` (below). | A new bundle from `static` (with `watch`) or `saas` is checked like the first, then replaces the skills and operations at once. A bundle that fails its checks isn't applied, and the reason is logged, with the first issues a schema check @@ -374,14 +374,37 @@ With `enableWebhook: true` on a `saas` source, the server takes bundles the SaaS - The `Authorization: Bearer` token is checked as `authToken` is, and must also carry a `jti` of 8 or more characters that no push used and an `iat` within five minutes. A missing or refused token answers `401 invalid_token`, a used - `jti` answers `409 push_replayed`, and a JWKS that can't be read answers `503 temporarily_unavailable`. + `jti` answers `409 push_replayed`, and a JWKS that can't be read, or a `jti` that can't be recorded, answers + `503 temporarily_unavailable`. - The body is the bundle as JSON, checked as a pulled bundle is: `400 invalid_bundle` when it is not a bundle, `409 bundle_outdated` when it is older than the bundle being served, `422 bundle_not_applied` when its signature or its other checks fail. The served bundle stays. - An applied bundle answers `200 { applied: true, bundleId, bundleVersion }` and is kept as the last good bundle. Pushes and pulls apply one at a time, in the order they arrive. -The used `jti`s are kept in memory: replicas, and the isolates of a Worker, don't share them. +By default the used `jti`s are kept in the process's memory, which is enough for a single process. With several +replicas, a push replayed to another replica would be accepted again, so give them a store they share with +`pushReplayStorage`, a config for `createStorage()` from `@frontmcp/utils`: + +```ts +SkilledOpenApiPlugin.init({ + source: { + type: 'saas', + // endpoint, authToken, expectedAudience, jwksUrl, expectedIssuer ... + enableWebhook: true, + pushReplayStorage: { type: 'redis', redis: { url: process.env.REDIS_URL } }, + }, +}); +``` + +- Its `type` is `redis`, `upstash`, `vercel-kv`, or `sqlite` for processes on one machine. `memory`, `auto` and + `cloudflare-kv` are refused, and so is `fallback: 'memory'`: each replica would keep its own `jti`s. Cloudflare KV is + eventually consistent, so two isolates could both read a `jti` as unused. +- Each `jti` is kept under `frontmcp:skilled-openapi:push-jti::` until its token's `iat` is five + minutes old. +- The store is connected on the first push. While it can't be reached, every push answers + `503 temporarily_unavailable` and the log says why: a push is never accepted unchecked. +- On a Cloudflare Worker, `@frontmcp/edge` keeps the `jti`s in a Durable Object instead (`managed.pushReplay`). ### When no bundle loads diff --git a/plugins/plugin-skilled-openapi/src/__tests__/bundle-push.integration.spec.ts b/plugins/plugin-skilled-openapi/src/__tests__/bundle-push.integration.spec.ts index 32c687834..20e192f51 100644 --- a/plugins/plugin-skilled-openapi/src/__tests__/bundle-push.integration.spec.ts +++ b/plugins/plugin-skilled-openapi/src/__tests__/bundle-push.integration.spec.ts @@ -1,11 +1,13 @@ // A real server whose saas source has enableWebhook takes pushed bundles at POST /__skilled_openapi/push. import 'reflect-metadata'; -import { BundleStore } from '@frontmcp/adapters/skills'; +import { BundleStore, StorageReplayNonceStore, type ReplayNonceStore } from '@frontmcp/adapters/skills'; import { createWebFetchHandler, FrontMcpInstance, LogLevel, type WebFetchHandler } from '@frontmcp/sdk'; +import { createMemoryStorage } from '@frontmcp/utils'; import { SkilledOpenApiPlugin } from '../index'; import { SKILLED_OPENAPI_RUNTIME_DEPS_TOKEN } from '../skilled-openapi.plugin'; +import { BundleSyncService } from '../sync/bundle-sync.service'; import { AUDIENCE, bundle, ENDPOINT, ISSUER, JWKS_URL, signed, tokenIssuer, trustedKey } from './saas-bundles.fixture'; const PUSH_URL = 'http://localhost/__skilled_openapi/push'; @@ -13,6 +15,7 @@ const PUSH_URL = 'http://localhost/__skilled_openapi/push'; interface PushServer { handler: WebFetchHandler; servedVersion: () => string | undefined; + loadBundle: () => Promise; } describe('pushed bundles on a server with enableWebhook', () => { @@ -20,7 +23,7 @@ describe('pushed bundles on a server with enableWebhook', () => { let jwksAvailable = true; const cache = { read: jest.fn(async () => undefined), write: jest.fn(async () => undefined) }; - async function startServer(enableWebhook: boolean): Promise { + async function startServer(enableWebhook: boolean, pushReplayStore?: ReplayNonceStore): Promise { const instance = await FrontMcpInstance.createForGraph({ info: { name: 'billing', version: '1.0.0' }, apps: [], @@ -43,7 +46,7 @@ describe('pushed bundles on a server with enableWebhook', () => { { name: 'test:runtime-deps', provide: SKILLED_OPENAPI_RUNTIME_DEPS_TOKEN, - useValue: { cache, disablePolling: true }, + useValue: { cache, disablePolling: true, pushReplayStore }, }, ], logging: { level: LogLevel.Off }, @@ -54,6 +57,7 @@ describe('pushed bundles on a server with enableWebhook', () => { return { handler: createWebFetchHandler(scope as Parameters[0]), servedVersion: () => scope.providers.get(BundleStore).current()?.version, + loadBundle: () => scope.providers.get(BundleSyncService).ensureReady(), }; } @@ -173,6 +177,63 @@ describe('pushed bundles on a server with enableWebhook', () => { expect(cache.write).not.toHaveBeenCalled(); }); + describe('replicas that share a replay store', () => { + let sharedStore: ReplayNonceStore; + let firstReplica: PushServer; + let secondReplica: PushServer; + + beforeAll(async () => { + const storage = createMemoryStorage({ enableSweeper: false }); + await storage.connect(); + sharedStore = new StorageReplayNonceStore(storage); + firstReplica = await startServer(true, sharedStore); + secondReplica = await startServer(true, sharedStore); + await Promise.all([firstReplica.loadBundle(), secondReplica.loadBundle()]); + }); + + it('refuse, with 409, a push replayed to another replica, which keeps serving the bundle it had', async () => { + const token = await issuer.pushToken(); + const first = await push(firstReplica, `Bearer ${token}`, signed(bundle('2.0.0', ['invoices', 'refunds']))); + + const replayed = await push(secondReplica, `Bearer ${token}`, signed(bundle('2.0.0', ['invoices', 'refunds']))); + + 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(firstReplica.servedVersion()).toBe('2.0.0'); + expect(secondReplica.servedVersion()).toBe('1.0.0'); + }); + + it('take a push with a fresh token on either replica', async () => { + const response = await push( + secondReplica, + `Bearer ${await issuer.pushToken()}`, + signed(bundle('2.0.0', ['invoices', 'refunds'])), + ); + + expect(response.status).toBe(200); + expect(secondReplica.servedVersion()).toBe('2.0.0'); + }); + }); + + it('refuses a push with 503 when the replay store fails, and applies nothing', async () => { + const failingStore = { claim: jest.fn(async () => Promise.reject(new Error('ECONNREFUSED 10.0.0.7:6379'))) }; + const storeDown = await startServer(true, failingStore); + await storeDown.loadBundle(); + + const response = await push(storeDown, `Bearer ${await issuer.pushToken()}`, signed(bundle('9.0.0', ['invoices']))); + + expect(response.status).toBe(503); + expect(await response.json()).toEqual({ + error: 'temporarily_unavailable', + error_description: "the replay guard's store could not record the push token", + }); + expect(storeDown.servedVersion()).toBe('1.0.0'); + }); + it('serves no push endpoint when enableWebhook is off', async () => { const pullOnly = await startServer(false); diff --git a/plugins/plugin-skilled-openapi/src/push/bundle-push.flow.ts b/plugins/plugin-skilled-openapi/src/push/bundle-push.flow.ts index 4f57a91d4..3abc1b95b 100644 --- a/plugins/plugin-skilled-openapi/src/push/bundle-push.flow.ts +++ b/plugins/plugin-skilled-openapi/src/push/bundle-push.flow.ts @@ -101,14 +101,18 @@ export default class BundlePushFlow extends FlowBase { } refused = error; } - if (refused) { - this.logger.warn(refused.message); - return this.respond( - refused.refusal === 'replayed' - ? refusal(409, 'push_replayed', 'a push already used this token') - : invalidToken('the push token was refused'), - ); + if (refused) return this.respond(this.answerRefusal(refused)); + } + + private answerRefusal(refused: SaasPushRefusedError) { + if (refused.refusal === 'replay-unchecked') { + this.logger.error(`bundle push not checked: ${refused.message}`); + return refusal(503, 'temporarily_unavailable', "the replay guard's store could not record the push token"); } + this.logger.warn(refused.message); + return refused.refusal === 'replayed' + ? refusal(409, 'push_replayed', 'a push already used this token') + : invalidToken('the push token was refused'); } @Stage('parseBundle')