diff --git a/AGENTS.md b/AGENTS.md index 345a0d1..280243a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -21,6 +21,7 @@ cloudflare-mcp/ │ ├── index.ts # Worker entry point & OAuth routing │ ├── mcp-handler.ts # Stateless MCP HTTP handler & deployment guards │ ├── server.ts # MCP server setup & tool registration +│ ├── events/ # Stateless MCP Events mapping to ANS policies and Vega automations │ ├── executor.ts # Code executor (Worker Loader API) │ ├── spec-processor.ts # OpenAPI spec fetching & $ref resolution │ ├── truncate.ts # Response truncation (~6K token limit) @@ -124,6 +125,13 @@ The core innovation: instead of 2,500 MCP tools (~244K tokens), two tools handle - No MCP session ID, protocol transport state, replay store, Durable Object, or Node async-context bridge is used. This server publishes no change notifications, so both tool modes advertise `tools.listChanged: false`. For `subscriptions/listen`, the SDK acknowledges with an empty honored filter and, since it honours none of the requested types, ends the stream straight away with a `complete` result (SDK 2.2.0+). No SSE stream stays open. - Deployment-static Host and browser Origin allowlists cover localhost, staging, and production. Do not derive either trust list from the incoming request URL or headers. +### MCP Events + +- `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. MCP metadata is encrypted in the policy description. ANS ignores updates to a destination's secret, so that secret is a fixed HMAC ingress credential; the delivery ticket (current bearer, policy ID, expiry) is encrypted in `OAUTH_KV` under `mcp-events:ticket:` and rewritten on every refresh. +- `src/events/callback.ts` checks ANS ingress authentication, current policy/access and synchronously signs the MCP callback. 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 Code execution uses Cloudflare's Worker Loader API to dynamically create isolated worker instances. The API token is passed via props (never enters user code isolate). A `globalOutbound` service restricts network access. diff --git a/docs/mcp-events.md b/docs/mcp-events.md new file mode 100644 index 0000000..0741ed7 --- /dev/null +++ b/docs/mcp-events.md @@ -0,0 +1,181 @@ +# MCP Events for real-time issues + +The authenticated `/mcp` endpoint advertises `events: { listChanged: false }` and +implements webhook delivery from the +[MCP Events draft](https://github.com/modelcontextprotocol/experimental-ext-triggers-events/blob/main/docs/design-sketch-proposal.md): +`events/list`, `events/subscribe` and `events/unsubscribe`. Poll and push +(`events/poll`, `events/stream`) are not offered. The supported event is +`cloudflare.alert.workers_observability_real_time_issue`. + +`events/list` describes event types and argument/payload schemas, not previous +notifications. Discovery checks ANS eligibility and Vega issue automation access. +User-account discovery is paginated with `nextCursor`. + +## Subscription mapping + +| MCP concept | Existing Cloudflare resource | +| --- | --- | +| Subscription | ANS notification policy | +| Callback destination | ANS generic webhook destination | +| Worker and trigger settings | Vega issue automation | +| Delivery attempts and retries | ANS webhook delivery | + +The MCP Worker is stateless. Resources have a deterministic `MCP Events sub_…` name +derived from the server, principal, event, arguments and callback URL. Refresh reuses +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`. + +ANS's generic webhook format differs from MCP Events, so its destination targets +`/events/ans/{subscriptionId}` on this server. This route checks the current policy +and permissions, translates the notification, signs it and awaits acceptance: + +```text +Vega → ANS policy/destination → MCP callback adapter → subscriber +``` + +### Metadata and credentials + +The policy description contains encrypted MCP metadata: owner, arguments, callback +URL, signing keys, destination ID and expiry. + +ANS keeps a destination's secret from creation and ignores later updates, so the +secret is a fixed per-subscription ingress credential, +`HMAC(key, MCP_RESOURCE + subscription ID)`. It only proves the caller is ANS for +that subscription. Everything that changes on refresh (policy ID, the subscriber's +current bearer, expiry) is a delivery ticket in `OAUTH_KV` under +`mcp-events:ticket:`, rewritten on every subscribe and refresh and +expiring with the subscription. The callback checks `cf-webhook-auth` against the +HMAC, reads the ticket, and acknowledges without forwarding when there is none +(ANS's creation probe, expiry or unsubscribe). Unsubscribe deletes the ticket +first, so forwarding stops even if ANS cleanup fails. + +Credentials are never placed in URLs or returned to MCP clients. The policy state +and the ticket use AES-GCM with a purpose-specific key derived from +`MCP_COOKIE_ENCRYPTION_KEY`, bound to `MCP_RESOURCE` and the subscription. + +No refresh token is retained. Provider-issued MCP tokens are revalidated through the +existing OAuth provider. Direct Cloudflare tokens are rechecked by reading the +policy and issue automations. Rotating the encryption key or changing the public +resource URL requires recreating subscriptions. Do not manually edit managed +resource names or policy descriptions. + +Subscriptions default to 30 minutes and expire before the authenticating MCP token. +`ttlMs` is clamped to between one minute and one hour, never rejected; `ttlMs: null` +receives the one-hour maximum. `cursor` and `maxAgeMs` are accepted and ignored. Clients refresh +before `refreshBefore`. Key rotation dual-signs for five minutes; callback verification +is cached in the policy for five minutes for the same owner, URL and key. + +Expiry stops delivery without a scheduler. Idle expired resources remain in ANS +until refresh or unsubscribe. If permissions are revoked, the policy is disabled and +an account administrator may need to remove the managed resources. `terminated` and +`gap` control envelopes are not sent (ChatGPT does not support them). + +ANS creation does not expose atomic create-if-absent. Sequential retries reuse +resources; overlapping calls within an isolate are serialized. Duplicate managed +names created concurrently by different isolates produce a conflict on subsequent +requests. Remove duplicates using ANS/Vega APIs before retrying. Globally atomic +creation requires backend idempotency support. + +## Arguments and permissions + +Provide `account_id`, optional Worker `service`, and exactly one trigger: + +- `afterOccurrences`: positive occurrence count; use `1` for new issues. +- `afterInactivitySeconds`: recurrence after 3,600–31,536,000 inactive seconds. + +These are Vega settings, not ANS policy filters. Issue automations must be enabled +for the account. Credentials must allow reading and managing notification +policies/destinations and Workers Observability issue automations. The consent page +defaults to read-only scopes, so OAuth users must add `notifications.write` and +`workers-observability.write` there before subscribing. + +ChatGPT supplies its callback URL and signing secret when the user asks it to +monitor an event. The server verifies the signed callback challenge before creating +resources. Refresh and unsubscribe use the original identity and arguments. + +## Errors + +| Code | When | +| --- | --- | +| `-32602` InvalidParams | Arguments, callback URL or `whsec_` secret are invalid | +| `-32011` NotFound (`data.kind: "event"`) | Unknown event, or not available on the account | +| `-32012` Forbidden | Missing Cloudflare permissions, another account, or another principal's subscription | +| `-32015` CallbackEndpointError | Verification failed; `data.reason` is `challenge_failed`, `timeout`, `connection_refused`, `tls_error`, `http_4xx` or `http_5xx` | +| `-32603` | Cloudflare API failure; retry, partial resources are reused | + +`events/unsubscribe` is idempotent and returns `{}` when nothing matches, as ChatGPT +expects; `delivery.mode` is optional there. + +## Delivery behavior + +Each ANS request makes at most one outbound attempt. A `2xx` is acknowledged. As +the draft requires, `410` and `413` are non-retryable for that delivery only: ANS +gets a `204` and the subscription stays active. Every other outcome (network +failures, timeouts, redirects, other `4xx` and `5xx`) returns `503` so ANS retries +with its own backoff. Expired, disabled, removed or unauthorized subscriptions are +acknowledged without forwarding. + +The entire callback path has a four-second deadline, below ANS's five-second default. +There is no background delivery or second retry scheduler. Event IDs use Vega's +persisted run ID (`alert_correlation_id`) and the subscription ID. Redeliveries keep +the event ID/body and receive fresh signatures. Missing run IDs are rejected. + +Callbacks require HTTPS on port 443, a public hostname, and no credentials or fragment. +Verification and delivery use Workers' public Internet fetch path with +`global_fetch_strictly_public`; no private-network binding is used. Workerd's default +Internet network restricts connections to public addresses. Redirects are handled +manually and never followed. Preserve those restrictions in other runtimes. +Payloads are limited to 256 KiB. Events have no replay support (`cursor: null`). +Callback acceptance acknowledges receipt, not completion of an investigation. + +## Local end-to-end tests + +```sh +npm ci +npm run test:events +``` + +The suite runs the real Worker, authentication, MCP transport, API mapping, +encryption and callback route under workerd. Only external Cloudflare APIs and the +subscriber HTTP endpoint are simulated with MSW. It covers discovery, ANS's creation +probe, provisioning, signatures, retry responses, refresh, expiry, revocation, +partial provisioning and unsubscribe. No credentials or deployment are required. + +## Live test with a local Worker + +This creates real resources in your selected account. Keep the original identity +and arguments for cleanup. + +1. Configure the usual `.dev.vars`, including a persistent random + `MCP_COOKIE_ENCRYPTION_KEY` of at least 32 characters. Start `npm run dev`. +2. Expose port 2529 with HTTPS: + `cloudflared tunnel --url http://localhost:2529 --http-host-header localhost:2529`. + Set `MCP_RESOURCE=https:///mcp` in `.dev.vars` and restart the Worker. + Keep that URL stable while subscribed. +3. Generate a key with + `export MCP_EVENT_SIGNING_SECRET="whsec_$(openssl rand -base64 32)"`. + Run `npm run events:receiver` in that environment. Expose port 8788 through another + HTTPS tunnel; set `MCP_EVENT_CALLBACK_URL=https:///events`. +4. In the client terminal set `CLOUDFLARE_API_TOKEN`, `CLOUDFLARE_ACCOUNT_ID`, + `MCP_EVENT_CALLBACK_URL`, the same `MCP_EVENT_SIGNING_SECRET`, and optionally + `MCP_EVENT_WORKER`. `MCP_URL` defaults to `http://localhost:2529/mcp`. +5. Run `node scripts/mcp-events-client.mjs list`, then + `node scripts/mcp-events-client.mjs subscribe`. The receiver prints + `Callback verified`; the client returns the subscription ID and expiry. +6. Trigger a new issue in the selected Worker. The receiver verifies the signature + and prints the event. To exercise ANS retries, restart the receiver with + `MCP_EVENT_RESPONSE_STATUS=503`, trigger an issue, then restart with its default + status. ANS redelivers with the same event ID. +7. Run `node scripts/mcp-events-client.mjs unsubscribe` with the same account, + Worker, callback URL and identity. Confirm removal in ANS. Unsubscribe even if + an earlier provisioning request failed. + +To use ChatGPT instead of the receiver, connect the HTTPS `/mcp` endpoint through +your plugin and ask it to monitor issues. ChatGPT supplies its own callback and key. + +Protocol: https://developers.openai.com/plugins/build/mcp-events + +Runtime network policy: +https://github.com/cloudflare/workerd/blob/main/src/workerd/server/workerd.capnp diff --git a/package.json b/package.json index 34d6377..99ec44b 100644 --- a/package.json +++ b/package.json @@ -13,6 +13,8 @@ "format": "oxfmt --write src/", "format:check": "oxfmt --check src/", "test": "vitest run", + "test:events": "vitest run tests/events-e2e.test.ts tests/ans-bridge.test.ts", + "events:receiver": "node scripts/mcp-events-receiver.mjs", "test:watch": "vitest", "check": "npm run format:check && npm run lint && npm run typecheck && npm run test", "seed:staging": "tsx scripts/seed-r2.ts staging", diff --git a/scripts/mcp-events-chatgpt-sim.mjs b/scripts/mcp-events-chatgpt-sim.mjs new file mode 100644 index 0000000..0eb494c --- /dev/null +++ b/scripts/mcp-events-chatgpt-sim.mjs @@ -0,0 +1,231 @@ +// Plays the ChatGPT side of MCP Events webhook delivery against a real server: +// server/discover → events/list → events/subscribe with a callback URL and a +// whsec_ secret, then receives, verifies and prints deliveries. A cloudflared +// quick tunnel gives the local receiver a public HTTPS URL. +// +// export CLOUDFLARE_API_TOKEN=... # or MCP_TOKEN: any bearer /mcp accepts +// export CLOUDFLARE_ACCOUNT_ID=... +// export MCP_EVENT_WORKER=my-worker # optional; omit to watch the account +// node scripts/mcp-events-chatgpt-sim.mjs +// +// MCP_URL defaults to https://mcp.cloudflare.com/mcp. Ctrl-C unsubscribes and +// closes the tunnel. Set MCP_EVENT_RESPONSE_STATUS=503 to make the receiver +// reject deliveries (to watch retries), or 410 to drop them. + +import { spawn } from 'node:child_process' +import { createHmac, randomBytes, timingSafeEqual } from 'node:crypto' +import { createServer } from 'node:http' + +const EVENT = 'cloudflare.alert.workers_observability_real_time_issue' +const PROTOCOL = '2026-07-28' +const endpoint = process.env.MCP_URL ?? 'https://mcp.cloudflare.com/mcp' +const token = process.env.MCP_TOKEN ?? process.env.CLOUDFLARE_API_TOKEN +const accountId = process.env.CLOUDFLARE_ACCOUNT_ID +const worker = process.env.MCP_EVENT_WORKER +const port = Number(process.env.MCP_EVENT_RECEIVER_PORT ?? 8788) +const deliveryStatus = Number(process.env.MCP_EVENT_RESPONSE_STATUS ?? 200) +if (!token) throw new Error('Set CLOUDFLARE_API_TOKEN or MCP_TOKEN') +if (!accountId) throw new Error('Set CLOUDFLARE_ACCOUNT_ID') + +const secret = `whsec_${randomBytes(32).toString('base64')}` +const key = Buffer.from(secret.slice(6), 'base64') +const seen = new Set() +const log = (label, value = '') => + console.log( + `${new Date().toISOString().slice(11, 19)} ${label}`, + typeof value === 'string' ? value : JSON.stringify(value, null, 2) + ) + +// --- MCP client (stateless 2026-07-28 requests, like ChatGPT) --------------- +let nextId = 1 +async function rpc(method, params = {}) { + const body = { + jsonrpc: '2.0', + id: nextId++, + method, + params: { + ...params, + _meta: { + 'io.modelcontextprotocol/protocolVersion': PROTOCOL, + 'io.modelcontextprotocol/clientInfo': { name: 'mcp-events-chatgpt-sim', version: '1.0.0' }, + 'io.modelcontextprotocol/clientCapabilities': {} + } + } + } + const response = await fetch(endpoint, { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'MCP-Protocol-Version': PROTOCOL, + 'Mcp-Method': method, + ...(params.name ? { 'Mcp-Name': params.name } : {}) + }, + body: JSON.stringify(body) + }) + const text = await response.text() + const json = response.headers.get('content-type')?.includes('text/event-stream') + ? text + .split('\n') + .filter((line) => line.startsWith('data:')) + .map((line) => JSON.parse(line.slice(5))) + .find((message) => message.id === body.id) + : JSON.parse(text || '{}') + if (!response.ok || json?.error) + throw new Error(`${method} failed (${response.status}): ${JSON.stringify(json?.error ?? text)}`) + return json.result +} + +// --- Webhook receiver (what ChatGPT's callback endpoint does) --------------- +function verify(headers, body) { + const id = headers['webhook-id'] + const timestamp = headers['webhook-timestamp'] + if (!id || !/^\d+$/.test(timestamp ?? '')) return 'missing webhook headers' + if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return 'stale timestamp' + const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${body}`).digest() + const ok = String(headers['webhook-signature'] ?? '') + .split(' ') + .some((signature) => { + if (!signature.startsWith('v1,')) return false + const actual = Buffer.from(signature.slice(3), 'base64') + return actual.length === expected.length && timingSafeEqual(actual, expected) + }) + return ok ? null : 'bad signature' +} + +const server = createServer(async (request, response) => { + if (request.method === 'GET') return response.writeHead(200).end('ok') + if (request.method !== 'POST' || request.url !== '/hook') return response.writeHead(404).end() + const chunks = [] + for await (const chunk of request) chunks.push(chunk) + const body = Buffer.concat(chunks).toString('utf8') + const problem = verify(request.headers, body) + if (problem) { + log(`✗ rejected delivery: ${problem}`, { headers: request.headers }) + return response.writeHead(401).end() + } + const message = JSON.parse(body) + if (message.type === 'verification') { + log('✓ verification challenge answered') + return response + .writeHead(200, { 'Content-Type': 'application/json' }) + .end(JSON.stringify({ challenge: message.challenge })) + } + const id = request.headers['webhook-id'] + const duplicate = seen.has(id) + seen.add(id) + log(duplicate ? '↺ duplicate delivery (retry)' : '★ EVENT', { + subscription: request.headers['x-mcp-subscription-id'], + userAgent: request.headers['user-agent'], + respondingWith: deliveryStatus, + event: message + }) + response.writeHead(deliveryStatus).end() +}) + +// --- cloudflared quick tunnel ----------------------------------------------- +function startTunnel() { + return new Promise((resolve, reject) => { + const child = spawn( + 'cloudflared', + ['tunnel', '--no-autoupdate', '--url', `http://127.0.0.1:${port}`], + { stdio: ['ignore', 'pipe', 'pipe'] } + ) + const timer = setTimeout(() => reject(new Error('cloudflared not ready in 45s')), 45_000) + let url + let registered = false + // Wait for both the URL and a registered edge connection. Probing the + // hostname earlier gets NXDOMAIN, which local resolvers then cache. + const onData = (chunk) => { + const text = String(chunk) + url ??= /https:\/\/[a-z0-9-]+\.trycloudflare\.com/.exec(text)?.[0] + registered ||= text.includes('Registered tunnel connection') + if (url && registered) { + clearTimeout(timer) + resolve({ child, url }) + } + } + child.stdout.on('data', onData) + child.stderr.on('data', onData) + child.on('error', reject) + }) +} + +async function waitUntilReachable(url) { + let lastError + await new Promise((done) => setTimeout(done, 3000)) + for (let attempt = 0; attempt < 60; attempt++) { + try { + const response = await fetch(url) + if (response.ok) return + lastError = `HTTP ${response.status}` + } catch (error) { + lastError = error.cause?.code ?? error.message + } + await new Promise((done) => setTimeout(done, 1000)) + } + throw new Error(`Tunnel ${url} never became reachable: ${lastError}`) +} + +// --- Run ---------------------------------------------------------------------- +await new Promise((done) => server.listen(port, '127.0.0.1', done)) +log(`receiver listening on http://127.0.0.1:${port}/hook`) +const tunnel = await startTunnel() +await waitUntilReachable(tunnel.url) +const callbackUrl = `${tunnel.url}/hook` +log('tunnel ready', callbackUrl) + +const discover = await rpc('server/discover') +log('server/discover capabilities.events =', discover.capabilities?.events ?? 'MISSING') +const events = [] +let cursor +do { + const page = await rpc('events/list', cursor ? { cursor } : {}) + events.push(...page.events) + cursor = page.nextCursor +} while (cursor && events.length === 0) +log( + 'events/list', + events.map((event) => event.name) +) +if (!events.some((event) => event.name === EVENT)) + throw new Error(`${EVENT} not offered to this token/account`) + +const subscribeParams = { + name: EVENT, + arguments: { account_id: accountId, afterOccurrences: 1, ...(worker ? { service: worker } : {}) }, + delivery: { mode: 'webhook', url: callbackUrl, secret } +} +let refreshTimer +async function subscribe(label) { + const result = await rpc('events/subscribe', subscribeParams) + log(label, result) + // Like ChatGPT: refresh a little before refreshBefore. + const delay = Math.max(Date.parse(result.refreshBefore) - Date.now() - 120_000, 30_000) + refreshTimer = setTimeout( + () => subscribe('↻ refreshed').catch((error) => log('✗ refresh failed', error.message)), + delay + ) +} +await subscribe('✓ subscribed') +log(`waiting for issues${worker ? ` on ${worker}` : ''}. Make the Worker throw. Ctrl-C to stop.`) + +let stopping = false +async function stop() { + if (stopping) return + stopping = true + clearTimeout(refreshTimer) + try { + const { secret: _secret, ...delivery } = subscribeParams.delivery + await rpc('events/unsubscribe', { ...subscribeParams, delivery }) + log('✓ unsubscribed') + } catch (error) { + log('✗ unsubscribe failed', error.message) + } + tunnel.child.kill() + server.close() + process.exit(0) +} +process.on('SIGINT', stop) +process.on('SIGTERM', stop) diff --git a/scripts/mcp-events-client.mjs b/scripts/mcp-events-client.mjs new file mode 100644 index 0000000..bffa3b0 --- /dev/null +++ b/scripts/mcp-events-client.mjs @@ -0,0 +1,53 @@ +const method = process.argv[2] ?? 'list' +if (!['list', 'subscribe', 'unsubscribe'].includes(method)) + throw new Error('Usage: node scripts/mcp-events-client.mjs list|subscribe|unsubscribe') +const token = process.env.CLOUDFLARE_API_TOKEN +if (!token) throw new Error('Set CLOUDFLARE_API_TOKEN') +const endpoint = process.env.MCP_URL ?? 'http://localhost:2529/mcp' +const params = { + _meta: { + 'io.modelcontextprotocol/protocolVersion': '2026-07-28', + 'io.modelcontextprotocol/clientInfo': { name: 'mcp-events-local-test', version: '1.0.0' }, + 'io.modelcontextprotocol/clientCapabilities': {} + } +} +if (method !== 'list') { + const accountId = process.env.CLOUDFLARE_ACCOUNT_ID + const callbackUrl = process.env.MCP_EVENT_CALLBACK_URL + if (!accountId || !callbackUrl) + throw new Error('Set CLOUDFLARE_ACCOUNT_ID and MCP_EVENT_CALLBACK_URL') + params.name = 'cloudflare.alert.workers_observability_real_time_issue' + params.arguments = { + account_id: accountId, + afterOccurrences: 1, + ...(process.env.MCP_EVENT_WORKER ? { service: process.env.MCP_EVENT_WORKER } : {}) + } + params.delivery = { mode: 'webhook', url: callbackUrl } + if (method === 'subscribe') { + if (!process.env.MCP_EVENT_SIGNING_SECRET) + throw new Error('Set MCP_EVENT_SIGNING_SECRET to the receiver’s key') + params.delivery.secret = process.env.MCP_EVENT_SIGNING_SECRET + } +} +const response = await fetch(endpoint, { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'MCP-Protocol-Version': '2026-07-28', + 'Mcp-Method': `events/${method}`, + ...(params.name ? { 'Mcp-Name': params.name } : {}) + }, + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: `events/${method}`, params }) +}) +const text = await response.text() +const data = response.headers.get('content-type')?.includes('text/event-stream') + ? text + .split('\n') + .find((line) => line.startsWith('data:')) + ?.slice(5) + : text +const result = JSON.parse(data) +console.log(JSON.stringify(result, null, 2)) +if (!response.ok || result.error) process.exitCode = 1 diff --git a/scripts/mcp-events-receiver.mjs b/scripts/mcp-events-receiver.mjs new file mode 100644 index 0000000..c3a5522 --- /dev/null +++ b/scripts/mcp-events-receiver.mjs @@ -0,0 +1,74 @@ +import { createServer } from 'node:http' +import { createHmac, timingSafeEqual } from 'node:crypto' + +const secret = process.env.MCP_EVENT_SIGNING_SECRET +if (!secret?.startsWith('whsec_')) + throw new Error('Set MCP_EVENT_SIGNING_SECRET to whsec_') +const key = Buffer.from(secret.slice(6), 'base64') +if (key.length < 24 || key.length > 64) throw new Error('Signing key must contain 24–64 bytes') +const port = Number(process.env.MCP_EVENT_RECEIVER_PORT ?? 8788) +const deliveryStatus = Number(process.env.MCP_EVENT_RESPONSE_STATUS ?? 204) +if (!Number.isInteger(deliveryStatus) || deliveryStatus < 200 || deliveryStatus > 599) + throw new Error('Invalid MCP_EVENT_RESPONSE_STATUS') + +createServer(async (request, response) => { + try { + if (request.method !== 'POST' || request.url !== '/events') { + response.writeHead(404).end() + return + } + const chunks = [] + let size = 0 + for await (const chunk of request) { + size += chunk.length + if (size > 256 * 1024) { + response.writeHead(413).end() + return + } + chunks.push(chunk) + } + const body = Buffer.concat(chunks).toString('utf8') + const id = request.headers['webhook-id'] + const timestamp = request.headers['webhook-timestamp'] + const signatures = String(request.headers['webhook-signature'] ?? '').split(' ') + const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${body}`).digest() + const valid = signatures.some((signature) => { + if (!signature.startsWith('v1,')) return false + const actual = Buffer.from(signature.slice(3), 'base64') + return actual.length === expected.length && timingSafeEqual(actual, expected) + }) + if ( + !id || + !/^\d+$/.test(timestamp ?? '') || + Math.abs(Date.now() / 1000 - Number(timestamp)) > 300 || + !valid + ) { + response.writeHead(401).end() + return + } + const event = JSON.parse(body) + if (event.type === 'verification') { + response + .writeHead(200, { 'Content-Type': 'application/json' }) + .end(JSON.stringify({ challenge: event.challenge })) + console.log('Callback verified') + return + } + if (event.eventId !== id) { + response.writeHead(400).end() + return + } + console.log( + JSON.stringify( + { subscription: request.headers['x-mcp-subscription-id'], status: deliveryStatus, event }, + null, + 2 + ) + ) + response.writeHead(deliveryStatus).end() + } catch { + response.writeHead(400).end() + } +}).listen(port, '127.0.0.1', () => + console.log(`MCP event receiver listening on http://127.0.0.1:${port}/events`) +) diff --git a/src/events/ans-bridge.ts b/src/events/ans-bridge.ts new file mode 100644 index 0000000..ca7003c --- /dev/null +++ b/src/events/ans-bridge.ts @@ -0,0 +1,567 @@ +import { z } from 'zod' + +export const MAX_EVENT_BYTES = 256 * 1024 +export const SUPPORTED_ALERT_TYPES: ReadonlySet = new Set([ + 'workers_observability_real_time_issue' +]) + +export const CallbackUrl = z + .string() + .max(2048) + .url() + .refine((value) => { + const url = new URL(value) + return ( + url.protocol === 'https:' && + !url.username && + !url.password && + !url.hash && + (!url.port || url.port === '443') && + url.hostname.includes('.') && + !/^[\d.]+$/.test(url.hostname) && + !url.hostname.includes(':') && + !/(^|\.)(localhost|local|internal|test|invalid|onion)\.?$/.test(url.hostname) + ) + }, 'Callback must use HTTPS on port 443 with a public hostname, without credentials or a fragment') + +export const AlertArguments = z + .object({ + account_id: z + .string() + .regex(/^[a-f0-9]{32}$/) + .describe('Cloudflare account to monitor'), + service: z + .string() + .trim() + .min(1) + .max(128) + .optional() + .describe('Worker name; omit to monitor the account'), + afterOccurrences: z + .number() + .int() + .min(1) + .max(Number.MAX_SAFE_INTEGER) + .optional() + .describe( + 'Notify when an issue reaches this occurrence count. Defaults to 1 (new issues). Do not combine with afterInactivitySeconds.' + ), + afterInactivitySeconds: z + .number() + .int() + .min(3600) + .max(365 * 86400) + .optional() + .describe( + 'Instead of afterOccurrences: notify when an issue recurs after this many inactive seconds' + ) + }) + .strict() + .refine( + (args) => args.afterOccurrences === undefined || args.afterInactivitySeconds === undefined, + 'Provide at most one of afterOccurrences or afterInactivitySeconds' + ) + +/** + * Event names this server offers. Unknown names are not invalid parameters; + * the methods answer them with `-32011 NotFound` (`data.kind: "event"`). + */ +const EventName = z.string().regex(/^cloudflare\.alert\.[a-z][a-z0-9_]*$/) + +/** `true` when the event name is in the catalogue this server implements. */ +export function isSupportedEventName(name: string): boolean { + return SUPPORTED_ALERT_TYPES.has(name.slice('cloudflare.alert.'.length)) +} + +const WebhookMode = z.literal('webhook') + +/** + * `events/subscribe` params. Follows the MCP Events draft: `cursor` and + * `maxAgeMs` are accepted and ignored because these events have no replay + * (`cursor: null`), and any `ttlMs` is clamped rather than rejected. + */ +export const SubscribeParams = z + .object({ + name: EventName, + arguments: AlertArguments, + delivery: z + .object({ + mode: WebhookMode, + url: CallbackUrl, + secret: z.string().refine((value) => { + try { + const key = decodeSecret(value) + return key.length >= 24 && key.length <= 64 + } catch { + return false + } + }, 'Expected whsec_ with a base64-encoded 24–64 byte key') + }) + .strict(), + cursor: z.string().nullable().optional(), + maxAgeMs: z.number().int().nonnegative().optional(), + ttlMs: z.number().nullable().optional() + }) + .strict() + +/** + * `events/unsubscribe` params. The subscription key is `(principal, url, name, + * arguments)`; the draft's example sends `delivery` as `{ url }` alone, so + * `mode` is optional here. + */ +export const UnsubscribeParams = z + .object({ + name: EventName, + arguments: AlertArguments, + delivery: z.object({ mode: WebhookMode.optional(), url: CallbackUrl }).strict() + }) + .strict() + +export const AvailableAlert = z.object({ + type: z.string().regex(/^[a-z][a-z0-9_]*$/), + display_name: z.string(), + description: z.string(), + filter_options: z + .array( + z.object({ + Key: z.string().min(1), + Range: z.string().optional(), + ComparisonOperator: z.string().optional(), + AvailableValues: z + .array(z.object({ ID: z.string(), Description: z.string() })) + .nullable() + .optional() + }) + ) + .nullish() + .transform((options) => options ?? []) +}) + +export const AlertPayload = z.object({ + name: z.string(), + text: z.string(), + // Absent from Workers issue alerts; checked only when present. + account_id: z.string().optional(), + account_name: z.string().optional(), + policy_id: z.string(), + policy_name: z.string().optional(), + alert_type: z.string(), + alert_correlation_id: z.string().optional(), + alert_event: z.string().optional(), + event_id: z.string().optional(), + severity: z.string().optional(), + ts: z.number().int().nonnegative(), + data: z.record(z.string(), z.unknown()).optional() +}) + +export type NotificationPayload = z.infer + +const IssueNotificationPayload = AlertPayload.extend({ + data: z + .object({ + issue: z + .object({ + id: z.string().min(1), + services: z.array(z.object({ name: z.string().min(1) }).passthrough()).min(1) + }) + .passthrough() + }) + .passthrough() +}) + +const EventPayloadSchemas: Readonly>> = { + workers_observability_real_time_issue: IssueNotificationPayload +} + +/** + * JSON Schema for an event descriptor, in the subset ChatGPT accepts for tool + * parameters: a plain object schema with no `$schema` and no top-level + * `oneOf`/`anyOf`/`not`. ChatGPT drops events whose schema it cannot use. + * The "at most one trigger" rule is enforced by `AlertArguments` instead. + */ +function descriptorSchema(schema: z.ZodType): Record { + const { $schema: _dialect, ...rest } = z.toJSONSchema(schema) + return rest +} + +export function alertEventDefinitions(catalog: unknown) { + const groups = z.record(z.string(), z.array(AvailableAlert)).parse(catalog) + return Object.values(groups) + .flat() + .filter((alert) => SUPPORTED_ALERT_TYPES.has(alert.type)) + .map((alert) => ({ + name: `cloudflare.alert.${alert.type}`, + description: `${alert.display_name}: ${alert.description}`, + delivery: ['webhook'], + inputSchema: descriptorSchema(AlertArguments), + payloadSchema: descriptorSchema(EventPayloadSchemas[alert.type]) + })) +} + +export interface Subscription { + id: string + principal: string + accountId: string + policyId: string + alertType: string + callbackUrl: string + signingSecret: string + previousSigningSecret?: string + previousSecretExpiresAt?: number + ingressSecret: string + expiresAt: number + active: boolean +} + +export interface BridgeDependencies { + loadSubscription(id: string): Promise + hasAccess(subscription: Subscription): Promise + deactivateSubscription(id: string): Promise + eventIdentity(subscription: Subscription, alert: NotificationPayload): Promise + webhookFetch: typeof fetch + now(): number +} + +function decodeSecret(secret: string): Uint8Array { + if (!secret.startsWith('whsec_')) throw new Error('Invalid signing key prefix') + const encoded = secret.slice(6) + if (!/^[A-Za-z0-9+/]+={0,2}$/.test(encoded)) throw new Error('Invalid base64 signing key') + const decoded = atob(encoded) + if (btoa(decoded).replace(/=+$/, '') !== encoded.replace(/=+$/, '')) { + throw new Error('Noncanonical base64 signing key') + } + return Uint8Array.from(decoded, (character) => character.charCodeAt(0)) +} + +async function digest(value: string): Promise { + const bytes = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(value)) + return Array.from(new Uint8Array(bytes), (byte) => byte.toString(16).padStart(2, '0')).join('') +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` + if (value !== null && typeof value === 'object') { + return `{${Object.entries(value) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, entry]) => `${JSON.stringify(key)}:${canonicalJson(entry)}`) + .join(',')}}` + } + return JSON.stringify(value) +} + +export async function subscriptionId(principal: string, input: unknown): Promise { + const subscription = SubscribeParams.safeParse(input) + const params = subscription.success ? subscription.data : UnsubscribeParams.parse(input) + return `sub_${await digest( + canonicalJson([principal, params.delivery.url, params.name, params.arguments]) + )}` +} + +const DEFAULT_TTL_MS = 30 * 60 * 1000 +const MIN_TTL_MS = 60 * 1000 +const MAX_TTL_MS = 60 * 60 * 1000 + +/** + * Grant a subscription lifetime from the client's `ttlMs` suggestion. + * + * The draft gives TTLs no rejection path: an omitted suggestion gets the + * default, `null` (no expiry) and anything too long are clamped to the + * maximum, and anything too short is clamped up to the floor. + * + * @param now - Current time in Unix milliseconds. + * @param ttlMs - The client's suggested lifetime. + * @returns The expiry time in Unix milliseconds. + */ +export function subscriptionExpiration(now: number, ttlMs?: number | null): number { + const suggested = ttlMs === undefined ? DEFAULT_TTL_MS : (ttlMs ?? MAX_TTL_MS) + const granted = Number.isFinite(suggested) ? suggested : MAX_TTL_MS + return now + Math.min(Math.max(granted, MIN_TTL_MS), MAX_TTL_MS) +} + +export async function constantTimeEqual(left: string, right: string): Promise { + const leftHash = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(left)) + const rightHash = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(right)) + const leftBytes = new Uint8Array(leftHash) + const rightBytes = new Uint8Array(rightHash) + let difference = 0 + for (let index = 0; index < leftBytes.length; index++) { + difference |= leftBytes[index] ^ rightBytes[index] + } + return difference === 0 +} + +export async function signedHeaders( + subscription: Subscription, + eventId: string, + body: string, + now: number +): Promise { + const timestamp = String(Math.floor(now / 1000)) + const secrets = [subscription.signingSecret] + if (subscription.previousSigningSecret && (subscription.previousSecretExpiresAt ?? 0) > now) { + secrets.push(subscription.previousSigningSecret) + } + const signatures = await Promise.all( + secrets.map(async (secret) => { + const bytes = decodeSecret(secret) + if (bytes.length < 24 || bytes.length > 64) throw new Error('Invalid signing key length') + const key = await crypto.subtle.importKey( + 'raw', + bytes.buffer as ArrayBuffer, + { name: 'HMAC', hash: 'SHA-256' }, + false, + ['sign'] + ) + const signature = await crypto.subtle.sign( + 'HMAC', + key, + new TextEncoder().encode(`${eventId}.${timestamp}.${body}`) + ) + return `v1,${btoa(String.fromCharCode(...new Uint8Array(signature)))}` + }) + ) + return new Headers({ + 'Content-Type': 'application/json', + 'webhook-id': eventId, + 'webhook-timestamp': timestamp, + 'webhook-signature': signatures.join(' '), + 'X-MCP-Subscription-Id': subscription.id + }) +} + +/** + * Why a callback endpoint failed verification. These are the draft's + * `lastError` categories; raw endpoint responses are never surfaced. + */ +export type CallbackFailureReason = + | 'connection_refused' + | 'timeout' + | 'tls_error' + | 'http_4xx' + | 'http_5xx' + | 'challenge_failed' + +/** Outcome of the callback verification handshake. */ +export type CallbackVerification = + | { readonly _tag: 'verified' } + | { readonly _tag: 'failed'; readonly reason: CallbackFailureReason } + +function transportFailureReason(error: unknown): CallbackFailureReason { + if ( + error instanceof DOMException && + (error.name === 'TimeoutError' || error.name === 'AbortError') + ) + return 'timeout' + const message = error instanceof Error ? error.message.toLowerCase() : '' + if (/tls|ssl|certificate/.test(message)) return 'tls_error' + return 'connection_refused' +} + +/** + * Run the draft's endpoint-verification handshake: POST a signed + * `verification` envelope and require the endpoint to echo the challenge in a + * 2xx body, compared in constant time. + * + * @param subscription - The subscription whose callback URL and secret to use. + * @param webhookFetch - Fetch restricted to public callback URLs. + * @param now - Current time in Unix milliseconds, used for the signature. + * @returns `verified`, or `failed` with the draft's failure category. + */ +export async function verifyCallback( + subscription: Subscription, + webhookFetch: typeof fetch, + now: number +): Promise { + if (!CallbackUrl.safeParse(subscription.callbackUrl).success) + return { _tag: 'failed', reason: 'connection_refused' } + const challenge = crypto.randomUUID() + const body = JSON.stringify({ type: 'verification', challenge }) + let response: Response + try { + response = await webhookFetch(subscription.callbackUrl, { + method: 'POST', + redirect: 'error', + signal: AbortSignal.timeout(10_000), + headers: await signedHeaders( + subscription, + `msg_verification_${crypto.randomUUID()}`, + body, + now + ), + body + }) + } catch (error) { + return { _tag: 'failed', reason: transportFailureReason(error) } + } + if (!response.ok) { + await response.body?.cancel().catch(() => {}) + if (response.status >= 500) return { _tag: 'failed', reason: 'http_5xx' } + if (response.status >= 400) return { _tag: 'failed', reason: 'http_4xx' } + // A 3xx under manual redirects: the endpoint did not accept the challenge. + return { _tag: 'failed', reason: 'challenge_failed' } + } + let echoed: unknown + try { + echoed = await readJson(response, 4096) + } catch { + return { _tag: 'failed', reason: 'challenge_failed' } + } + const result = z.object({ challenge: z.string() }).safeParse(echoed) + if (!result.success || !(await constantTimeEqual(challenge, result.data.challenge))) + return { _tag: 'failed', reason: 'challenge_failed' } + return { _tag: 'verified' } +} + +export async function readJson( + message: Request | Response, + limit = MAX_EVENT_BYTES +): Promise { + const reader = message.body?.getReader() + if (!reader) throw new Error('Missing JSON body') + const decoder = new TextDecoder() + let text = '' + let size = 0 + try { + while (true) { + const part = await reader.read() + if (part.done) break + size += part.value.length + if (size > limit) throw new Error('JSON body too large') + text += decoder.decode(part.value, { stream: true }) + } + return JSON.parse(text + decoder.decode()) + } finally { + await reader.cancel().catch(() => {}) + } +} + +/** Key names and value types of a JSON value, without the values themselves. */ +function payloadShape(value: unknown, depth = 0): unknown { + if (depth > 4) return '…' + if (Array.isArray(value)) return value.length ? [payloadShape(value[0], depth + 1)] : [] + if (value !== null && typeof value === 'object') + return Object.fromEntries( + Object.entries(value).map(([key, entry]) => [key, payloadShape(entry, depth + 1)]) + ) + return value === null ? 'null' : typeof value +} + +export async function forwardAnsWebhook( + request: Request, + id: string, + dependencies: BridgeDependencies +): Promise { + if (request.method !== 'POST') return new Response(null, { status: 405 }) + const subscription = await dependencies.loadSubscription(id) + if ( + !subscription || + !(await constantTimeEqual( + request.headers.get('cf-webhook-auth') ?? '', + subscription.ingressSecret + )) || + !subscription.ingressSecret + ) { + return new Response(null, { status: 401 }) + } + const now = dependencies.now() + if (!subscription.active || subscription.expiresAt <= now) { + return new Response(null, { status: 204 }) + } + try { + if (!(await dependencies.hasAccess(subscription))) { + await dependencies.deactivateSubscription(id) + return new Response(null, { status: 204 }) + } + } catch { + return new Response(null, { status: 503 }) + } + const reader = request.body?.getReader() + if (!reader) return new Response(null, { status: 400 }) + const chunks: Uint8Array[] = [] + let size = 0 + let part = await reader.read() + while (!part.done) { + size += part.value.length + if (size > MAX_EVENT_BYTES) { + await reader.cancel() + return new Response(null, { status: 413 }) + } + chunks.push(part.value) + part = await reader.read() + } + const bytes = new Uint8Array(size) + let offset = 0 + for (const chunk of chunks) { + bytes.set(chunk, offset) + offset += chunk.length + } + const input = new TextDecoder().decode(bytes) + let raw: unknown + try { + raw = JSON.parse(input) + } catch { + return new Response(null, { status: 400 }) + } + const parsed = (EventPayloadSchemas[subscription.alertType] ?? AlertPayload).safeParse(raw) + if (!parsed.success) { + // Shape and failing paths only: alert values can hold customer data. + console.log( + JSON.stringify({ + event: 'mcp_events_payload_rejected', + subscription: subscription.id, + issues: parsed.error.issues.map((issue) => ({ + path: issue.path.join('.'), + code: issue.code + })), + shape: payloadShape(raw) + }) + ) + return new Response(null, { status: 400 }) + } + const alert = parsed.data + if ( + !SUPPORTED_ALERT_TYPES.has(subscription.alertType) || + (alert.account_id !== undefined && alert.account_id !== subscription.accountId) || + alert.policy_id !== subscription.policyId || + alert.alert_type !== subscription.alertType + ) { + return new Response(null, { status: 403 }) + } + const occurrenceTime = new Date(alert.ts * 1000) + if (!Number.isFinite(occurrenceTime.getTime())) return new Response(null, { status: 400 }) + const identity = await dependencies.eventIdentity(subscription, alert) + if (!identity) return new Response(null, { status: 503 }) + const eventId = `evt_${await digest(JSON.stringify([subscription.id, identity]))}` + const body = JSON.stringify({ + eventId, + name: `cloudflare.alert.${subscription.alertType}`, + timestamp: occurrenceTime.toISOString(), + data: alert, + cursor: null + }) + if (new TextEncoder().encode(body).length > MAX_EVENT_BYTES) { + return new Response(null, { status: 413 }) + } + try { + CallbackUrl.parse(subscription.callbackUrl) + const response = await dependencies.webhookFetch(subscription.callbackUrl, { + method: 'POST', + redirect: 'error', + signal: AbortSignal.timeout(10_000), + headers: await signedHeaders(subscription, eventId, body, now), + body + }) + await response.body?.cancel().catch(() => {}) + if (response.ok) return new Response(null, { status: 204 }) + // The draft makes 410 and 413 non-retryable for this delivery only; the + // subscription itself stays active. Acknowledge so ANS does not retry. + if (response.status === 410 || response.status === 413) { + return new Response(null, { status: 204 }) + } + // Every other non-2xx (including redirects, which are never followed) is + // retried. ANS owns retries, with backoff and bounded attempts. + return new Response(null, { status: 503 }) + } catch { + return new Response(null, { status: 503 }) + } +} diff --git a/src/events/api.ts b/src/events/api.ts new file mode 100644 index 0000000..d662170 --- /dev/null +++ b/src/events/api.ts @@ -0,0 +1,113 @@ +import { z } from 'zod' +import { fetchWithRetry } from '../utils/fetch-retry' +import { readJson } from './ans-bridge' + +export class EventApiError extends Error { + constructor(readonly status: number) { + super( + status === 401 || status === 403 + ? 'Cloudflare permission required: allow Notifications Edit and Workers Observability Edit for the selected account.' + : status === 404 + ? 'Issue automations are unavailable for this account.' + : 'Cloudflare could not complete the subscription operation. Retry the request.' + ) + } +} + +const Envelope = z.object({ + success: z.boolean(), + result: z.unknown(), + result_info: z.object({ total_pages: z.number().optional() }).optional() +}) +const Resource = z.object({ id: z.string().min(1), name: z.string().optional() }) + +export class EventApi { + constructor( + readonly base: string, + readonly token: string, + readonly accountId: string, + readonly signal?: AbortSignal + ) {} + + async request(path: string, method = 'GET', body?: unknown) { + const response = await fetchWithRetry( + `${this.base}/accounts/${this.accountId}${path}`, + { + method, + redirect: 'manual', + signal: this.signal + ? AbortSignal.any([this.signal, AbortSignal.timeout(10_000)]) + : AbortSignal.timeout(10_000), + headers: { Authorization: `Bearer ${this.token}`, 'Content-Type': 'application/json' }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }) + }, + { maxRetries: 0, caller: 'mcp_events_api' } + ) + if (!response.ok) { + await response.body?.cancel() + throw new EventApiError(response.status) + } + const envelope = Envelope.parse(await readJson(response, 2 * 1024 * 1024)) + if (!envelope.success) throw new EventApiError(502) + return envelope + } + + async result(path: string, method = 'GET', body?: unknown): Promise { + return (await this.request(path, method, body)).result + } + + /** + * Find a managed resource by its exact name. + * + * ANS lists are paginated. The issue automations list is not, and rejects + * any `page` or `per_page` query with 400 (an account holds at most 50). + */ + async find(path: string, name: string, listing: Listing = PAGED): Promise { + for (let page = 1; page <= 100; page++) { + const envelope = await this.request( + listing.pagination === 'paged' ? `${path}?page=${page}&per_page=100` : path + ) + const value = listing.collection + ? z.record(z.string(), z.unknown()).parse(envelope.result)[listing.collection] + : envelope.result + const resources = z.array(Resource).parse(value) + const found = resources.find((resource) => resource.name === name) + if (resources.filter((resource) => resource.name === name).length > 1) + throw new EventApiError(409) + if (found) return found.id + if (listing.pagination === 'unpaged') return + if (!envelope.result_info?.total_pages || page >= envelope.result_info.total_pages) return + } + throw new EventApiError(503) + } + + async create(path: string, body: unknown, collection?: string): Promise { + const result = await this.result(path, 'POST', body) + return Resource.parse( + collection ? z.record(z.string(), z.unknown()).parse(result)[collection] : result + ).id + } + + async remove(path: string): Promise { + try { + await this.result(path, 'DELETE') + } catch (error) { + if (!(error instanceof EventApiError && error.status === 404)) throw error + } + } +} + +/** How a Cloudflare list endpoint pages its results and where it nests them. */ +export interface Listing { + readonly pagination: 'paged' | 'unpaged' + readonly collection?: string +} + +const PAGED: Listing = { pagination: 'paged' } + +/** The issue automations list: one unpaginated page under `result.automations`. */ +export const AUTOMATIONS_LISTING: Listing = { pagination: 'unpaged', collection: 'automations' } + +export const WEBHOOKS = '/alerting/v3/destinations/webhooks' +export const POLICIES = '/alerting/v3/policies' +export const AUTOMATIONS = '/workers/observability/issues/automations' diff --git a/src/events/callback.ts b/src/events/callback.ts new file mode 100644 index 0000000..3ecae6d --- /dev/null +++ b/src/events/callback.ts @@ -0,0 +1,115 @@ +import { EventApi, EventApiError, POLICIES, AUTOMATIONS } from './api' +import { constantTimeEqual, forwardAnsWebhook } from './ans-bridge' +import { ManagedPolicy, asSubscription, publicWebhookFetch, readPolicyState } from './service' +import { + deliveryAccessToken, + ingressSecret, + readDeliveryTicket, + type DeliveryTicket, + type EventEnvironment +} from './state' + +/** + * Why a callback from ANS was acknowledged, dropped or bounced. Logged once + * per callback with safe fields only, so silent 204s can be told apart. + */ +type CallbackOutcome = + | 'no_ticket' + | 'ticket_expired' + | 'credential_unavailable' + | 'access_denied' + | 'account_mismatch' + | 'forwarded' + | 'upstream_error' + +function logOutcome( + subscription: string, + outcome: CallbackOutcome, + status: number, + detail: Record = {} +): void { + console.log( + JSON.stringify({ event: 'mcp_events_callback', subscription, outcome, status, ...detail }) + ) +} + +/** + * Receive an ANS webhook for one subscription, re-check access and forward it + * to the subscriber as a signed MCP event. + */ +export async function handleAnsCallback( + request: Request, + env: EventEnvironment +): Promise { + const deadline = AbortSignal.timeout(4000) + if (request.method !== 'POST') return new Response(null, { status: 405 }) + const match = /^\/events\/ans\/(sub_[a-f0-9]{64})$/.exec(new URL(request.url).pathname) + if (!match) return new Response(null, { status: 404 }) + const id = match[1] + const secret = await ingressSecret(env, id) + if (!(await constantTimeEqual(request.headers.get('cf-webhook-auth') ?? '', secret))) + return new Response(null, { status: 401 }) + let ticket: DeliveryTicket | null + try { + ticket = await readDeliveryTicket(env, id) + } catch { + // Unreadable (e.g. the encryption key rotated): fail closed, don't retry. + return new Response(null, { status: 401 }) + } + // No ticket: ANS's creation probe, or the subscription expired or ended. + if (!ticket) { + logOutcome(id, 'no_ticket', 204) + return new Response(null, { status: 204 }) + } + if (ticket.expiresAt <= Date.now()) { + logOutcome(id, 'ticket_expired', 204, { expiredAt: new Date(ticket.expiresAt).toISOString() }) + return new Response(null, { status: 204 }) + } + try { + const accessToken = await deliveryAccessToken(env, ticket) + if (!accessToken) { + logOutcome(id, 'credential_unavailable', 204, { providerIssued: ticket.providerIssued }) + return new Response(null, { status: 204 }) + } + const api = new EventApi(env.CLOUDFLARE_API_BASE, accessToken, ticket.accountId, deadline) + const [policyData] = await Promise.all([ + api.result(`${POLICIES}/${encodeURIComponent(ticket.policyId)}`), + api.result(AUTOMATIONS) + ]) + const policy = ManagedPolicy.parse(policyData) + const state = await readPolicyState(env, policy, id) + if (state.params.arguments.account_id !== ticket.accountId) { + logOutcome(id, 'account_mismatch', 403) + return new Response(null, { status: 403 }) + } + const subscription = asSubscription(state, policy, secret) + const response = await forwardAnsWebhook(request, id, { + loadSubscription: async () => subscription, + hasAccess: async () => true, + eventIdentity: async (_subscription, payload) => payload.alert_correlation_id ?? '', + now: Date.now, + webhookFetch: (input, init) => publicWebhookFetch(input, { ...init, signal: deadline }), + deactivateSubscription: async () => { + const { id: _policyId, ...body } = policy + await api.result(`${POLICIES}/${encodeURIComponent(policy.id)}`, 'PUT', { + ...body, + enabled: false + }) + } + }) + logOutcome(id, 'forwarded', response.status, { + active: subscription.active, + expiresAt: new Date(subscription.expiresAt).toISOString() + }) + return response + } catch (error) { + if (error instanceof EventApiError && [401, 403, 404].includes(error.status)) { + logOutcome(id, 'access_denied', 204, { upstreamStatus: error.status }) + return new Response(null, { status: 204 }) + } + logOutcome(id, 'upstream_error', 503, { + error: error instanceof Error ? error.name : 'unknown' + }) + return new Response(null, { status: 503 }) + } +} diff --git a/src/events/methods.ts b/src/events/methods.ts new file mode 100644 index 0000000..1fdec80 --- /dev/null +++ b/src/events/methods.ts @@ -0,0 +1,102 @@ +import { ProtocolError, type McpServer } from '@modelcontextprotocol/server' +import { z } from 'zod' +import type { AuthProps } from '../auth/types' +import { EventApiError } from './api' +import { SubscribeParams, UnsubscribeParams } from './ans-bridge' +import { CallbackError, EventNotFound, EventService, SubscriptionForbidden } from './service' +import type { EventEnvironment } from './state' + +const Meta = z.record(z.string(), z.unknown()).optional() + +/** Error codes from the MCP Events draft (implementation-defined server range). */ +const NOT_FOUND = -32011 +const FORBIDDEN = -32012 +const CALLBACK_ENDPOINT_ERROR = -32015 +const INTERNAL_ERROR = -32603 + +/** Project service failures onto the draft's JSON-RPC error codes. */ +async function operation(run: () => Promise) { + try { + return await run() + } catch (error) { + if (error instanceof CallbackError) + throw new ProtocolError(CALLBACK_ENDPOINT_ERROR, 'CallbackEndpointError', { + reason: error.reason + }) + if (error instanceof EventNotFound) + throw new ProtocolError(NOT_FOUND, 'NotFound', { kind: 'event' }) + if (error instanceof SubscriptionForbidden) + throw new ProtocolError(FORBIDDEN, 'Forbidden', { reason: error.message }) + if (error instanceof EventApiError && (error.status === 401 || error.status === 403)) + throw new ProtocolError(FORBIDDEN, 'Forbidden', { reason: error.message }) + if (error instanceof EventApiError) + throw new ProtocolError(INTERNAL_ERROR, error.message, { status: error.status }) + throw new ProtocolError( + INTERNAL_ERROR, + 'Subscription operation failed. Retry with current credentials; partially created resources will be reused.' + ) + } +} + +/** + * Register the MCP Events webhook methods (`events/list`, `events/subscribe`, + * `events/unsubscribe`) and advertise the `events` capability. + * + * @param server - The per-request MCP server. + * @param env - Bindings used for subscription state and the Cloudflare API. + * @param props - The authenticated principal. + * @param bearer - The MCP bearer token for this request. + */ +export function registerEventMethods( + server: McpServer, + env: EventEnvironment, + props: AuthProps, + bearer: string +) { + const service = new EventService(env, props, bearer) + // The catalogue is fixed per deployment, so no notifications/events/list_changed. + // The SDK's capability type does not know the draft's `events` key yet. A + // named value skips the excess-property check; the SDK merges it as-is. + const capabilities = { ...server.server.getCapabilities(), events: { listChanged: false } } + server.server.registerCapabilities(capabilities) + server.server.setRequestHandler( + 'events/list', + { + params: z + .object({ + cursor: z + .string() + .regex(/^[1-9][0-9]{0,3}$/) + .optional(), + _meta: Meta + }) + .strict() + }, + async (params) => { + const result = await operation(() => service.list(params.cursor)) + console.log( + JSON.stringify({ + event: 'mcp_events_list', + principal: props.type, + events: result.events.length, + nextCursor: 'nextCursor' in result + }) + ) + return result + } + ) + server.server.setRequestHandler( + 'events/subscribe', + { + params: SubscribeParams.extend({ _meta: Meta }) + }, + ({ _meta, ...params }) => operation(() => service.subscribe(params)) + ) + server.server.setRequestHandler( + 'events/unsubscribe', + { + params: UnsubscribeParams.extend({ _meta: Meta }) + }, + ({ _meta, ...params }) => operation(() => service.unsubscribe(params)) + ) +} diff --git a/src/events/service.ts b/src/events/service.ts new file mode 100644 index 0000000..82d29f2 --- /dev/null +++ b/src/events/service.ts @@ -0,0 +1,361 @@ +import { z } from 'zod' +import type { AuthProps } from '../auth/types' +import { USER_AGENT } from '../constants' +import { fetchWithRetry } from '../utils/fetch-retry' +import { + EventApi, + EventApiError, + AUTOMATIONS, + AUTOMATIONS_LISTING, + POLICIES, + WEBHOOKS +} from './api' +import { + CallbackUrl, + SubscribeParams, + UnsubscribeParams, + alertEventDefinitions, + isSupportedEventName, + readJson, + subscriptionExpiration, + subscriptionId, + verifyCallback, + type CallbackFailureReason, + type Subscription +} from './ans-bridge' +import { + PolicyState, + credentialInfo, + decryptState, + deleteDeliveryTicket, + encryptState, + ingressSecret, + writeDeliveryTicket, + type EventEnvironment +} from './state' + +export const ManagedPolicy = z.object({ + id: z.string().min(1), + name: z.string(), + description: z.string().default(''), + enabled: z.boolean(), + alert_type: z.string(), + mechanisms: z.object({ webhooks: z.array(z.object({ id: z.string() })).default([]) }) +}) +export type ManagedPolicy = z.infer + +const mutations = new Map>() + +async function serialized(id: string, operation: () => Promise): Promise { + const pending = (mutations.get(id) ?? Promise.resolve()).catch(() => {}).then(operation) + mutations.set(id, pending) + try { + return await pending + } finally { + if (mutations.get(id) === pending) mutations.delete(id) + } +} + +/** The callback endpoint failed the verification handshake (`-32015`). */ +export class CallbackError extends Error { + readonly _tag = 'CallbackError' as const + + constructor(readonly reason: CallbackFailureReason) { + super('Callback endpoint verification failed') + } +} + +/** The event is not offered to this principal, or not offered at all (`-32011`). */ +export class EventNotFound extends Error { + readonly _tag = 'EventNotFound' as const + + constructor(readonly eventName: string) { + super(`Unknown event: ${eventName}`) + } +} + +/** The subscription key belongs to a different principal (`-32012`). */ +export class SubscriptionForbidden extends Error { + readonly _tag = 'SubscriptionForbidden' as const + + constructor() { + super('This subscription belongs to a different principal') + } +} + +/** + * Fetch for client callback URLs: re-checks the URL, never follows redirects, + * and identifies this server with the shared `User-Agent`. + */ +export const publicWebhookFetch: typeof fetch = async (input, init) => { + const url = input instanceof Request ? input.url : String(input) + CallbackUrl.parse(url) + const headers = new Headers(init?.headers) + headers.set('User-Agent', USER_AGENT) + return fetch(url, { ...init, headers, redirect: 'manual' }) +} + +export function asSubscription( + state: PolicyState, + policy: ManagedPolicy, + ingressSecret: string +): Subscription { + return { + id: state.id, + principal: state.principal, + accountId: state.params.arguments.account_id, + policyId: policy.id, + alertType: policy.alert_type, + callbackUrl: state.params.delivery.url, + signingSecret: state.params.delivery.secret, + ingressSecret, + expiresAt: state.expiresAt, + active: + policy.enabled && policy.mechanisms.webhooks.some((entry) => entry.id === state.webhookId), + previousSigningSecret: state.previousSigningSecret, + previousSecretExpiresAt: state.previousSecretExpiresAt + } +} + +export async function readPolicyState( + env: EventEnvironment, + policy: ManagedPolicy, + id: string +): Promise { + const state = PolicyState.parse(await decryptState(env, `policy:${id}`, policy.description)) + if ( + state.id !== id || + policy.name !== managedName(id) || + policy.alert_type !== state.params.name.slice('cloudflare.alert.'.length) + ) { + throw new Error('Policy subscription identity mismatch') + } + return state +} + +export function managedName(id: string) { + return `MCP Events ${id}` +} + +export class EventService { + private readonly principal: string + + constructor( + private readonly env: EventEnvironment, + private readonly props: AuthProps, + private readonly bearer: string + ) { + this.principal = + props.type === 'account_token' ? `account:${props.account.id}` : `user:${props.user.id}` + } + + private api(accountId: string) { + if (this.props.type === 'account_token' && this.props.account.id !== accountId) + throw new EventApiError(403) + return new EventApi(this.env.CLOUDFLARE_API_BASE, this.props.accessToken, accountId) + } + + async list(cursor?: string) { + let accounts: Array<{ id: string }> + let nextCursor: string | undefined + if (this.props.type === 'account_token') { + accounts = [this.props.account] + } else { + const page = Number(cursor ?? '1') + const response = await fetchWithRetry( + `${this.env.CLOUDFLARE_API_BASE}/accounts?page=${page}&per_page=20`, + { + headers: { Authorization: `Bearer ${this.props.accessToken}` }, + redirect: 'manual', + signal: AbortSignal.timeout(10_000) + }, + { maxRetries: 0, caller: 'mcp_events_account_discovery' } + ) + if (!response.ok) throw new EventApiError(response.status) + const value = z + .object({ + success: z.literal(true), + result: z.array(z.object({ id: z.string().regex(/^[a-f0-9]{32}$/) })), + result_info: z.object({ total_pages: z.number() }) + }) + .parse(await readJson(response)) + accounts = value.result + if (page < value.result_info.total_pages) nextCursor = String(page + 1) + } + for (const account of accounts) { + try { + const events = await this.available(account.id) + if (events.length) return { events } + } catch (error) { + if (!(error instanceof EventApiError && [401, 403, 404].includes(error.status))) throw error + } + } + return { events: [], ...(nextCursor ? { nextCursor } : {}) } + } + + private async available(accountId: string) { + const api = this.api(accountId) + const events = alertEventDefinitions(await api.result('/alerting/v3/available_alerts')) + if (events.length) await api.result(AUTOMATIONS) + return events + } + + async subscribe(input: unknown) { + const params = SubscribeParams.parse(input) + if (!isSupportedEventName(params.name)) throw new EventNotFound(params.name) + const id = await subscriptionId(`${this.env.MCP_RESOURCE}:${this.principal}`, params) + return serialized(id, async () => { + if ( + !(await this.available(params.arguments.account_id)).some( + (event) => event.name === params.name + ) + ) + throw new EventNotFound(params.name) + const api = this.api(params.arguments.account_id) + const name = managedName(id) + const policyId = await api.find(POLICIES, name) + const previousPolicy = policyId + ? ManagedPolicy.parse(await api.result(`${POLICIES}/${encodeURIComponent(policyId)}`)) + : undefined + const previous = previousPolicy + ? await readPolicyState(this.env, previousPolicy, id) + : undefined + if (previous && previous.principal !== this.principal) throw new SubscriptionForbidden() + const credential = await credentialInfo(this.env, this.bearer) + const now = Date.now() + const expiresAt = Math.min( + subscriptionExpiration(now, params.ttlMs), + credential.expiresAt - 60_000 + ) + if (expiresAt <= now) throw new Error('Refresh authentication before subscribing') + const state: PolicyState = { + version: 1, + id, + principal: this.principal, + params, + webhookId: previous?.webhookId ?? '', + expiresAt, + verifiedUntil: previous?.verifiedUntil ?? 0 + } + if ( + !previous || + previous.params.delivery.secret !== params.delivery.secret || + state.verifiedUntil <= now + ) { + const verification = await verifyCallback( + asSubscription( + state, + { + id: '', + name, + description: '', + enabled: false, + alert_type: params.name.slice('cloudflare.alert.'.length), + mechanisms: { webhooks: [] } + }, + '' + ), + publicWebhookFetch, + now + ) + if (verification._tag === 'failed') throw new CallbackError(verification.reason) + state.verifiedUntil = now + 5 * 60_000 + } + if (previous && previous.params.delivery.secret !== params.delivery.secret) { + state.previousSigningSecret = previous.params.delivery.secret + state.previousSecretExpiresAt = now + 5 * 60_000 + } else if (previous?.previousSigningSecret) { + state.previousSigningSecret = previous.previousSigningSecret + state.previousSecretExpiresAt = previous.previousSecretExpiresAt + } + const url = new URL(`/events/ans/${id}`, this.env.MCP_RESOURCE).href + CallbackUrl.parse(url) + // ANS keeps a destination's secret from creation and ignores updates, so + // the secret is a fixed per-subscription ingress credential. Everything + // that changes on refresh lives in the KV delivery ticket below. ANS + // probes the destination on creation; with no ticket yet, the callback + // acknowledges the probe without forwarding it. + state.webhookId = + (await api.find(WEBHOOKS, name)) ?? + (await api.create(WEBHOOKS, { name, url, secret: await ingressSecret(this.env, id) })) + const policyBody = { + name, + enabled: false, + alert_type: params.name.slice('cloudflare.alert.'.length), + description: await encryptState(this.env, `policy:${id}`, state), + mechanisms: { webhooks: [{ id: state.webhookId }] } + } + const resolvedPolicyId = policyId ?? (await api.create(POLICIES, policyBody)) + const { account_id: _accountId, ...options } = params.arguments + // With no trigger given, notify on new issues (first occurrence). + const trigger = + options.afterInactivitySeconds === undefined + ? { ...options, afterOccurrences: options.afterOccurrences ?? 1 } + : options + const automationBody = { ...trigger, name, policyId: resolvedPolicyId, enabled: false } + const automationId = + (await api.find(AUTOMATIONS, name, AUTOMATIONS_LISTING)) ?? + (await api.create(AUTOMATIONS, automationBody, 'automation')) + // Write the ticket before enabling delivery, so the first alert finds it. + // Refreshes rewrite it with the caller's current bearer and expiry. + await writeDeliveryTicket( + this.env, + { + id, + accountId: params.arguments.account_id, + policyId: resolvedPolicyId, + bearer: this.bearer, + providerIssued: credential.providerIssued, + expiresAt + }, + now + ) + await api.result(`${POLICIES}/${encodeURIComponent(resolvedPolicyId)}`, 'PUT', { + ...policyBody, + enabled: true + }) + await api.result(`${AUTOMATIONS}/${encodeURIComponent(automationId)}`, 'PUT', { + ...automationBody, + enabled: true + }) + return { + id, + refreshBefore: new Date(expiresAt).toISOString(), + cursor: null, + truncated: false + } + }) + } + + async unsubscribe(input: unknown) { + const params = UnsubscribeParams.parse(input) + if (!isSupportedEventName(params.name)) throw new EventNotFound(params.name) + 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. + await deleteDeliveryTicket(this.env, id) + const api = this.api(params.arguments.account_id) + const name = managedName(id) + const policyId = await api.find(POLICIES, name) + if (policyId) { + const policy = ManagedPolicy.parse( + await api.result(`${POLICIES}/${encodeURIComponent(policyId)}`) + ) + const state = await readPolicyState(this.env, policy, id) + if (state.principal !== this.principal) throw new SubscriptionForbidden() + const { id: _policyId, ...body } = policy + await api.result(`${POLICIES}/${encodeURIComponent(policyId)}`, 'PUT', { + ...body, + enabled: false + }) + } + const automationId = await api.find(AUTOMATIONS, name, AUTOMATIONS_LISTING) + if (automationId) await api.remove(`${AUTOMATIONS}/${encodeURIComponent(automationId)}`) + if (policyId) await api.remove(`${POLICIES}/${encodeURIComponent(policyId)}`) + const webhookId = await api.find(WEBHOOKS, name) + if (webhookId) await api.remove(`${WEBHOOKS}/${encodeURIComponent(webhookId)}`) + return {} + }) + } +} diff --git a/src/events/state.ts b/src/events/state.ts new file mode 100644 index 0000000..85cbb78 --- /dev/null +++ b/src/events/state.ts @@ -0,0 +1,195 @@ +import { getOAuthApi } from '@cloudflare/workers-oauth-provider' +import { z } from 'zod' +import { SubscribeParams } from './ans-bridge' + +export interface EventEnvironment { + OAUTH_KV: KVNamespace + MCP_COOKIE_ENCRYPTION_KEY: string + CLOUDFLARE_API_BASE: string + MCP_RESOURCE: string +} + +export const PolicyState = z.object({ + version: z.literal(1), + id: z.string().regex(/^sub_[a-f0-9]{64}$/), + principal: z.string(), + params: SubscribeParams, + webhookId: z.string().min(1), + expiresAt: z.number(), + verifiedUntil: z.number(), + previousSigningSecret: z.string().optional(), + previousSecretExpiresAt: z.number().optional() +}) +export type PolicyState = z.infer + +/** + * The credential and routing a callback needs to forward one subscription's + * alerts. Kept in KV, not in the ANS destination secret: ANS keeps a + * destination's secret from creation and ignores updates, while the bearer and + * expiry change on every refresh. + */ +export const DeliveryTicket = z.object({ + id: z.string().regex(/^sub_[a-f0-9]{64}$/), + accountId: z.string().regex(/^[a-f0-9]{32}$/), + policyId: z.string().min(1), + bearer: z.string().min(1), + providerIssued: z.boolean(), + expiresAt: z.number() +}) +export type DeliveryTicket = z.infer + +const TICKET_PREFIX = 'mcp-events:ticket:' +/** KV's minimum TTL; also covers clock skew between refresh and expiry. */ +const TICKET_GRACE_SECONDS = 60 + +/** + * Store a subscription's delivery ticket, encrypted, until it expires. + * + * @param env - Bindings holding OAUTH_KV and the encryption key. + * @param ticket - The ticket to store; replaces any previous one. + * @param now - Current time in Unix milliseconds. + */ +export async function writeDeliveryTicket( + env: EventEnvironment, + ticket: DeliveryTicket, + now: number +): Promise { + const ttl = Math.ceil((ticket.expiresAt - now) / 1000) + TICKET_GRACE_SECONDS + await env.OAUTH_KV.put( + `${TICKET_PREFIX}${ticket.id}`, + await encryptState(env, `delivery:${ticket.id}`, ticket), + { expirationTtl: Math.max(ttl, TICKET_GRACE_SECONDS) } + ) +} + +/** + * Read a subscription's delivery ticket. + * + * @returns The ticket, or `null` when none is stored (never created, expired + * or unsubscribed). + */ +export async function readDeliveryTicket( + env: EventEnvironment, + id: string +): Promise { + const stored = await env.OAUTH_KV.get(`${TICKET_PREFIX}${id}`) + if (stored === null) return null + const ticket = DeliveryTicket.parse(await decryptState(env, `delivery:${id}`, stored)) + if (ticket.id !== id) throw new Error('Delivery ticket identity mismatch') + return ticket +} + +/** Delete a subscription's delivery ticket so its callbacks stop forwarding. */ +export async function deleteDeliveryTicket(env: EventEnvironment, id: string): Promise { + await env.OAUTH_KV.delete(`${TICKET_PREFIX}${id}`) +} + +/** + * The secret ANS sends in `cf-webhook-auth` for one subscription's destination. + * + * It never changes for a subscription, so it survives ANS ignoring secret + * updates. It only proves the caller is ANS for this subscription; the + * credential to act on lives in the delivery ticket. + */ +export async function ingressSecret(env: EventEnvironment, id: string): Promise { + const key = await crypto.subtle.importKey( + 'raw', + await crypto.subtle.digest( + 'SHA-256', + new TextEncoder().encode(`mcp-events:ingress:v1:${env.MCP_COOKIE_ENCRYPTION_KEY}`) + ), + { name: 'HMAC', hash: 'SHA-256' }, + false, + ['sign'] + ) + const mac = await crypto.subtle.sign( + 'HMAC', + key, + new TextEncoder().encode(`${env.MCP_RESOURCE}:${id}`) + ) + return `mcp-events-ingress-v1:${btoa(String.fromCharCode(...new Uint8Array(mac)))}` +} + +function oauth(env: EventEnvironment) { + return getOAuthApi( + { + apiRoute: '/mcp', + apiHandler: { fetch: () => new Response(null, { status: 404 }) }, + defaultHandler: { fetch: () => new Response(null, { status: 404 }) }, + authorizeEndpoint: '/authorize', + tokenEndpoint: '/token', + resourceMetadata: { resource: env.MCP_RESOURCE } + }, + env + ) +} + +export async function credentialInfo(env: EventEnvironment, bearer: string) { + const token = await oauth(env).unwrapToken(bearer) + return { providerIssued: token !== null, expiresAt: token ? token.expiresAt * 1000 : Infinity } +} + +export async function deliveryAccessToken( + env: EventEnvironment, + ticket: DeliveryTicket +): Promise { + if (!ticket.providerIssued) return ticket.bearer + const token = await oauth(env).unwrapToken(ticket.bearer) + if (!token || token.expiresAt * 1000 <= Date.now()) return null + return z.object({ accessToken: z.string().min(1) }).parse(token.grant.props).accessToken +} + +async function key(env: EventEnvironment) { + if (env.MCP_COOKIE_ENCRYPTION_KEY.length < 32) + throw new Error('MCP_COOKIE_ENCRYPTION_KEY must be at least 32 characters') + const digest = await crypto.subtle.digest( + 'SHA-256', + new TextEncoder().encode(`mcp-events:v1:${env.MCP_COOKIE_ENCRYPTION_KEY}`) + ) + return crypto.subtle.importKey('raw', digest, 'AES-GCM', false, ['encrypt', 'decrypt']) +} + +export async function encryptState( + env: EventEnvironment, + purpose: string, + value: unknown +): Promise { + const nonce = crypto.getRandomValues(new Uint8Array(12)) + const ciphertext = await crypto.subtle.encrypt( + { + name: 'AES-GCM', + iv: nonce, + additionalData: new TextEncoder().encode(`${env.MCP_RESOURCE}:${purpose}`) + }, + await key(env), + new TextEncoder().encode(JSON.stringify(value)) + ) + const bytes = new Uint8Array(12 + ciphertext.byteLength) + bytes.set(nonce) + bytes.set(new Uint8Array(ciphertext), 12) + const result = `mcp-events-v1:${btoa(String.fromCharCode(...bytes))}` + if (result.length > 16_384) throw new Error('Event state exceeds size limit') + return result +} + +export async function decryptState( + env: EventEnvironment, + purpose: string, + value: string +): Promise { + if (!value.startsWith('mcp-events-v1:') || value.length > 16_384) + throw new Error('Invalid event state') + const bytes = Uint8Array.from(atob(value.slice('mcp-events-v1:'.length)), (character) => + character.charCodeAt(0) + ) + const plaintext = await crypto.subtle.decrypt( + { + name: 'AES-GCM', + iv: bytes.slice(0, 12), + additionalData: new TextEncoder().encode(`${env.MCP_RESOURCE}:${purpose}`) + }, + await key(env), + bytes.slice(12) + ) + return JSON.parse(new TextDecoder().decode(plaintext)) +} diff --git a/src/index.ts b/src/index.ts index a15de36..4d895ac 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,5 @@ import OAuthProvider from '@cloudflare/workers-oauth-provider' +import { handleAnsCallback } from './events/callback' import { env as workerEnv } from 'cloudflare:workers' import { createAuthHandlers, handleTokenExchangeCallback } from './auth/oauth-handler' import { ALL_SCOPES, REQUIRED_SCOPES } from './auth/scopes' @@ -61,6 +62,7 @@ const oauthProvider = new OAuthProvider({ export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { const url = new URL(request.url) + if (url.pathname.startsWith('/events/ans/')) return handleAnsCallback(request, env) if (url.pathname === OPENAI_APPS_CHALLENGE_PATH) { return new Response(OPENAI_APPS_CHALLENGE_TOKEN, { headers: { 'Content-Type': 'text/plain; charset=utf-8' } diff --git a/src/mcp-handler.ts b/src/mcp-handler.ts index 5fa0dab..f5a6a21 100644 --- a/src/mcp-handler.ts +++ b/src/mcp-handler.ts @@ -34,13 +34,17 @@ function serverOptionsFromUrl(url: string): ServerOptions { } } -function createAuthenticatedHandler(props: AuthProps) { +function createAuthenticatedHandler(props: AuthProps, env?: Env, bearer = '') { return createMcpHandler(({ requestInfo }) => { if (!requestInfo) { throw new Error('The Cloudflare MCP server requires an HTTP request') } - return createServer(props, serverOptionsFromUrl(requestInfo.url)) + return createServer( + props, + serverOptionsFromUrl(requestInfo.url), + env ? { env, bearer } : undefined + ) }) } @@ -94,7 +98,8 @@ export function handleMcpPreflight(request: Request): Response { /** Serve one authenticated MCP exchange with a fresh SDK v2 server instance. */ export async function handleAuthenticatedMcpRequest( request: Request, - rawProps: unknown + rawProps: unknown, + env?: Env ): Promise { if (new URL(request.url).pathname !== MCP_ROUTE) { return new Response('Not Found', { status: 404 }) @@ -104,7 +109,11 @@ export async function handleAuthenticatedMcpRequest( if (rejected) return rejected const props = AuthPropsSchema.parse(rawProps) - const handler = createAuthenticatedHandler(props) + const handler = createAuthenticatedHandler( + props, + env, + request.headers.get('Authorization')?.replace(/^Bearer\s+/i, '') ?? '' + ) const response = await handler.fetch(request) return withCors(response, request) @@ -112,7 +121,7 @@ export async function handleAuthenticatedMcpRequest( /** ExportedHandler adapter required by workers-oauth-provider 0.8.x. */ export const oauthMcpHandler = { - fetch(request: Request, _env: Env, ctx: ExecutionContext) { - return handleAuthenticatedMcpRequest(request, ctx.props) + fetch(request: Request, env: Env, ctx: ExecutionContext) { + return handleAuthenticatedMcpRequest(request, ctx.props, env) } } diff --git a/src/server.ts b/src/server.ts index 257f559..a64760c 100644 --- a/src/server.ts +++ b/src/server.ts @@ -9,6 +9,8 @@ import { attachMetrics } from './metrics' import { SERVER_INFO } from './constants' import { stringifyResponse, truncateResponse } from './truncate' import type { AuthProps } from './auth/types' +import { registerEventMethods } from './events/methods' +import type { EventEnvironment } from './events/state' /** Per-request options the client picks through the MCP URL query string. */ export interface ServerOptions { @@ -34,7 +36,8 @@ export interface ServerOptions { */ export async function createServer( props: AuthProps, - { codemode = true, truncateToolResult = true }: ServerOptions = {} + { codemode = true, truncateToolResult = true }: ServerOptions = {}, + events?: { env: EventEnvironment; bearer: string } ): Promise { const server = new McpServer(SERVER_INFO) // The tool set is fixed for the life of each per-request server, so this @@ -43,6 +46,7 @@ export async function createServer( server.server.registerCapabilities({ tools: { listChanged: false } }) // Skills (io.modelcontextprotocol/skills) are served in both tool modes. registerSkills(server) + if (events) registerEventMethods(server, events.env, props, events.bearer) const formatResult = truncateToolResult ? truncateResponse : stringifyResponse if (!codemode) { diff --git a/tests/ans-bridge.test.ts b/tests/ans-bridge.test.ts new file mode 100644 index 0000000..1513947 --- /dev/null +++ b/tests/ans-bridge.test.ts @@ -0,0 +1,420 @@ +import { describe, expect, it, vi } from 'vitest' +import { + MAX_EVENT_BYTES, + alertEventDefinitions, + SubscribeParams, + UnsubscribeParams, + forwardAnsWebhook, + isSupportedEventName, + signedHeaders, + subscriptionExpiration, + subscriptionId, + verifyCallback, + type BridgeDependencies, + type Subscription +} from '../src/events/ans-bridge' + +const accountId = 'a'.repeat(32) +const secret = `whsec_${btoa('a'.repeat(32))}` +const subscription: Subscription = { + id: 'sub_test', + principal: 'user:123', + accountId, + policyId: 'policy-1', + alertType: 'workers_observability_real_time_issue', + callbackUrl: 'https://receiver.example/callback', + signingSecret: secret, + ingressSecret: 'ans-only-secret', + expiresAt: 200_000, + active: true +} +const params = { + name: 'cloudflare.alert.workers_observability_real_time_issue', + arguments: { account_id: accountId, service: 'my-worker', afterOccurrences: 1 }, + delivery: { mode: 'webhook', url: subscription.callbackUrl, secret } +} +const alert = { + name: 'ANS notification', + text: 'Untrusted notification text', + account_id: accountId, + policy_id: 'policy-1', + alert_type: 'workers_observability_real_time_issue', + alert_correlation_id: 'persisted-vega-run-id', + ts: 100, + data: { + issue: { id: 'issue-1', services: [{ name: 'my-worker' }], title: 'Untrusted issue text' } + } +} + +function request(value: unknown = alert, token = subscription.ingressSecret) { + return new Request('https://mcp.example/events/ans/sub_test', { + method: 'POST', + headers: { 'cf-webhook-auth': token }, + body: JSON.stringify(value) + }) +} + +function dependencies(status = 204): BridgeDependencies { + return { + loadSubscription: vi.fn(async () => ({ ...subscription })), + hasAccess: vi.fn(async () => true), + deactivateSubscription: vi.fn(async () => {}), + eventIdentity: vi.fn(async (_subscription, payload) => payload.alert_correlation_id ?? ''), + webhookFetch: vi.fn(async () => new Response(null, { status })) as unknown as typeof fetch, + now: () => 100_000 + } +} + +describe('MCP event contract', () => { + it('validates subscription arguments and signing keys', () => { + expect(SubscribeParams.parse(params)).toEqual(params) + for (const value of ['bad', 'whsec_!!!!', `whsec_${btoa('short')}`]) { + expect( + SubscribeParams.safeParse({ ...params, delivery: { ...params.delivery, secret: value } }) + .success + ).toBe(false) + } + }) + + it.each([ + 'http://receiver.example', + 'https://user:pass@receiver.example', + 'https://receiver.example/#fragment', + 'https://127.0.0.1/events', + 'https://2130706433/events', + 'https://[::1]/events', + 'https://service.internal/events', + 'https://service.local./events', + 'https://receiver.example:8080/events' + ])('rejects an invalid callback %s', (url) => { + expect( + SubscribeParams.safeParse({ ...params, delivery: { ...params.delivery, url } }).success + ).toBe(false) + }) + + it('keeps identity stable across argument ordering, refresh and key rotation', async () => { + const id = await subscriptionId('user:123', params) + expect( + await subscriptionId('user:123', { + ...params, + ttlMs: 1000, + arguments: { afterOccurrences: 1, service: 'my-worker', account_id: accountId }, + delivery: { ...params.delivery, secret: `whsec_${btoa('b'.repeat(32))}` } + }) + ).toBe(id) + expect(await subscriptionId('user:456', params)).not.toBe(id) + expect( + await subscriptionId('user:123', { + ...params, + delivery: { mode: 'webhook', url: params.delivery.url } + }) + ).toBe(id) + }) + + it('grants a finite bounded lifetime, clamping instead of rejecting', () => { + expect(subscriptionExpiration(100, 600_000)).toBe(600_100) + expect(subscriptionExpiration(100)).toBe(1_800_100) + expect(subscriptionExpiration(100, null)).toBe(3_600_100) + expect(subscriptionExpiration(100, 172_800_000)).toBe(3_600_100) + expect(subscriptionExpiration(100, 1000)).toBe(60_100) + expect(subscriptionExpiration(100, -1)).toBe(60_100) + expect(subscriptionExpiration(100, Number.POSITIVE_INFINITY)).toBe(3_600_100) + }) + + it('accepts the draft fields for events without replay', () => { + expect( + SubscribeParams.safeParse({ ...params, cursor: 'opaque', maxAgeMs: 300_000, ttlMs: 0 }) + .success + ).toBe(true) + expect( + UnsubscribeParams.safeParse({ + name: params.name, + arguments: params.arguments, + delivery: { url: params.delivery.url } + }).success + ).toBe(true) + }) + + it('matches an independently calculated Standard Webhooks signature', async () => { + const body = '{"eventId":"evt_1"}' + const headers = await signedHeaders(subscription, 'evt_1', body, 100_000) + const expected = 'Uk++cO75SKUwxzWywfoTryzRPN77PsDiIixiMPVNwdg=' + expect(headers.get('webhook-signature')).toBe(`v1,${expected}`) + expect(headers.get('webhook-id')).toBe('evt_1') + expect(headers.get('X-MCP-Subscription-Id')).toBe(subscription.id) + }) + + it('supports a bounded dual-signature rotation window', async () => { + const rotated = { + ...subscription, + previousSigningSecret: `whsec_${btoa('b'.repeat(32))}`, + previousSecretExpiresAt: 101_000 + } + expect( + (await signedHeaders(rotated, 'evt_1', '{}', 100_000)).get('webhook-signature')?.split(' ') + ).toHaveLength(2) + expect( + (await signedHeaders(rotated, 'evt_1', '{}', 102_000)).get('webhook-signature')?.split(' ') + ).toHaveLength(1) + }) + + it('verifies a fresh callback challenge with redirect blocking', async () => { + const transport = vi.fn(async (_url: unknown, init: RequestInit) => { + expect(init.redirect).toBe('error') + const body = JSON.parse(init.body as string) + expect(body.type).toBe('verification') + expect(new Headers(init.headers).get('webhook-id')).toMatch(/^msg_verification_/) + return Response.json({ challenge: body.challenge }) + }) + await expect( + verifyCallback(subscription, transport as unknown as typeof fetch, 100_000) + ).resolves.toEqual({ _tag: 'verified' }) + }) + + it('rejects a callback that does not echo the challenge', async () => { + const transport = vi.fn(async () => Response.json({ challenge: 'wrong' })) + await expect( + verifyCallback(subscription, transport as unknown as typeof fetch, 100_000) + ).resolves.toEqual({ _tag: 'failed', reason: 'challenge_failed' }) + }) + + it.each([ + [async () => new Response(null, { status: 404 }), 'http_4xx'], + [async () => new Response(null, { status: 502 }), 'http_5xx'], + [ + async () => { + throw new DOMException('timed out', 'TimeoutError') + }, + 'timeout' + ], + [ + async () => { + throw new TypeError('Network connection lost') + }, + 'connection_refused' + ] + ])('categorizes callback verification failures', async (transport, reason) => { + await expect( + verifyCallback(subscription, transport as unknown as typeof fetch, 100_000) + ).resolves.toEqual({ _tag: 'failed', reason }) + }) +}) + +describe('ANS synchronous delivery bridge', () => { + it('advertises only implemented events that are eligible in the ANS catalog', () => { + const events = alertEventDefinitions({ + Workers: [ + { + type: 'workers_observability_real_time_issue', + display_name: 'Worker Issue', + description: 'An issue', + filter_options: [] + } + ], + Origin: [ + { + type: 'http_alert_origin_error', + display_name: 'Origin Error', + description: 'Origin errors', + filter_options: [{ Key: 'zones', Range: '1-n', AvailableValues: null }] + } + ] + }) + expect(events.map((event) => event.name)).toEqual([ + 'cloudflare.alert.workers_observability_real_time_issue' + ]) + expect( + alertEventDefinitions({ + Origin: [ + { + type: 'http_alert_origin_error', + display_name: 'Origin Error', + description: 'Origin errors', + filter_options: [] + } + ] + }) + ).toEqual([]) + }) + + it('does not accept subscriptions to events without an implemented contract', async () => { + const deps = dependencies() + deps.loadSubscription = vi.fn(async () => ({ + ...subscription, + alertType: 'http_alert_origin_error' + })) + const notification = { + ...alert, + alert_type: 'http_alert_origin_error', + data: { zone_id: 'zone-1' } + } + expect((await forwardAnsWebhook(request(notification), subscription.id, deps)).status).toBe(403) + expect(deps.webhookFetch).not.toHaveBeenCalled() + expect(isSupportedEventName('cloudflare.alert.http_alert_origin_error')).toBe(false) + expect(isSupportedEventName(params.name)).toBe(true) + }) + + it.each([ + [204, 204], + [202, 204], + [408, 503], + [429, 503], + [500, 503], + [502, 503], + [302, 503], + [401, 503], + [404, 503], + [410, 204], + [413, 204] + ])('maps callback status %i to ANS status %i', async (callbackStatus, ansStatus) => { + expect( + (await forwardAnsWebhook(request(), subscription.id, dependencies(callbackStatus))).status + ).toBe(ansStatus) + }) + + it('waits for callback acceptance before acknowledging ANS', async () => { + const deps = dependencies() + let resolve: (response: Response) => void = () => {} + deps.webhookFetch = vi.fn( + () => + new Promise((done) => { + resolve = done + }) + ) as unknown as typeof fetch + const delivery = forwardAnsWebhook(request(), subscription.id, deps) + await vi.waitFor(() => expect(deps.webhookFetch).toHaveBeenCalledOnce()) + let acknowledged = false + void delivery.then(() => { + acknowledged = true + }) + expect(acknowledged).toBe(false) + resolve(new Response(null, { status: 204 })) + expect((await delivery).status).toBe(204) + }) + + it('preserves event ID and body across retries while refreshing signatures', async () => { + const deps = dependencies(503) + await forwardAnsWebhook(request(), subscription.id, deps) + deps.now = () => 110_000 + await forwardAnsWebhook(request(), subscription.id, deps) + const calls = vi.mocked(deps.webhookFetch).mock.calls + const first = calls[0][1]! + const second = calls[1][1]! + expect(first.body).toBe(second.body) + const firstHeaders = new Headers(first.headers) + const secondHeaders = new Headers(second.headers) + expect(firstHeaders.get('webhook-id')).toBe(secondHeaders.get('webhook-id')) + expect(firstHeaders.get('webhook-signature')).not.toBe(secondHeaders.get('webhook-signature')) + expect(JSON.parse(first.body as string).timestamp).toBe('1970-01-01T00:01:40.000Z') + expect(first.redirect).toBe('error') + }) + + it('forwards a real Workers issue alert, which carries no account_id', async () => { + const { account_id: _account, ...issueAlert } = alert + const deps = dependencies() + expect((await forwardAnsWebhook(request(issueAlert), subscription.id, deps)).status).toBe(204) + expect(deps.webhookFetch).toHaveBeenCalledOnce() + }) + + it('requires the ANS ingress secret and account/policy/alert-type scope', async () => { + const deps = dependencies() + expect((await forwardAnsWebhook(request(alert, 'wrong'), subscription.id, deps)).status).toBe( + 401 + ) + for (const invalid of [ + { ...alert, account_id: 'b'.repeat(32) }, + { ...alert, policy_id: 'other-policy' }, + { ...alert, alert_type: 'other-alert' } + ]) { + expect((await forwardAnsWebhook(request(invalid), subscription.id, deps)).status).toBe(403) + } + expect(deps.webhookFetch).not.toHaveBeenCalled() + }) + + it('never uses the resource ID as the delivery identity', async () => { + const deps = dependencies() + expect( + ( + await forwardAnsWebhook( + request({ ...alert, alert_correlation_id: '' }), + subscription.id, + deps + ) + ).status + ).toBe(503) + expect(deps.webhookFetch).not.toHaveBeenCalled() + }) + + it('rejects notifications that do not match the issue payload contract', async () => { + const deps = dependencies() + expect( + ( + await forwardAnsWebhook( + request({ ...alert, data: { unrelated: true } }), + subscription.id, + deps + ) + ).status + ).toBe(400) + expect(deps.webhookFetch).not.toHaveBeenCalled() + }) + + it('does not send expired or inactive subscriptions', async () => { + for (const record of [ + { ...subscription, active: false }, + { ...subscription, expiresAt: 99_000 } + ]) { + const deps = dependencies() + deps.loadSubscription = vi.fn(async () => record) + expect((await forwardAnsWebhook(request(), subscription.id, deps)).status).toBe(204) + expect(deps.webhookFetch).not.toHaveBeenCalled() + } + }) + + it('deactivates subscriptions on revoked access, but not on a callback 410', async () => { + const revoked = dependencies() + revoked.hasAccess = vi.fn(async () => false) + expect((await forwardAnsWebhook(request(), subscription.id, revoked)).status).toBe(204) + expect(revoked.webhookFetch).not.toHaveBeenCalled() + expect(revoked.deactivateSubscription).toHaveBeenCalledWith(subscription.id) + const gone = dependencies(410) + expect((await forwardAnsWebhook(request(), subscription.id, gone)).status).toBe(204) + expect(gone.deactivateSubscription).not.toHaveBeenCalled() + }) + + it('returns a retryable status for network failures', async () => { + const deps = dependencies() + deps.webhookFetch = vi.fn(async () => { + throw new Error('timeout') + }) as unknown as typeof fetch + expect((await forwardAnsWebhook(request(), subscription.id, deps)).status).toBe(503) + expect(deps.webhookFetch).toHaveBeenCalledOnce() + }) + + it('makes one delivery attempt and leaves retries to ANS', async () => { + const deps = dependencies(503) + expect((await forwardAnsWebhook(request(), subscription.id, deps)).status).toBe(503) + expect(deps.webhookFetch).toHaveBeenCalledOnce() + expect(deps.deactivateSubscription).not.toHaveBeenCalled() + }) + + it('leaves transient access-check failures to ANS without deactivating', async () => { + const deps = dependencies() + deps.hasAccess = vi.fn(async () => { + throw new Error('Service unavailable') + }) + expect((await forwardAnsWebhook(request(), subscription.id, deps)).status).toBe(503) + expect(deps.webhookFetch).not.toHaveBeenCalled() + expect(deps.deactivateSubscription).not.toHaveBeenCalled() + }) + + it('bounds inbound and outgoing event bodies', async () => { + const deps = dependencies() + const oversized = { + ...alert, + data: { issue: { ...alert.data.issue, title: 'x'.repeat(MAX_EVENT_BYTES) } } + } + expect((await forwardAnsWebhook(request(oversized), subscription.id, deps)).status).toBe(413) + expect(deps.webhookFetch).not.toHaveBeenCalled() + }) +}) diff --git a/tests/events-e2e.test.ts b/tests/events-e2e.test.ts new file mode 100644 index 0000000..3342e6e --- /dev/null +++ b/tests/events-e2e.test.ts @@ -0,0 +1,600 @@ +import { env, exports } from 'cloudflare:workers' +import { getOAuthApi } from '@cloudflare/workers-oauth-provider' +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 { API_BASE, cfSuccess, mockIdentityProbe } from './helpers/cloudflare-api' +import { clearKv } from './helpers/kv' +import { modernMcpRequest } from './helpers/mcp' +import { clearSpec, seedSpec } from './helpers/spec' +import { server } from './setup/msw' + +const accountId = 'a'.repeat(32) +const token = 'cfat_events-local-test-token' +const callbackUrl = 'https://receiver.example.com/events' +const signingSecret = `whsec_${btoa('a'.repeat(32))}` +const params = { + name: 'cloudflare.alert.workers_observability_real_time_issue', + arguments: { account_id: accountId, service: 'checkout', afterOccurrences: 1 }, + delivery: { mode: 'webhook', url: callbackUrl, secret: signingSecret } +} + +interface Resource { + id: string + name: string + [key: string]: unknown +} +interface RpcResponse { + result?: { + id?: string + refreshBefore?: string + capabilities?: { events?: object; extensions?: Record } + events?: unknown[] + } + error?: { code: number; data?: Record } +} +const resources = new Map>() +const requests: Array<{ method: string; path: string }> = [] +const deliveries: Array<{ body: string; headers: Headers }> = [] +let callbackStatus = 204 +let rejectVerification = false +let failNext = '' +let policyReadStatus = 200 +let verificationCount = 0 +let bootstrapCount = 0 +let validKeys = [signingSecret] + +function records(path: string) { + return resources.get(path)! +} + +async function rpc( + method: string, + input: Record = {}, + bearer = token +): Promise { + const response = await exports.default.fetch(modernMcpRequest(bearer, method, input)) + const text = await response.text() + const body = + text.startsWith('event:') || text.startsWith('data:') + ? text + .split('\n') + .find((line) => line.startsWith('data:'))! + .slice(5) + : text + return JSON.parse(body) +} + +async function checkSignature(request: { headers: Headers }, body: string) { + const signatures = request.headers.get('webhook-signature')!.split(' ') + const signed = `${request.headers.get('webhook-id')}.${request.headers.get('webhook-timestamp')}.${body}` + for (const secret of validKeys) { + const key = await crypto.subtle.importKey( + 'raw', + Uint8Array.from(atob(secret.slice(6)), (character) => character.charCodeAt(0)), + { name: 'HMAC', hash: 'SHA-256' }, + false, + ['sign'] + ) + const signature = await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(signed)) + if (signatures.includes(`v1,${btoa(String.fromCharCode(...new Uint8Array(signature)))}`)) return + } + throw new Error('Receiver rejected webhook signature') +} + +function ansRequest(destination = [...records(WEBHOOKS).values()][0], overrides = {}) { + const policy = [...records(POLICIES).values()][0] + return new Request(destination.url as string, { + method: 'POST', + headers: { 'cf-webhook-auth': destination.secret as string }, + // The shape of a real Workers issue alert from ANS (2026-10-12). It has no + // account_id; the policy ID binds it to the account. + body: JSON.stringify({ + name: destination.name, + text: 'Checkout issue', + ts: 1_800_000_000, + policy_id: policy?.id ?? 'deleted', + policy_name: destination.name, + alert_type: 'workers_observability_real_time_issue', + alert_event: 'ALERT_STATE_EVENT_START', + alert_correlation_id: 'vega-persisted-run-1', + data: { + issue: { + id: 'issue-1', + fingerprint: 'fp-1', + title: 'Unhandled exception', + type: 'exception', + status: 'active', + count: '9', + first_observed: '2026-10-12T16:00:00Z', + last_observed: '2026-10-12T16:05:00Z', + services: [ + { + name: 'checkout', + type: 'worker', + first_observed: '2026-10-12T16:00:00Z', + last_observed: '2026-10-12T16:05:00Z' + } + ] + }, + state_correlation_id: 'state-1', + state_event: 'ALERT_STATE_EVENT_START' + }, + ...overrides + }) + }) +} + +beforeEach(async () => { + resources.clear() + for (const path of [WEBHOOKS, POLICIES, AUTOMATIONS]) resources.set(path, new Map()) + requests.length = 0 + deliveries.length = 0 + callbackStatus = 204 + rejectVerification = false + failNext = '' + policyReadStatus = 200 + verificationCount = 0 + bootstrapCount = 0 + validKeys = [signingSecret] + await seedSpec({}) + mockIdentityProbe({ accounts: [{ id: accountId, name: 'Events test' }] }) + server.use( + http.get(`${API_BASE}/accounts/${accountId}/alerting/v3/available_alerts`, () => + HttpResponse.json( + cfSuccess({ + Workers: [ + { + type: 'workers_observability_real_time_issue', + display_name: 'Worker issue', + description: 'An issue automation fires', + filter_options: [] + } + ] + }) + ) + ), + http.post(callbackUrl, async ({ request }) => { + const body = await request.text() + await checkSignature(request, body) + expect(request.headers.get('User-Agent')).toBe(USER_AGENT) + const value = JSON.parse(body) + if (value.type === 'verification') { + verificationCount++ + return HttpResponse.json({ challenge: rejectVerification ? 'wrong' : value.challenge }) + } + expect(request.headers.get('webhook-id')).toBe(value.eventId) + deliveries.push({ body, headers: request.headers }) + return new HttpResponse(null, { status: callbackStatus }) + }), + http.all(`${API_BASE}/accounts/${accountId}/*`, async ({ request }) => { + expect(request.headers.get('Authorization')).toBe(`Bearer ${token}`) + expect(request.headers.get('User-Agent')).toBe(USER_AGENT) + const path = new URL(request.url).pathname.slice(`/client/v4/accounts/${accountId}`.length) + const collection = [WEBHOOKS, POLICIES, AUTOMATIONS].find( + (prefix) => path === prefix || path.startsWith(prefix + '/') + ) + if (!collection) throw new Error(`Unexpected API path ${path}`) + const records = resources.get(collection)! + const resourceId = path === collection ? undefined : path.slice(collection.length + 1) + requests.push({ method: request.method, path }) + // The real automations list is unpaginated and rejects page/per_page with 400. + const search = new URL(request.url).searchParams + if ( + collection === AUTOMATIONS && + request.method === 'GET' && + (search.has('page') || search.has('per_page')) + ) + return HttpResponse.json({ success: false, errors: [] }, { status: 400 }) + if (failNext === `${request.method} ${collection}`) { + failNext = '' + return HttpResponse.json({ success: false, errors: [] }, { status: 503 }) + } + if ( + resourceId && + collection === POLICIES && + request.method === 'GET' && + policyReadStatus !== 200 + ) + return new HttpResponse(null, { status: policyReadStatus }) + if (request.method === 'GET') { + if (resourceId && !records.has(resourceId)) return new HttpResponse(null, { status: 404 }) + const value = resourceId ? records.get(resourceId) : [...records.values()] + return HttpResponse.json( + cfSuccess( + collection === AUTOMATIONS + ? { [resourceId ? 'automation' : 'automations']: value } + : value + ) + ) + } + if (request.method === 'DELETE') { + records.delete(resourceId!) + return HttpResponse.json(cfSuccess({ id: resourceId })) + } + const input = (await request.json()) as Record + if (collection === WEBHOOKS && request.method === 'POST') { + const probe = await exports.default.fetch( + new Request(input.url as string, { + method: 'POST', + headers: { 'cf-webhook-auth': input.secret as string }, + body: JSON.stringify({ + text: 'Hello World! This is a test message sent from https://cloudflare.com.' + }) + }) + ) + expect(probe.status).toBe(204) + bootstrapCount++ + } + const id = resourceId ?? crypto.randomUUID().replaceAll('-', '') + const value = { ...input, id } as Resource + // Like real ANS: a destination keeps the secret it was created with. + if (collection === WEBHOOKS && request.method === 'PUT' && records.has(id)) + value.secret = records.get(id)!.secret + records.set(id, value) + return HttpResponse.json( + cfSuccess(collection === AUTOMATIONS ? { automation: value } : { id }) + ) + }) + ) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await clearKv(env.OAUTH_KV) + await clearSpec() +}) + +describe('MCP Events through the real Worker', () => { + it('uses provider-issued credentials and stops when their grant is revoked', async () => { + const redirectUri = 'https://app.example.com/cb' + const registration = await exports.default.fetch( + new Request('https://mcp.cloudflare.com/register', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ redirect_uris: [redirectUri], token_endpoint_auth_method: 'none' }) + }) + ) + expect(registration.status).toBe(201) + const client = (await registration.json()) as { client_id: string } + const verifier = 'x'.repeat(43) + const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(verifier)) + const challenge = btoa(String.fromCharCode(...new Uint8Array(digest))) + .replaceAll('+', '-') + .replaceAll('/', '_') + .replaceAll('=', '') + const helpers = getOAuthApi( + { + apiHandlers: { '/mcp': { fetch: () => new Response(null, { status: 404 }) } }, + defaultHandler: { fetch: () => new Response(null, { status: 404 }) }, + authorizeEndpoint: '/authorize', + tokenEndpoint: '/token', + resourceMetadata: { resource: env.MCP_RESOURCE } + }, + env + ) + const authorize = new URL('https://mcp.cloudflare.com/authorize') + authorize.search = new URLSearchParams({ + client_id: client.client_id, + redirect_uri: redirectUri, + response_type: 'code', + scope: 'user:read account:read', + resource: env.MCP_RESOURCE, + code_challenge: challenge, + code_challenge_method: 'S256' + }).toString() + const request = await helpers.parseAuthRequest(new Request(authorize)) + const approved = await helpers.completeAuthorization({ + request, + userId: 'events-user', + scope: request.scope, + metadata: {}, + props: { + type: 'account_token', + accessToken: token, + account: { id: accountId, name: 'Events test' } + } + }) + const exchange = await exports.default.fetch( + new Request('https://mcp.cloudflare.com/token', { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + grant_type: 'authorization_code', + code: new URL(approved.redirectTo).searchParams.get('code')!, + client_id: client.client_id, + redirect_uri: redirectUri, + code_verifier: verifier, + resource: env.MCP_RESOURCE + }).toString() + }) + ) + expect(exchange.status).toBe(200) + const issued = (await exchange.json()) as { access_token: string } + expect((await rpc('events/subscribe', params, issued.access_token)).error).toBeUndefined() + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(1) + const summary = await helpers.unwrapToken(issued.access_token) + await helpers.revokeGrant(summary!.grantId, summary!.userId) + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(1) + }) + it('advertises a catalogue of event types without creating resources', async () => { + const capabilities = (await rpc('server/discover')).result?.capabilities + expect(capabilities?.events).toEqual({ listChanged: false }) + expect(capabilities?.extensions).toHaveProperty(['io.modelcontextprotocol/skills']) + const list = await rpc('events/list') + expect(list.result?.events).toHaveLength(1) + expect(requests.every((request) => request.method === 'GET')).toBe(true) + }) + + it('subscribes, provisions real API shapes, delivers a signed event and unsubscribes', async () => { + const subscribed = await rpc('events/subscribe', params) + expect(subscribed.error).toBeUndefined() + expect(subscribed.result?.id).toMatch(/^sub_[a-f0-9]{64}$/) + expect(bootstrapCount).toBe(1) + expect(verificationCount).toBe(1) + const policy = [...records(POLICIES).values()][0] + expect(policy.enabled).toBe(true) + expect(JSON.stringify(policy)).not.toContain(signingSecret) + expect(JSON.stringify(policy)).not.toContain(token) + expect([...records(AUTOMATIONS).values()][0]).toMatchObject({ + service: 'checkout', + afterOccurrences: 1, + policyId: policy.id, + enabled: true + }) + const request = ansRequest() + expect((await exports.default.fetch(request.clone())).status).toBe(204) + expect(deliveries).toHaveLength(1) + expect(JSON.parse(deliveries[0].body)).toMatchObject({ + name: params.name, + data: { data: { issue: { id: 'issue-1' } } }, + cursor: null + }) + // The draft's unsubscribe sends delivery as `{ url }` alone. + const delivery = { url: params.delivery.url } + expect((await rpc('events/unsubscribe', { ...params, delivery })).error).toBeUndefined() + expect((await rpc('events/unsubscribe', { ...params, delivery })).error).toBeUndefined() + expect([...resources.values()].every((collection) => collection.size === 0)).toBe(true) + expect((await exports.default.fetch(request.clone())).status).toBe(204) + expect(deliveries).toHaveLength(1) + }) + + it('refreshes without duplicates and verifies a replacement signing key', async () => { + const first = await rpc('events/subscribe', params) + expect((await rpc('events/subscribe', params)).result?.id).toBe(first.result?.id) + expect(verificationCount).toBe(1) + const replacement = `whsec_${btoa('b'.repeat(32))}` + validKeys = [replacement] + expect( + ( + await rpc('events/subscribe', { + ...params, + delivery: { ...params.delivery, secret: replacement } + }) + ).result?.id + ).toBe(first.result?.id) + expect(verificationCount).toBe(2) + expect([...resources.values()].every((collection) => collection.size === 1)).toBe(true) + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries[0].headers.get('webhook-signature')?.split(' ')).toHaveLength(2) + }) + + it('returns transient failures to ANS and preserves event identity on redelivery', async () => { + await rpc('events/subscribe', params) + callbackStatus = 429 + expect((await exports.default.fetch(ansRequest())).status).toBe(503) + expect(deliveries).toHaveLength(1) + callbackStatus = 204 + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(2) + expect(deliveries[1].body).toBe(deliveries[0].body) + }) + + it('rejects failed verification before creating ANS resources', async () => { + rejectVerification = true + const failed = await rpc('events/subscribe', params) + expect(failed.error?.code).toBe(-32015) + expect(failed.error?.data).toEqual({ reason: 'challenge_failed' }) + expect([...resources.values()].every((collection) => collection.size === 0)).toBe(true) + }) + + it('recovers partial provisioning on the next subscribe', async () => { + failNext = `POST ${AUTOMATIONS}` + expect((await rpc('events/subscribe', params)).error).toBeDefined() + expect([...records(POLICIES).values()][0].enabled).toBe(false) + expect((await rpc('events/subscribe', params)).error).toBeUndefined() + expect([...resources.values()].every((collection) => collection.size === 1)).toBe(true) + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + }) + + it('stops after expiry without a background scheduler', async () => { + const result = await rpc('events/subscribe', params) + vi.spyOn(Date, 'now').mockReturnValue(Date.parse(result.result!.refreshBefore!) + 1) + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(0) + }) + + it('keeps delivering after a refresh extends the subscription past its first expiry', async () => { + const initial = await rpc('events/subscribe', { ...params, ttlMs: 10 * 60_000 }) + const firstExpiry = Date.parse(initial.result!.refreshBefore!) + // Refresh shortly before expiry, with a longer lifetime. + 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. + vi.spyOn(Date, 'now').mockReturnValue(firstExpiry + 60_000) + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(1) + }) + + it('stops forwarding on unsubscribe even when ANS cleanup fails', async () => { + await rpc('events/subscribe', params) + const destination = [...records(WEBHOOKS).values()][0] + failNext = `GET ${POLICIES}` + const { secret: _secret, ...delivery } = params.delivery + expect((await rpc('events/unsubscribe', { ...params, delivery })).error).toBeDefined() + expect((await exports.default.fetch(ansRequest(destination))).status).toBe(204) + expect(deliveries).toHaveLength(0) + }) + + it('keeps the delivery ticket encrypted in KV and out of ANS', async () => { + await rpc('events/subscribe', params) + const { keys } = await env.OAUTH_KV.list({ prefix: 'mcp-events:ticket:' }) + expect(keys).toHaveLength(1) + const stored = await env.OAUTH_KV.get(keys[0].name) + expect(stored).not.toContain(token) + const destination = [...records(WEBHOOKS).values()][0] + expect(String(destination.secret)).toMatch(/^mcp-events-ingress-v1:/) + expect(JSON.stringify([...resources.values()].map((r) => [...r.values()]))).not.toContain(token) + }) + + it('drops a delivery the receiver rejects with 410 but keeps the subscription', async () => { + await rpc('events/subscribe', params) + callbackStatus = 410 + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect([...records(POLICIES).values()][0].enabled).toBe(true) + callbackStatus = 204 + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(2) + }) + + it('honors live permission revocation and leaves transient API failures to ANS', async () => { + await rpc('events/subscribe', params) + policyReadStatus = 503 + expect((await exports.default.fetch(ansRequest())).status).toBe(503) + policyReadStatus = 403 + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(0) + }) + + it('leaves rate-limited delivery authorization checks to ANS without local retries', async () => { + await rpc('events/subscribe', params) + requests.length = 0 + policyReadStatus = 429 + expect((await exports.default.fetch(ansRequest())).status).toBe(503) + expect( + requests.filter( + (request) => request.method === 'GET' && request.path.startsWith(POLICIES + '/') + ) + ).toHaveLength(1) + expect(deliveries).toHaveLength(0) + }) + + it('rejects another account and forged ingress credentials', async () => { + expect( + ( + await rpc('events/subscribe', { + ...params, + arguments: { ...params.arguments, account_id: 'b'.repeat(32) } + }) + ).error + ).toBeDefined() + await rpc('events/subscribe', params) + const request = ansRequest() + request.headers.set('cf-webhook-auth', 'forged') + expect((await exports.default.fetch(request)).status).toBe(401) + expect( + (await exports.default.fetch(ansRequest(undefined, { account_id: 'b'.repeat(32) }))).status + ).toBe(403) + expect(deliveries).toHaveLength(0) + }) + + it('serializes overlapping local subscriptions and keeps one resource set', async () => { + const results = await Promise.all([ + rpc('events/subscribe', params), + rpc('events/subscribe', params) + ]) + expect(results[0].result?.id).toBeTruthy() + expect(results[1].result?.id).toBe(results[0].result?.id) + expect([...resources.values()].every((collection) => collection.size === 1)).toBe(true) + }) + + it('accepts the draft subscribe fields and clamps the requested lifetime', async () => { + const result = await rpc('events/subscribe', { + ...params, + cursor: 'opaque-cursor', + maxAgeMs: 300_000, + ttlMs: 1 + }) + expect(result.error).toBeUndefined() + expect(result.result).toMatchObject({ cursor: null, truncated: false }) + }) + + it('answers unknown events with NotFound before provisioning', async () => { + const result = await rpc('events/subscribe', { + ...params, + name: 'cloudflare.alert.http_alert_origin_error' + }) + expect(result.error?.code).toBe(-32011) + expect(result.error?.data).toEqual({ kind: 'event' }) + expect([...resources.values()].every((collection) => collection.size === 0)).toBe(true) + }) + + it('describes the event with a schema ChatGPT accepts', async () => { + const list = await rpc('events/list') + const [event] = (list.result?.events ?? []) as Array<{ + inputSchema: Record + payloadSchema: Record + }> + for (const schema of [event.inputSchema, event.payloadSchema]) { + expect(schema.type).toBe('object') + for (const key of ['$schema', 'oneOf', 'anyOf', 'allOf', 'not']) expect(schema).not.toHaveProperty(key) + } + }) + + it('notifies on new issues when no trigger is given', async () => { + const { afterOccurrences: _trigger, ...args } = params.arguments + expect((await rpc('events/subscribe', { ...params, arguments: args })).error).toBeUndefined() + expect([...records(AUTOMATIONS).values()][0]).toMatchObject({ afterOccurrences: 1 }) + }) + + it('rejects invalid trigger combinations before provisioning', async () => { + for (const args of [ + { ...params.arguments, afterInactivitySeconds: 3600 }, + { account_id: accountId, filters: { services: ['checkout'] } } + ]) { + expect((await rpc('events/subscribe', { ...params, arguments: args })).error?.code).toBe( + -32602 + ) + } + expect([...resources.values()].every((collection) => collection.size === 0)).toBe(true) + }) + + it('honors policy edits made through the existing API', async () => { + await rpc('events/subscribe', params) + const policy = [...records(POLICIES).values()][0] + policy.enabled = false + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + policy.enabled = true + policy.mechanisms = { webhooks: [] } + expect((await exports.default.fetch(ansRequest())).status).toBe(204) + expect(deliveries).toHaveLength(0) + }) + + it('fails closed when stored policy metadata is changed', async () => { + await rpc('events/subscribe', params) + const policy = [...records(POLICIES).values()][0] + policy.description = 'changed' + expect((await exports.default.fetch(ansRequest())).status).toBe(503) + expect(deliveries).toHaveLength(0) + }) + + it('never follows a callback redirect', async () => { + await rpc('events/subscribe', params) + server.use( + http.post( + callbackUrl, + () => + new HttpResponse(null, { + status: 302, + headers: { Location: 'https://127.0.0.1/private' } + }) + ) + ) + expect((await exports.default.fetch(ansRequest())).status).toBe(503) + expect(deliveries).toHaveLength(0) + }) +})