From 8d3f089803641abcf0ae7a1d3fd0cf2fe5cbe4a9 Mon Sep 17 00:00:00 2001 From: Nikos Benakis Date: Fri, 13 Nov 2026 23:10:31 +0200 Subject: [PATCH 1/2] feat: hookgres/mcp-events, webhook delivery for MCP Events (SEP-3415) Handles events/list, events/subscribe and events/unsubscribe for webhook mode, verifies callback endpoints with a signed challenge, and emit() queues an event per matching subscription inside the caller's transaction. Co-Authored-By: Claude Opus 5.5 --- .changeset/mcp-events.md | 5 + docs/.vitepress/config.ts | 1 + docs/guides/mcp-events.md | 104 +++++++ package.json | 4 + src/mcp-events.ts | 589 ++++++++++++++++++++++++++++++++++++++ test/mcp-events.test.ts | 375 ++++++++++++++++++++++++ 6 files changed, 1078 insertions(+) create mode 100644 .changeset/mcp-events.md create mode 100644 docs/guides/mcp-events.md create mode 100644 src/mcp-events.ts create mode 100644 test/mcp-events.test.ts diff --git a/.changeset/mcp-events.md b/.changeset/mcp-events.md new file mode 100644 index 0000000..74144ec --- /dev/null +++ b/.changeset/mcp-events.md @@ -0,0 +1,5 @@ +--- +"hookgres": minor +--- + +Add `hookgres/mcp-events`: webhook delivery for MCP Events (SEP-3415, draft). It handles `events/list`, `events/subscribe` and `events/unsubscribe`, verifies callback endpoints with a signed challenge, and `emit()` queues an event for every matching subscription inside your transaction, with bounded retries, `X-MCP-Subscription-Id` and Standard Webhooks signatures. diff --git a/docs/.vitepress/config.ts b/docs/.vitepress/config.ts index 1f0567c..96cbb6a 100644 --- a/docs/.vitepress/config.ts +++ b/docs/.vitepress/config.ts @@ -170,6 +170,7 @@ export default defineConfig({ }, { text: "Privacy and GDPR", link: "/guides/privacy" }, { text: "Migrating from Svix", link: "/guides/migrating-from-svix" }, + { text: "MCP Events webhooks", link: "/guides/mcp-events" }, ], }, { diff --git a/docs/guides/mcp-events.md b/docs/guides/mcp-events.md new file mode 100644 index 0000000..b016af4 --- /dev/null +++ b/docs/guides/mcp-events.md @@ -0,0 +1,104 @@ +# MCP Events webhooks + +MCP Events ([SEP-3415](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/3415), +extension `io.modelcontextprotocol/events`) lets an MCP server push events to +agents. In webhook mode the client registers a callback URL and a `whsec_` +secret, and the server POSTs each event to it, signed with Standard Webhooks. + +`hookgres/mcp-events` is that server side on your Postgres. It answers +`events/list`, `events/subscribe` and `events/unsubscribe`, and `emit()` queues +an event for every matching subscription, inside your transaction if you pass +one. + +::: warning Draft specification +SEP-3415 is still under review, and its error codes are provisional. This +module follows the draft as of November 2026 and may change with it. +::: + +## Set it up + +```ts +import pg from "pg"; +import { McpEvents, McpEventsError } from "hookgres/mcp-events"; + +const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); + +const events = new McpEvents({ + pool, + events: [ + { + name: "incident.created", + description: "A new incident", + inputSchema: { + type: "object", + properties: { severity: { type: "string" } }, + }, + // Which subscriptions get an event, from their arguments. + matches: (args, data) => + !args.severity || args.severity === data.severity, + }, + ], +}); + +await events.migrate(); // its own schema, "hookgres_mcp_events" +events.start(); // delivery worker in this process +``` + +Add `events.capabilities` to the capabilities your server returns, and route +`events/*` requests to `handle()` with the authenticated principal: + +```ts +try { + return await events.handle(request.method, request.params, { + principal: auth.subject, // OAuth sub, API key id, ... + }); +} catch (error) { + if (error instanceof McpEventsError) return { error: error.toJSON() }; + throw error; +} +``` + +## Emit events + +```ts +const client = await pool.connect(); +try { + await client.query("BEGIN"); + const incident = await createIncident(client, input); + await events.emit( + "incident.created", + { incidentId: incident.id, severity: incident.severity }, + { client, eventId: incident.id }, + ); + await client.query("COMMIT"); +} finally { + client.release(); +} +``` + +Roll back and nothing is sent. Commit, and the event reaches every live +subscription even if the process restarts before delivery. + +## What it does for you + +| SEP-3415 requirement | How | +| -------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- | +| Standard Webhooks signature with the client's secret | Each subscription is a Hookgres endpoint with that secret | +| `X-MCP-Subscription-Id` on every POST | Set per attempt; the id is derived from `(principal, url, name, arguments)` | +| `webhook-id` unique per message, kept across retries, not derived from `eventId` | Hookgres message ids, one message per subscription | +| Bounded retries (3-5 attempts within 10-15 minutes) | 5 attempts over about 8 minutes by default | +| `410` and `413` fail that delivery only | `nonRetryableStatuses: [410, 413]` | +| `https` only, no private addresses, checked at delivery time, no redirects | `blockPrivateAddresses` and Hookgres's pinned connections | +| Endpoint verification before the first delivery | A signed `verification` challenge, cached per `(principal, url)`, or your own allowlist with `verify` | +| TTL grant, refresh and secret rotation | `ttl` bounds; a new secret on refresh dual-signs for 5 minutes | +| `deliveryStatus` on refresh | `active`, `lastDeliveryAt`, `lastError` category, `failedSince` | + +## Not covered yet + +- Poll and push delivery: they answer `Unsupported`. +- Cursor replay: events are emit-only (`cursor: null`). Hookgres keeps queued + events across restarts, which is more than the SEP asks of emit-only types. +- `gap` and `terminated` control envelopes. +- No-expiry subscriptions are only granted with `allowNoExpiry: true`. + +Call `events.deleteExpired()` on a schedule to remove expired subscriptions. diff --git a/package.json b/package.json index a20a3bc..0b47c90 100644 --- a/package.json +++ b/package.json @@ -33,6 +33,10 @@ "types": "./dist/kysely.d.ts", "default": "./dist/kysely.js" }, + "./mcp-events": { + "types": "./dist/mcp-events.d.ts", + "default": "./dist/mcp-events.js" + }, "./prisma": { "types": "./dist/prisma.d.ts", "default": "./dist/prisma.js" diff --git a/src/mcp-events.ts b/src/mcp-events.ts new file mode 100644 index 0000000..279a014 --- /dev/null +++ b/src/mcp-events.ts @@ -0,0 +1,589 @@ +/** + * Webhook delivery for MCP Events (SEP-3415, extension + * `io.modelcontextprotocol/events`), on Hookgres. + * + * Handles `events/list`, `events/subscribe` and `events/unsubscribe` for the + * webhook delivery mode, and `emit()` queues an event for every matching + * subscription, optionally inside your own transaction. Each subscription is + * a Hookgres endpoint, so deliveries are signed with Standard Webhooks, + * retried within a bounded window, logged, and survive restarts. + * + * Wire `handle()` into your MCP server's request handler; this module + * doesn't depend on an MCP SDK. + * + * @see https://hookgres.dev/docs/guides/mcp-events + */ +import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; +import type pg from "pg"; +import { HookgresValidationError } from "./errors.js"; +import { Hookgres, type HookgresOptions } from "./hookgres.js"; +import { newId } from "./ids.js"; +import { assertSecret, sign } from "./signing.js"; +import { AddressGuard, guardedTransport, type Transport } from "./ssrf.js"; +import type { Queryable } from "./types.js"; + +export const EVENTS_EXTENSION = "io.modelcontextprotocol/events"; +export const SUBSCRIPTION_ID_HEADER = "x-mcp-subscription-id"; + +/** SEP-3415 error codes (provisional until the SEP is accepted). */ +export const McpEventsErrorCode = { + InvalidParams: -32602, + NotFound: -32023, + Forbidden: -32024, + ResourceExhausted: -32025, + Unsupported: -32026, + CallbackEndpointError: -32027, +} as const; + +/** A JSON-RPC error to return to the client as is. */ +export class McpEventsError extends Error { + constructor( + readonly code: number, + message: string, + readonly data?: Record, + ) { + super(message); + this.name = "McpEventsError"; + } + + toJSON() { + return { + code: this.code, + message: this.message, + ...(this.data ? { data: this.data } : {}), + }; + } +} + +export interface McpEventType { + name: string; + description?: string; + inputSchema?: Record; + payloadSchema?: Record; + /** Does an event's `data` match a subscription's `arguments`? Defaults to always. */ + matches?: (args: Record, data: unknown) => boolean; +} + +export interface McpEventsOptions { + pool: pg.Pool; + events: McpEventType[]; + /** Postgres schema for the subscriptions and deliveries. Defaults to `hookgres_mcp_events`. */ + schema?: string; + /** TTLs in ms: the default grant, and the range suggestions are clamped to. */ + ttl?: { defaultMs?: number; minMs?: number; maxMs?: number }; + /** Grant `refreshBefore: null` when a client asks for no expiry. Defaults to `false`. */ + allowNoExpiry?: boolean; + /** + * How a callback URL proves it wants deliveries before the first one: + * `"handshake"` (default) POSTs a signed `verification` challenge, or a + * function that allowlists URLs. + */ + verify?: "handshake" | ((url: string) => boolean); + /** + * For local testing only: accept `http://` callback URLs and these private + * addresses (IPs or CIDRs). Production servers must leave both off. + */ + insecure?: { allowHttp?: boolean; allowAddresses?: string[] }; + /** Passed to the underlying Hookgres instance. */ + hookgres?: Partial< + Omit< + HookgresOptions, + | "pool" + | "schema" + | "envelope" + | "deliveryHeaders" + | "blockPrivateAddresses" + > + >; +} + +/** SEP-3415 suggests 3-5 attempts within 10-15 minutes. */ +const RETRY_SCHEDULE = [5_000, 30_000, 120_000, 300_000]; +const HOUR = 3_600_000; + +interface Subscription { + endpointId: string; + id: string; + disabled: boolean; + failingSince: Date | null; + secret: string; + refreshBefore: string | null; + verified: boolean; +} + +export class McpEvents { + readonly hookgres: Hookgres; + private readonly pool: pg.Pool; + private readonly types: Map; + private readonly table: string; + private readonly ttl: { defaultMs: number; minMs: number; maxMs: number }; + private readonly allowNoExpiry: boolean; + private readonly verifyMode: McpEventsOptions["verify"]; + private readonly allowHttp: boolean; + private readonly guard: AddressGuard; + private readonly transport: { send: Transport; close(): void }; + + constructor(options: McpEventsOptions) { + const schema = options.schema ?? "hookgres_mcp_events"; + const allow = options.insecure?.allowAddresses ?? []; + this.guard = new AddressGuard({ allow }); + this.transport = guardedTransport(this.guard); + this.hookgres = new Hookgres({ + retrySchedule: RETRY_SCHEDULE, + requestTimeoutMs: 5_000, + // Suspension would fail queued events; TTL expiry cleans up instead. + disableEndpointAfterMs: null, + nonRetryableStatuses: [410, 413], + ...options.hookgres, + pool: options.pool, + schema, + envelope: false, + blockPrivateAddresses: { allow }, + deliveryHeaders: ({ tenantId }) => ({ + [SUBSCRIPTION_ID_HEADER]: tenantId, + }), + }); + this.pool = options.pool; + this.table = `"${schema}".endpoints`; + this.types = new Map(options.events.map((type) => [type.name, type])); + this.ttl = { + defaultMs: options.ttl?.defaultMs ?? HOUR, + minMs: options.ttl?.minMs ?? 60_000, + maxMs: options.ttl?.maxMs ?? 24 * HOUR, + }; + this.allowNoExpiry = options.allowNoExpiry ?? false; + this.verifyMode = options.verify ?? "handshake"; + this.allowHttp = options.insecure?.allowHttp ?? false; + } + + /** Merge into the capabilities your server returns. */ + get capabilities() { + return { extensions: { [EVENTS_EXTENSION]: {} } }; + } + + migrate(): Promise { + return this.hookgres.migrate(); + } + + start(): void { + this.hookgres.start(); + } + + async stop(): Promise { + await this.hookgres.stop(); + this.transport.close(); + } + + /** + * Handles an `events/*` request. `principal` is the authenticated subject + * (OAuth `sub`, API key id); webhook methods throw `Forbidden` without one. + * Throws `McpEventsError`; return `error.toJSON()` as the JSON-RPC error. + */ + async handle( + method: string, + params: Record | undefined, + context: { principal?: string | null }, + ): Promise> { + switch (method) { + case "events/list": + return this.list(); + case "events/subscribe": + return this.subscribe(params ?? {}, context.principal); + case "events/unsubscribe": + return this.unsubscribe(params ?? {}, context.principal); + case "events/poll": + case "events/stream": + throw new McpEventsError( + McpEventsErrorCode.Unsupported, + "Unsupported", + { + feature: "deliveryMode", + value: method === "events/poll" ? "poll" : "push", + }, + ); + default: + throw new McpEventsError(-32601, `Method not found: ${method}`); + } + } + + list(): Record { + return { + resultType: "complete", + events: [...this.types.values()].map((type) => ({ + name: type.name, + ...(type.description ? { description: type.description } : {}), + delivery: ["webhook"], + inputSchema: type.inputSchema ?? { type: "object" }, + ...(type.payloadSchema ? { payloadSchema: type.payloadSchema } : {}), + })), + }; + } + + /** + * Queues `name` for every live subscription whose arguments match. Pass + * your transaction's `client` and the deliveries are queued only if it + * commits. Returns the number of subscriptions it was queued for. + */ + async emit( + name: string, + data: Record, + options: { eventId?: string; timestamp?: Date; client?: Queryable } = {}, + ): Promise { + const type = this.types.get(name); + if (!type) { + throw new HookgresValidationError(`Unknown MCP event type "${name}"`); + } + const db = options.client ?? this.pool; + const { rows } = await db.query<{ + tenant_id: string; + metadata: { arguments?: Record }; + }>( + `SELECT tenant_id, metadata FROM ${this.table} + WHERE $1 = ANY(event_types) AND metadata->>'verified' = 'true' + AND (metadata->>'refreshBefore' IS NULL + OR (metadata->>'refreshBefore')::timestamptz > now())`, + [name], + ); + const eventId = options.eventId ?? newId("evt"); + const timestamp = (options.timestamp ?? new Date()).toISOString(); + let queued = 0; + for (const row of rows) { + if (type.matches && !type.matches(row.metadata.arguments ?? {}, data)) { + continue; + } + // Emit-only: no durable upstream to replay from, so no cursor. + const payload = { eventId, name, timestamp, data, cursor: null }; + const result = await this.hookgres.send( + { + tenantId: row.tenant_id, + eventType: name, + payload, + idempotencyKey: eventId, + }, + { client: options.client }, + ); + if (!result.deduplicated) queued += result.deliveryCount; + } + return queued; + } + + /** Deletes subscriptions past their `refreshBefore`. Returns how many. */ + async deleteExpired(): Promise { + const { rowCount } = await this.pool.query( + `DELETE FROM ${this.table} + WHERE (metadata->>'refreshBefore')::timestamptz <= now()`, + ); + return rowCount ?? 0; + } + + private async subscribe( + params: Record, + principal: string | null | undefined, + ): Promise> { + if (!principal) { + throw new McpEventsError(McpEventsErrorCode.Forbidden, "Forbidden", { + reason: "events/subscribe needs an authenticated principal", + }); + } + const { name, args, url } = this.keyParams(params); + const delivery = params.delivery as Record; + if (delivery.mode !== undefined && delivery.mode !== "webhook") { + throw new McpEventsError(McpEventsErrorCode.Unsupported, "Unsupported", { + feature: "deliveryMode", + value: delivery.mode, + }); + } + const secret = delivery.secret; + try { + assertSecret(String(secret)); + } catch { + throw invalid( + "delivery.secret must be whsec_ followed by base64 of 24-64 bytes", + ); + } + if (typeof secret !== "string") + throw invalid("delivery.secret is required"); + const refreshBefore = this.grant(params.ttlMs); + const id = subscriptionId(principal, url, name, args); + const existing = await this.find(id); + + let verified = + existing?.verified || (await this.isVerified(principal, url)); + if (!verified) { + await this.verifyEndpoint(url, secret, id); + verified = true; + } + const metadata = { + principal, + name, + arguments: args, + refreshBefore, + verified, + }; + + if (!existing) { + await this.hookgres.createEndpoint({ + tenantId: id, + url, + eventTypes: [name], + secret, + metadata, + }); + return { + resultType: "complete", + id, + refreshBefore, + cursor: null, + truncated: false, + }; + } + + if (existing.secret !== secret) { + // Dual-sign with the old secret for a while, as the SEP recommends. + await this.hookgres.rotateEndpointSecret(existing.endpointId, { + secret, + graceMs: 5 * 60_000, + }); + } + await this.hookgres.updateEndpoint(existing.endpointId, { + metadata, + ...(existing.disabled ? { disabled: false } : {}), + }); + return { + resultType: "complete", + id, + refreshBefore, + cursor: null, + truncated: false, + deliveryStatus: await this.deliveryStatus(existing), + }; + } + + private async unsubscribe( + params: Record, + principal: string | null | undefined, + ): Promise> { + if (!principal) { + throw new McpEventsError(McpEventsErrorCode.Forbidden, "Forbidden", { + reason: "events/unsubscribe needs an authenticated principal", + }); + } + const { name, args, url } = this.keyParams(params); + const existing = await this.find( + subscriptionId(principal, url, name, args), + ); + if (!existing) { + throw new McpEventsError(McpEventsErrorCode.NotFound, "NotFound", { + kind: "subscription", + }); + } + await this.hookgres.deleteEndpoint(existing.endpointId); + return { resultType: "complete" }; + } + + private keyParams(params: Record) { + const name = params.name; + if (typeof name !== "string") throw invalid("name is required"); + if (!this.types.has(name)) { + throw new McpEventsError(McpEventsErrorCode.NotFound, "NotFound", { + kind: "event", + }); + } + const args = (params.arguments ?? {}) as Record; + if (typeof args !== "object" || args === null || Array.isArray(args)) { + throw invalid("arguments must be an object"); + } + const delivery = params.delivery; + if (typeof delivery !== "object" || delivery === null) { + throw invalid("delivery is required"); + } + const url = (delivery as Record).url; + let parsed: URL; + try { + parsed = new URL(String(url)); + } catch { + throw invalid("delivery.url must be an absolute https URL"); + } + const httpOk = this.allowHttp && parsed.protocol === "http:"; + if (parsed.protocol !== "https:" && !httpOk) { + throw invalid("delivery.url must use https"); + } + return { name, args, url: String(url) }; + } + + private grant(ttlMs: unknown): string | null { + if (ttlMs === null && this.allowNoExpiry) return null; + const suggested = + typeof ttlMs === "number" && Number.isFinite(ttlMs) + ? ttlMs + : this.ttl.defaultMs; + const ms = Math.min(Math.max(suggested, this.ttl.minMs), this.ttl.maxMs); + return new Date(Date.now() + ms).toISOString(); + } + + private async find(id: string): Promise { + const { rows } = await this.pool.query( + `SELECT id, tenant_id, disabled, failing_since, secret, metadata + FROM ${this.table} WHERE tenant_id = $1 LIMIT 1`, + [id], + ); + const row = rows[0]; + if (!row) return null; + return { + endpointId: row.id, + id: row.tenant_id, + disabled: row.disabled, + failingSince: row.failing_since, + secret: row.secret, + refreshBefore: row.metadata.refreshBefore ?? null, + verified: row.metadata.verified === true, + }; + } + + /** Verification is cached per (principal, url), across arguments. */ + private async isVerified(principal: string, url: string): Promise { + if (typeof this.verifyMode === "function") return this.verifyMode(url); + const { rows } = await this.pool.query( + `SELECT 1 FROM ${this.table} + WHERE url = $1 AND metadata->>'principal' = $2 + AND metadata->>'verified' = 'true' LIMIT 1`, + [url, principal], + ); + return rows.length > 0; + } + + private async verifyEndpoint(url: string, secret: string, id: string) { + if (typeof this.verifyMode === "function") { + throw new McpEventsError( + McpEventsErrorCode.CallbackEndpointError, + "Callback URL is not allowed", + { reason: "challenge_failed" }, + ); + } + const challenge = randomBytes(24).toString("base64url"); + const body = JSON.stringify({ type: "verification", challenge }); + const messageId = `msg_verification_${randomBytes(16).toString("hex")}`; + const timestamp = Math.floor(Date.now() / 1000); + let reason: string; + try { + const response = await this.transport.send(url, { + method: "POST", + headers: { + "content-type": "application/json", + "webhook-id": messageId, + "webhook-timestamp": String(timestamp), + "webhook-signature": sign(secret, messageId, timestamp, body), + [SUBSCRIPTION_ID_HEADER]: id, + }, + body, + redirect: "manual", + signal: AbortSignal.timeout(5_000), + }); + const text = response.body + ? await new Response(response.body).text() + : ""; + if (response.status >= 200 && response.status < 300) { + const echoed = parseChallenge(text); + if (echoed !== null && safeEqual(echoed, challenge)) return; + reason = "challenge_failed"; + } else { + reason = response.status >= 500 ? "http_5xx" : "http_4xx"; + } + } catch (error) { + reason = errorCategory(error); + } + throw new McpEventsError( + McpEventsErrorCode.CallbackEndpointError, + "Callback endpoint failed verification", + { reason }, + ); + } + + private async deliveryStatus(sub: Subscription) { + const { rows } = await this.pool.query<{ + last_success: Date | null; + last_error: string | null; + }>( + `SELECT + (SELECT max(updated_at) FROM "${this.schemaName()}".deliveries + WHERE endpoint_id = $1 AND status = 'succeeded') AS last_success, + (SELECT last_error FROM "${this.schemaName()}".deliveries + WHERE endpoint_id = $1 AND last_error IS NOT NULL + ORDER BY updated_at DESC LIMIT 1) AS last_error`, + [sub.endpointId], + ); + const row = rows[0]; + return { + active: true, + lastDeliveryAt: row?.last_success?.toISOString() ?? null, + lastError: + sub.failingSince && row?.last_error + ? lastErrorCategory(row.last_error) + : null, + ...(sub.failingSince + ? { failedSince: sub.failingSince.toISOString() } + : {}), + }; + } + + private schemaName(): string { + return this.table.slice(1, this.table.indexOf('"', 1)); + } +} + +/** Deterministic over (principal, url, name, arguments), per the SEP. */ +export function subscriptionId( + principal: string, + url: string, + name: string, + args: Record, +): string { + const key = canonicalJson([principal, url, name, args]); + return `sub_${createHash("sha256").update(key).digest("hex").slice(0, 24)}`; +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + if (value && typeof value === "object") { + const entries = Object.entries(value as Record) + .filter(([, v]) => v !== undefined) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)); + return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`).join(",")}}`; + } + return JSON.stringify(value); +} + +function invalid(message: string): McpEventsError { + return new McpEventsError(McpEventsErrorCode.InvalidParams, message); +} + +function parseChallenge(text: string): string | null { + try { + const parsed = JSON.parse(text) as { challenge?: unknown }; + return typeof parsed.challenge === "string" ? parsed.challenge : null; + } catch { + return null; + } +} + +function safeEqual(a: string, b: string): boolean { + const x = Buffer.from(a); + const y = Buffer.from(b); + return x.length === y.length && timingSafeEqual(x, y); +} + +function errorCategory(error: unknown): string { + const text = `${(error as Error)?.name ?? ""} ${(error as Error)?.message ?? ""} ${ + (error as { code?: string })?.code ?? "" + }`; + if (/timeout|abort/i.test(text)) return "timeout"; + if (/cert|tls|ssl|self.signed/i.test(text)) return "tls_error"; + return "connection_refused"; +} + +/** Maps Hookgres's `last_error` to the SEP's `lastError` categories. */ +function lastErrorCategory(lastError: string): string { + const status = /^HTTP (\d{3})/.exec(lastError)?.[1]; + if (status) return status.startsWith("5") ? "http_5xx" : "http_4xx"; + return errorCategory(new Error(lastError)); +} diff --git a/test/mcp-events.test.ts b/test/mcp-events.test.ts new file mode 100644 index 0000000..4ec7f80 --- /dev/null +++ b/test/mcp-events.test.ts @@ -0,0 +1,375 @@ +import { randomBytes } from "node:crypto"; +import pg from "pg"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + inject, + it, +} from "vitest"; +import { verifyWebhook } from "../src/index.js"; +import { + McpEvents, + McpEventsErrorCode, + subscriptionId, +} from "../src/mcp-events.js"; +import { drain, type Receiver, startReceiver } from "./helpers.js"; + +const schema = `mcp_${randomBytes(6).toString("hex")}`; +let pool: pg.Pool; +let events: McpEvents; +let receiver: Receiver; +const secret = `whsec_${randomBytes(32).toString("base64")}`; + +function create( + overrides: Partial[0]> = {}, +) { + return new McpEvents({ + pool, + schema, + events: [ + { + name: "incident.created", + description: "A new incident", + inputSchema: { + type: "object", + properties: { severity: { type: "string" } }, + }, + matches: (args, data) => + args.severity === undefined || + args.severity === (data as { severity?: string }).severity, + }, + { name: "deploy.finished" }, + ], + insecure: { allowHttp: true, allowAddresses: ["127.0.0.1"] }, + hookgres: { retrySchedule: [0, 0], pollIntervalMs: 20 }, + ...overrides, + }); +} + +/** Echoes verification challenges; answers deliveries with `respond`. */ +function echoChallenges() { + const respond = receiver.respond; + receiver.respond = (n, request) => { + const body = JSON.parse(request.body || "{}"); + if (body.type === "verification") { + receiver.body = JSON.stringify({ challenge: body.challenge }); + return 200; + } + receiver.body = undefined; + return respond(n, request); + }; +} + +const deliveries = () => + receiver.requests.filter((r) => !r.body.includes('"type":"verification"')); + +async function subscribe( + params: Record = {}, + principal = "user_1", +) { + return events.handle( + "events/subscribe", + { + name: "incident.created", + arguments: {}, + delivery: { mode: "webhook", url: receiver.url, secret }, + ...params, + }, + { principal }, + ); +} + +async function expectError( + promise: Promise, + code: number, + data?: object, +) { + const error = await promise.then( + () => { + throw new Error("expected an error"); + }, + (e) => e, + ); + expect(error.code).toBe(code); + if (data) expect(error.data).toMatchObject(data); +} + +beforeAll(async () => { + pool = new pg.Pool({ connectionString: inject("databaseUrl"), max: 10 }); + await create().migrate(); +}); +afterAll(async () => { + await pool.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + await pool.end(); +}); +beforeEach(async () => { + receiver = await startReceiver(); + echoChallenges(); + events = create(); +}); +afterEach(async () => { + await events.stop(); + await receiver.close(); + await pool.query( + `TRUNCATE "${schema}".endpoints, "${schema}".messages CASCADE`, + ); +}); + +describe("events/list", () => { + it("lists event types with webhook delivery", async () => { + const result = await events.handle("events/list", {}, {}); + expect(result).toMatchObject({ + resultType: "complete", + events: [ + { + name: "incident.created", + delivery: ["webhook"], + description: "A new incident", + }, + { + name: "deploy.finished", + delivery: ["webhook"], + inputSchema: { type: "object" }, + }, + ], + }); + }); + + it("advertises the extension", () => { + expect(events.capabilities).toEqual({ + extensions: { "io.modelcontextprotocol/events": {} }, + }); + }); +}); + +describe("events/subscribe", () => { + it("verifies the endpoint with a signed challenge, then subscribes", async () => { + const result = await subscribe(); + expect(result).toMatchObject({ + resultType: "complete", + cursor: null, + truncated: false, + }); + expect(result.id).toBe( + subscriptionId("user_1", receiver.url, "incident.created", {}), + ); + expect(Date.parse(String(result.refreshBefore))).toBeGreaterThan( + Date.now(), + ); + + const [challenge] = receiver.requests; + expect(challenge?.headers["x-mcp-subscription-id"]).toBe(result.id); + expect(challenge?.headers["webhook-id"]).toMatch(/^msg_verification_/); + expect(() => + verifyWebhook({ + secret, + headers: challenge?.headers ?? {}, + body: challenge?.body ?? "", + }), + ).not.toThrow(); + }); + + it("fails with challenge_failed when the endpoint doesn't echo", async () => { + receiver.respond = () => 200; + receiver.body = "ok"; + await expectError(subscribe(), McpEventsErrorCode.CallbackEndpointError, { + reason: "challenge_failed", + }); + }); + + it("verifies once per (principal, url), across arguments", async () => { + await subscribe(); + await subscribe({ arguments: { severity: "P1" } }); + const challenges = receiver.requests.filter((r) => + r.body.includes("verification"), + ); + expect(challenges).toHaveLength(1); + }); + + it("rejects unauthenticated, http, bad-secret and unknown-event requests", async () => { + await expectError(subscribe({}, ""), McpEventsErrorCode.Forbidden); + await expectError( + subscribe({ delivery: { url: receiver.url, secret: "whsec_short" } }), + McpEventsErrorCode.InvalidParams, + ); + await expectError( + subscribe({ name: "nope" }), + McpEventsErrorCode.NotFound, + { + kind: "event", + }, + ); + const strict = create({ insecure: undefined }); + await expectError( + strict.handle( + "events/subscribe", + { name: "incident.created", delivery: { url: receiver.url, secret } }, + { principal: "user_1" }, + ), + McpEventsErrorCode.InvalidParams, + ); + }); + + it("derives ids per principal", () => { + const a = subscriptionId("a", "https://x/h", "incident.created", { + s: 1, + t: 2, + }); + expect( + subscriptionId("a", "https://x/h", "incident.created", { t: 2, s: 1 }), + ).toBe(a); + expect( + subscriptionId("b", "https://x/h", "incident.created", { s: 1, t: 2 }), + ).not.toBe(a); + }); + + it("clamps the suggested TTL and grants no expiry only when allowed", async () => { + const short = await subscribe({ ttlMs: 1 }); + expect( + Date.parse(String(short.refreshBefore)) - Date.now(), + ).toBeGreaterThan(50_000); + const none = await subscribe({ ttlMs: null }); + expect(none.refreshBefore).not.toBeNull(); + }); +}); + +describe("emit and delivery", () => { + it("delivers only when the emitting transaction commits", async () => { + const { id } = await subscribe(); + + const client = await pool.connect(); + try { + await client.query("BEGIN"); + await events.emit("incident.created", { n: 1 }, { client }); + await client.query("ROLLBACK"); + await client.query("BEGIN"); + await events.emit( + "incident.created", + { n: 2 }, + { client, eventId: "evt_up_2" }, + ); + await client.query("COMMIT"); + } finally { + client.release(); + } + await drain(events.hookgres); + + const got = deliveries(); + expect(got).toHaveLength(1); + const request = got[0]; + if (!request) throw new Error("no delivery"); + const body = JSON.parse(request.body); + expect(body).toMatchObject({ + eventId: "evt_up_2", + name: "incident.created", + data: { n: 2 }, + cursor: null, + }); + expect(Number.isNaN(Date.parse(body.timestamp))).toBe(false); + expect(request.headers["x-mcp-subscription-id"]).toBe(id); + expect(request.headers["webhook-id"]).not.toContain("evt_up_2"); + expect(() => + verifyWebhook({ secret, headers: request.headers, body: request.body }), + ).not.toThrow(); + }); + + it("routes by arguments and gives each subscription its own webhook-id", async () => { + await subscribe({ arguments: { severity: "P1" } }); + await subscribe({ arguments: {} }); + await events.emit("incident.created", { severity: "P2" }); + await events.emit("incident.created", { severity: "P1" }); + await drain(events.hookgres); + + const got = deliveries(); + expect(got).toHaveLength(3); + expect(new Set(got.map((r) => r.headers["webhook-id"])).size).toBe(3); + }); + + it("treats 410 as final for that delivery only", async () => { + await subscribe(); + let n = 0; + receiver.respond = (_, request) => { + if (request.body.includes("verification")) return 200; + n++; + return n === 1 ? 410 : 200; + }; + await events.emit("incident.created", { n: 1 }); + await drain(events.hookgres); + await events.emit("incident.created", { n: 2 }); + await drain(events.hookgres); + expect(deliveries().map((r) => JSON.parse(r.body).data.n)).toEqual([1, 2]); + }); + + it("dual-signs after the client rotates its secret on refresh", async () => { + await subscribe(); + const next = `whsec_${randomBytes(32).toString("base64")}`; + const refreshed = await subscribe({ + delivery: { mode: "webhook", url: receiver.url, secret: next }, + }); + expect(refreshed.deliveryStatus).toMatchObject({ active: true }); + await events.emit("incident.created", {}); + await drain(events.hookgres); + const request = deliveries()[0]; + if (!request) throw new Error("no delivery"); + for (const s of [secret, next]) { + expect(() => + verifyWebhook({ + secret: s, + headers: request.headers, + body: request.body, + }), + ).not.toThrow(); + } + }); + + it("stops after unsubscribe and after expiry", async () => { + await subscribe(); + await events.handle( + "events/unsubscribe", + { + name: "incident.created", + arguments: {}, + delivery: { url: receiver.url }, + }, + { principal: "user_1" }, + ); + await expectError( + events.handle( + "events/unsubscribe", + { + name: "incident.created", + arguments: {}, + delivery: { url: receiver.url }, + }, + { principal: "user_1" }, + ), + McpEventsErrorCode.NotFound, + { kind: "subscription" }, + ); + const expiring = create({ ttl: { minMs: 1, defaultMs: 1 } }); + await expiring.handle( + "events/subscribe", + { name: "deploy.finished", delivery: { url: receiver.url, secret } }, + { principal: "user_1" }, + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(await events.emit("incident.created", {})).toBe(0); + expect(await expiring.emit("deploy.finished", {})).toBe(0); + expect(await expiring.deleteExpired()).toBe(1); + await expiring.stop(); + }); + + it("answers poll and push with Unsupported", async () => { + await expectError( + events.handle("events/poll", {}, {}), + McpEventsErrorCode.Unsupported, + { + value: "poll", + }, + ); + }); +}); From d30006b85d421ac98e649ab715312f3ce6e4dbe9 Mon Sep 17 00:00:00 2001 From: Nikos Benakis Date: Fri, 13 Nov 2026 23:26:39 +0200 Subject: [PATCH 2/2] fix(mcp-events): serialize subscribe per key, bound verification, rate-limit Fixes from review: an advisory lock stops concurrent subscribes from creating duplicate endpoints and unsubscribe removes all of them; the fetch passthrough is no longer allowed (it would skip delivery-time address checks); the challenge response is read with a size cap; verification POSTs are rate-limited per host; arguments are capped at 4 KB; deleteExpired uses deleteEndpoint; deliveryStatus.active reflects suspension; invalid callback addresses map to InvalidParams. Co-Authored-By: Claude Opus 5.5 --- docs/guides/mcp-events.md | 8 ++ src/mcp-events.ts | 158 ++++++++++++++++++++++++++++++++------ test/mcp-events.test.ts | 44 +++++++++++ 3 files changed, 186 insertions(+), 24 deletions(-) diff --git a/docs/guides/mcp-events.md b/docs/guides/mcp-events.md index b016af4..108202d 100644 --- a/docs/guides/mcp-events.md +++ b/docs/guides/mcp-events.md @@ -99,6 +99,14 @@ subscription even if the process restarts before delivery. - Cursor replay: events are emit-only (`cursor: null`). Hookgres keeps queued events across restarts, which is more than the SEP asks of emit-only types. - `gap` and `terminated` control envelopes. +- `arguments` aren't validated against `inputSchema`; they're only capped at + 4 KB. Validate them before calling `handle()` if your `matches` relies on + their shape. - No-expiry subscriptions are only granted with `allowNoExpiry: true`. +Verification POSTs are limited to 10 per destination host per minute +(`verificationsPerHostPerMinute`); over that, subscribe answers +`ResourceExhausted`. Concurrent subscribe and unsubscribe calls for the same +subscription are serialized with a Postgres advisory lock. + Call `events.deleteExpired()` on a schedule to remove expired subscriptions. diff --git a/src/mcp-events.ts b/src/mcp-events.ts index 279a014..c4ad438 100644 --- a/src/mcp-events.ts +++ b/src/mcp-events.ts @@ -93,13 +93,20 @@ export interface McpEventsOptions { | "envelope" | "deliveryHeaders" | "blockPrivateAddresses" + // A custom fetch would turn off the delivery-time address checks. + | "fetch" > >; + /** Verification POSTs allowed per destination host per minute. Defaults to 10. */ + verificationsPerHostPerMinute?: number; } /** SEP-3415 suggests 3-5 attempts within 10-15 minutes. */ const RETRY_SCHEDULE = [5_000, 30_000, 120_000, 300_000]; const HOUR = 3_600_000; +const MAX_ARGUMENTS_BYTES = 4096; +/** Enough for {"challenge": "..."}; a longer body can't be a valid answer. */ +const MAX_CHALLENGE_RESPONSE_BYTES = 4096; interface Subscription { endpointId: string; @@ -122,6 +129,9 @@ export class McpEvents { private readonly allowHttp: boolean; private readonly guard: AddressGuard; private readonly transport: { send: Transport; close(): void }; + private readonly verificationLimit: number; + /** Verification POST times per host, for the SEP's per-host rate limit. */ + private readonly verificationLog = new Map(); constructor(options: McpEventsOptions) { const schema = options.schema ?? "hookgres_mcp_events"; @@ -154,6 +164,7 @@ export class McpEvents { this.allowNoExpiry = options.allowNoExpiry ?? false; this.verifyMode = options.verify ?? "handshake"; this.allowHttp = options.insecure?.allowHttp ?? false; + this.verificationLimit = options.verificationsPerHostPerMinute ?? 10; } /** Merge into the capabilities your server returns. */ @@ -269,11 +280,39 @@ export class McpEvents { /** Deletes subscriptions past their `refreshBefore`. Returns how many. */ async deleteExpired(): Promise { - const { rowCount } = await this.pool.query( - `DELETE FROM ${this.table} + const { rows } = await this.pool.query<{ id: string }>( + `SELECT id FROM ${this.table} WHERE (metadata->>'refreshBefore')::timestamptz <= now()`, ); - return rowCount ?? 0; + let deleted = 0; + // deleteEndpoint removes deliveries first, so it can't deadlock workers. + for (const row of rows) { + if (await this.hookgres.deleteEndpoint(row.id)) deleted++; + } + return deleted; + } + + /** + * Runs `fn` while holding a session advisory lock on the subscription id, + * so concurrent subscribe and unsubscribe calls for one key can't race. + */ + private async withKeyLock(id: string, fn: () => Promise): Promise { + const client = await this.pool.connect(); + try { + await client.query("SELECT pg_advisory_lock(hashtextextended($1, 0))", [ + id, + ]); + try { + return await fn(); + } finally { + await client.query( + "SELECT pg_advisory_unlock(hashtextextended($1, 0))", + [id], + ); + } + } finally { + client.release(); + } } private async subscribe( @@ -305,6 +344,21 @@ export class McpEvents { throw invalid("delivery.secret is required"); const refreshBefore = this.grant(params.ttlMs); const id = subscriptionId(principal, url, name, args); + return this.withKeyLock(id, () => + this.upsert({ id, principal, name, args, url, secret, refreshBefore }), + ); + } + + private async upsert(input: { + id: string; + principal: string; + name: string; + args: Record; + url: string; + secret: string; + refreshBefore: string | null; + }): Promise> { + const { id, principal, name, args, url, secret, refreshBefore } = input; const existing = await this.find(id); let verified = @@ -322,13 +376,20 @@ export class McpEvents { }; if (!existing) { - await this.hookgres.createEndpoint({ - tenantId: id, - url, - eventTypes: [name], - secret, - metadata, - }); + try { + await this.hookgres.createEndpoint({ + tenantId: id, + url, + eventTypes: [name], + secret, + metadata, + }); + } catch (error) { + if (error instanceof HookgresValidationError) { + throw invalid("delivery.url is not an allowed callback address"); + } + throw error; + } return { resultType: "complete", id, @@ -369,16 +430,20 @@ export class McpEvents { }); } const { name, args, url } = this.keyParams(params); - const existing = await this.find( - subscriptionId(principal, url, name, args), - ); - if (!existing) { - throw new McpEventsError(McpEventsErrorCode.NotFound, "NotFound", { - kind: "subscription", - }); - } - await this.hookgres.deleteEndpoint(existing.endpointId); - return { resultType: "complete" }; + const id = subscriptionId(principal, url, name, args); + return this.withKeyLock(id, async () => { + const { rows } = await this.pool.query<{ id: string }>( + `SELECT id FROM ${this.table} WHERE tenant_id = $1`, + [id], + ); + if (rows.length === 0) { + throw new McpEventsError(McpEventsErrorCode.NotFound, "NotFound", { + kind: "subscription", + }); + } + for (const row of rows) await this.hookgres.deleteEndpoint(row.id); + return { resultType: "complete" }; + }); } private keyParams(params: Record) { @@ -393,6 +458,9 @@ export class McpEvents { if (typeof args !== "object" || args === null || Array.isArray(args)) { throw invalid("arguments must be an object"); } + if (canonicalJson(args).length > MAX_ARGUMENTS_BYTES) { + throw invalid(`arguments must be at most ${MAX_ARGUMENTS_BYTES} bytes`); + } const delivery = params.delivery; if (typeof delivery !== "object" || delivery === null) { throw invalid("delivery is required"); @@ -460,6 +528,7 @@ export class McpEvents { { reason: "challenge_failed" }, ); } + this.limitVerifications(new URL(url).hostname); const challenge = randomBytes(24).toString("base64url"); const body = JSON.stringify({ type: "verification", challenge }); const messageId = `msg_verification_${randomBytes(16).toString("hex")}`; @@ -479,9 +548,10 @@ export class McpEvents { redirect: "manual", signal: AbortSignal.timeout(5_000), }); - const text = response.body - ? await new Response(response.body).text() - : ""; + const text = await readBounded( + response.body, + MAX_CHALLENGE_RESPONSE_BYTES, + ); if (response.status >= 200 && response.status < 300) { const echoed = parseChallenge(text); if (echoed !== null && safeEqual(echoed, challenge)) return; @@ -499,6 +569,22 @@ export class McpEvents { ); } + private limitVerifications(host: string) { + const now = Date.now(); + const recent = (this.verificationLog.get(host) ?? []).filter( + (at) => at > now - 60_000, + ); + if (recent.length >= this.verificationLimit) { + throw new McpEventsError( + McpEventsErrorCode.ResourceExhausted, + "Too many verification requests to this host", + { limit: "verifications", max: this.verificationLimit }, + ); + } + recent.push(now); + this.verificationLog.set(host, recent); + } + private async deliveryStatus(sub: Subscription) { const { rows } = await this.pool.query<{ last_success: Date | null; @@ -514,7 +600,8 @@ export class McpEvents { ); const row = rows[0]; return { - active: true, + // false when delivery was suspended; this refresh has just resumed it. + active: !sub.disabled, lastDeliveryAt: row?.last_success?.toISOString() ?? null, lastError: sub.failingSince && row?.last_error @@ -553,6 +640,29 @@ function canonicalJson(value: unknown): string { return JSON.stringify(value); } +/** Reads at most `limit` bytes of a body, then cancels the rest. */ +async function readBounded( + body: ReadableStream | null, + limit: number, +): Promise { + if (!body) return ""; + const reader = body.getReader(); + const chunks: Uint8Array[] = []; + let size = 0; + try { + while (size <= limit) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + size += value.byteLength; + } + } finally { + await reader.cancel().catch(() => {}); + } + if (size > limit) return ""; + return Buffer.concat(chunks).toString("utf8"); +} + function invalid(message: string): McpEventsError { return new McpEventsError(McpEventsErrorCode.InvalidParams, message); } diff --git a/test/mcp-events.test.ts b/test/mcp-events.test.ts index 4ec7f80..91b273c 100644 --- a/test/mcp-events.test.ts +++ b/test/mcp-events.test.ts @@ -237,6 +237,50 @@ describe("events/subscribe", () => { }); }); +describe("races and limits", () => { + it("creates one endpoint for concurrent subscribes with the same key", async () => { + // Verify the URL first, so the three calls race on find-then-create. + await subscribe({ arguments: { severity: "P9" } }); + const args = { arguments: { severity: "P1" } }; + // Warm the pool so the three calls get their own connections at once. + const clients = await Promise.all( + [1, 2, 3, 4, 5, 6].map(() => pool.connect()), + ); + for (const client of clients) client.release(); + await Promise.all([subscribe(args), subscribe(args), subscribe(args)]); + const { rows } = await pool.query( + `SELECT count(*)::int AS n FROM "${schema}".endpoints + WHERE metadata->'arguments'->>'severity' = 'P1'`, + ); + expect(rows[0].n).toBe(1); + }); + + it("rate-limits verification POSTs per host", async () => { + const limited = create({ verificationsPerHostPerMinute: 1 }); + receiver.respond = () => 200; + receiver.body = "{}"; + const attempt = (path: string) => + limited.handle( + "events/subscribe", + { + name: "incident.created", + delivery: { url: `${receiver.url}/${path}`, secret }, + }, + { principal: "user_1" }, + ); + await expectError(attempt("a"), McpEventsErrorCode.CallbackEndpointError); + await expectError(attempt("b"), McpEventsErrorCode.ResourceExhausted); + await limited.stop(); + }); + + it("caps the size of arguments", async () => { + await expectError( + subscribe({ arguments: { blob: "x".repeat(5000) } }), + McpEventsErrorCode.InvalidParams, + ); + }); +}); + describe("emit and delivery", () => { it("delivers only when the emitting transaction commits", async () => { const { id } = await subscribe();