diff --git a/AGENTS.md b/AGENTS.md index 6c58cd9..f5756e2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -128,7 +128,7 @@ The core innovation: instead of 2,500 MCP tools (~244K tokens), two tools handle - `src/events/methods.ts` registers the event catalogue and subscription methods on the authenticated MCP endpoint. - ANS policies/destinations and Vega automations own subscription state. No token or signing secret is stored: the client's `whsec_` secret and callback URL are sealed into the ANS destination secret (ANS presents it on every call), `OAUTH_KV` `mcp-events:subscription:` holds plain routing/expiry and a hash of the signing secret, and the policy description holds encrypted metadata with that hash. A new signing secret gets a new destination. Access is checked on subscribe/refresh, not per delivery. Keep KV to hashes, never raw tokens or secrets. -- `src/events/callback.ts` checks the ANS ingress HMAC, reads the delivery ticket and synchronously signs the MCP callback, with no Cloudflare API call. ANS owns retries. +- `src/events/callback.ts` opens the sealed destination secret ANS presents, checks the KV subscription record and synchronously signs the MCP callback, with no Cloudflare API call. ANS owns retries. - `npm run test:events` runs local workerd coverage with simulated external HTTP endpoints. See `docs/mcp-events.md` for live tests and lifecycle constraints. ### Worker Loader API diff --git a/docs/mcp-events.md b/docs/mcp-events.md index 71ab2c3..50da320 100644 --- a/docs/mcp-events.md +++ b/docs/mcp-events.md @@ -25,7 +25,8 @@ derived from the server, principal, event, arguments and callback URL. Refresh r them. Unsubscribe disables the policy first, then removes the automation, policy and destination. Retry a failed subscribe to resume partial provisioning, or unsubscribe to remove it. No Durable Object, queue or new binding is used; the only -state outside ANS is a short-lived delivery ticket in the existing `OAUTH_KV`. +state outside ANS is a short-lived subscription record (no secrets) in the existing +`OAUTH_KV`. ANS's generic webhook format differs from MCP Events, so its destination targets `/events/ans/{subscriptionId}` on this server. This route checks ANS's ingress diff --git a/node_modules b/node_modules new file mode 120000 index 0000000..59684af --- /dev/null +++ b/node_modules @@ -0,0 +1 @@ +../mcp-events-no-stored-token/node_modules \ No newline at end of file diff --git a/src/events/service.ts b/src/events/service.ts index 0189c89..f5e15cb 100644 --- a/src/events/service.ts +++ b/src/events/service.ts @@ -102,7 +102,7 @@ export async function readPolicyState( policy: ManagedPolicy, id: string ): Promise { - const state = await parsePolicyState(await decryptState(env, `policy:${id}`, policy.description)) + const state = parsePolicyState(await decryptState(env, `policy:${id}`, policy.description)) if ( state.id !== id || policy.name !== managedName(id) || @@ -241,11 +241,10 @@ export class EventService { } const url = new URL(`/events/ans/${id}`, this.env.MCP_RESOURCE).href CallbackUrl.parse(url) - // TODO(2026-10-16): drop the "destination from an older version" case once subscriptions made before #283 have expired or been refreshed (they last at most an hour). // The signing secret lives only in the ANS destination's secret, sealed // with the callback URL. ANS ignores updates to a destination's secret, - // so a new signing secret (or a destination from an older version, or a - // half-finished subscribe) gets a new destination; the old ones are + // so a new signing secret (or a half-finished subscribe) gets a new + // destination; the old ones are // deleted once the policy points at the new one. ANS probes a new // destination on creation; the callback acknowledges the probe because // the KV record does not match it yet. @@ -324,7 +323,7 @@ export class EventService { const id = await subscriptionId(`${this.env.MCP_RESOURCE}:${this.principal}`, params) return serialized(id, async () => { // Stop forwarding first, so delivery ends even if ANS cleanup fails. The - // ID is derived from this principal, so it can only name its own ticket. + // ID is derived from this principal, so it can only name its own record. await deleteSubscriptionRecord(this.env, id) const api = this.api(params.arguments.account_id) const name = managedName(id) diff --git a/src/events/state.ts b/src/events/state.ts index 2c436b3..2ec4488 100644 --- a/src/events/state.ts +++ b/src/events/state.ts @@ -1,6 +1,6 @@ import { getOAuthApi } from '@cloudflare/workers-oauth-provider' import { z } from 'zod' -import { SubscribeParams, UnsubscribeParams } from './ans-bridge' +import { UnsubscribeParams } from './ans-bridge' export interface EventEnvironment { OAUTH_KV: KVNamespace @@ -23,21 +23,6 @@ export async function hashSecret(secret: string): Promise { return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, '0')).join('') } -/** - * Policy state written before the signing secret was hashed. Read only, to migrate. - * - * TODO(2026-10-16): remove once subscriptions made before #283 have expired or been refreshed (they last at most an hour). - */ -const LegacyPolicyState = z.object({ - version: z.literal(1), - id: SubscriptionIdSchema, - principal: z.string(), - params: SubscribeParams, - webhookId: z.string(), - expiresAt: z.number(), - verifiedUntil: z.number() -}) - /** * Subscription metadata kept, encrypted, in the ANS policy description. It * identifies the owner and the subscription key, and holds only a hash of the @@ -54,34 +39,12 @@ const CurrentPolicyState = z.object({ verifiedUntil: z.number() }) -/** Parsed policy state, with older versions upgraded in memory. */ +/** Subscription metadata from the ANS policy description. */ export type PolicyState = z.infer -/** - * Parse policy state from its decrypted JSON, upgrading the version that held - * the raw signing secret. The upgraded state is written back on the next - * subscribe or refresh. - * - * TODO(2026-10-16): remove once subscriptions made before #283 have expired or been refreshed (they last at most an hour). Then parse `CurrentPolicyState` directly. - */ -export async function parsePolicyState(value: unknown): Promise { - const legacy = LegacyPolicyState.safeParse(value) - if (!legacy.success) return CurrentPolicyState.parse(value) - const { secret, ...delivery } = legacy.data.params.delivery - return { - version: 2, - id: legacy.data.id, - principal: legacy.data.principal, - params: { - name: legacy.data.params.name, - arguments: legacy.data.params.arguments, - delivery - }, - signingSecretHash: await hashSecret(secret), - webhookId: legacy.data.webhookId, - expiresAt: legacy.data.expiresAt, - verifiedUntil: legacy.data.verifiedUntil - } +/** Parse policy state from its decrypted JSON. */ +export function parsePolicyState(value: unknown): PolicyState { + return CurrentPolicyState.parse(value) } /** @@ -145,12 +108,6 @@ const SubscriptionRecord = z.object({ export type SubscriptionRecord = z.infer const RECORD_PREFIX = 'mcp-events:subscription:' -/** - * Earlier versions kept an encrypted ticket here; removed on refresh. - * - * TODO(2026-10-16): remove once subscriptions made before #283 have expired or been refreshed (they last at most an hour). - */ -const LEGACY_TICKET_PREFIX = 'mcp-events:ticket:' /** KV's minimum TTL; also covers clock skew between refresh and expiry. */ const RECORD_GRACE_SECONDS = 60 @@ -168,7 +125,6 @@ export async function writeSubscriptionRecord( await env.OAUTH_KV.put(`${RECORD_PREFIX}${record.id}`, JSON.stringify(record), { expirationTtl: Math.max(ttl, RECORD_GRACE_SECONDS) }) - await env.OAUTH_KV.delete(`${LEGACY_TICKET_PREFIX}${record.id}`) } /** @@ -186,12 +142,9 @@ export async function readSubscriptionRecord( return record.success && record.data.id === id ? record.data : null } -/** Delete a subscription's record (and any legacy ticket) so delivery stops. */ +/** Delete a subscription's record so delivery stops. */ export async function deleteSubscriptionRecord(env: EventEnvironment, id: string): Promise { - await Promise.all([ - env.OAUTH_KV.delete(`${RECORD_PREFIX}${id}`), - env.OAUTH_KV.delete(`${LEGACY_TICKET_PREFIX}${id}`) - ]) + await env.OAUTH_KV.delete(`${RECORD_PREFIX}${id}`) } function oauth(env: EventEnvironment) { diff --git a/tests/events-e2e.test.ts b/tests/events-e2e.test.ts index 0a2ef06..079a6ea 100644 --- a/tests/events-e2e.test.ts +++ b/tests/events-e2e.test.ts @@ -4,7 +4,7 @@ import { http, HttpResponse } from 'msw' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { AUTOMATIONS, POLICIES, WEBHOOKS } from '../src/events/api' import { USER_AGENT } from '../src/constants' -import { decryptState, encryptState } from '../src/events/state' +import { decryptState } from '../src/events/state' import { API_BASE, cfSuccess, mockIdentityProbe } from './helpers/cloudflare-api' import { clearKv } from './helpers/kv' import { modernMcpRequest } from './helpers/mcp' @@ -490,7 +490,7 @@ describe('MCP Events through the real Worker', () => { vi.spyOn(Date, 'now').mockReturnValue(firstExpiry - 60_000) const refreshed = await rpc('events/subscribe', { ...params, ttlMs: 30 * 60_000 }) expect(Date.parse(refreshed.result!.refreshBefore!)).toBeGreaterThan(firstExpiry) - // Past the first expiry: the ANS secret never changed, but the ticket did. + // Past the first expiry: the destination never changed, but the KV record did. vi.spyOn(Date, 'now').mockReturnValue(firstExpiry + 60_000) expect((await exports.default.fetch(ansRequest())).status).toBe(204) expect(deliveries).toHaveLength(1) @@ -521,37 +521,6 @@ describe('MCP Events through the real Worker', () => { expect(String([...records(WEBHOOKS).values()][0].secret)).not.toContain(signingSecret) }) - // TODO(2026-10-16): remove once subscriptions made before #283 have expired or been refreshed (they last at most an hour). - it('upgrades a subscription made by the previous version on refresh', async () => { - const subscribed = await rpc('events/subscribe', params) - const id = subscribed.result!.id! - // Rewrite it the way the previous version stored it: raw secret in the - // policy state, an ingress-HMAC destination and a KV ticket. - const policy = [...records(POLICIES).values()][0] - const destination = [...records(WEBHOOKS).values()][0] - destination.secret = 'mcp-events-ingress-v1:legacy' - policy.description = await encryptState(env, `policy:${id}`, { - version: 1, - id, - principal: `account:${accountId}`, - params, - webhookId: destination.id, - expiresAt: Date.now() + 60_000, - verifiedUntil: 0 - }) - await env.OAUTH_KV.delete(`mcp-events:subscription:${id}`) - await env.OAUTH_KV.put(`mcp-events:ticket:${id}`, 'legacy-ticket') - expect((await exports.default.fetch(ansRequest(destination))).status).toBe(401) - - expect((await rpc('events/subscribe', params)).error).toBeUndefined() - expect([...records(WEBHOOKS).values()]).toHaveLength(1) - expect([...records(WEBHOOKS).values()][0].id).not.toBe(destination.id) - expect(await env.OAUTH_KV.get(`mcp-events:ticket:${id}`)).toBeNull() - expect(await storedByServer()).not.toContain(signingSecret) - expect((await exports.default.fetch(ansRequest())).status).toBe(204) - expect(deliveries).toHaveLength(1) - }) - it('drops a delivery the receiver rejects with 410 but keeps the subscription', async () => { await rpc('events/subscribe', params) callbackStatus = 410