diff --git a/.changeset/max-subscriptions-per-session.md b/.changeset/max-subscriptions-per-session.md new file mode 100644 index 0000000000..06c6e8c116 --- /dev/null +++ b/.changeset/max-subscriptions-per-session.md @@ -0,0 +1,7 @@ +--- +'@modelcontextprotocol/server': minor +--- + +Add `maxSubscriptionsPerSession` and `getSessionId` options to `createMcpHandler` for per-session `subscriptions/listen` fairness. Open listen streams are counted per session key resolved by `getSessionId(options, request)` and compared against the cap before the SSE ack; a refusal returns the same `-32603 "Subscription limit reached"` error as the existing handler-wide `maxSubscriptions` guard. The SDK does not authenticate the resolver's returned value. Deployments must derive the key from a trusted source (for example `authInfo.extra`), not from raw inbound headers. + +Fail-closed by construction: setting `maxSubscriptionsPerSession` without `getSessionId` throws `TypeError` at `createMcpHandler` time; a resolver returning `undefined` for a given listen refuses the stream rather than silently falling back to the handler-wide cap. Closes #3036. diff --git a/packages/server/src/server/createMcpHandler.ts b/packages/server/src/server/createMcpHandler.ts index 4ef836c636..deed3cde54 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -197,6 +197,41 @@ export interface CreateMcpHandlerOptions { * @default 1024 */ maxSubscriptions?: number; + /** + * Reject a new `subscriptions/listen` with `-32603` 'Subscription limit + * reached' (in-band, HTTP 200, before the ack) when this many subscription + * streams are already open under the same resolved session key (see + * {@linkcode getSessionId}). Enforced alongside the handler-wide + * {@linkcode maxSubscriptions} cap; whichever fires first wins and both + * return the same error. Requires {@linkcode getSessionId} to be + * configured. {@linkcode createMcpHandler} throws `TypeError` at + * construction otherwise, since without a resolver the SDK has no safe + * source for per-session attribution. + * @default no per-session cap + */ + maxSubscriptionsPerSession?: number; + /** + * Resolver for the trusted session key used by + * {@linkcode maxSubscriptionsPerSession} to attribute a + * `subscriptions/listen` stream to a session bucket. Required when + * `maxSubscriptionsPerSession` is configured, ignored otherwise. + * + * Receives the pass-through {@linkcode McpHandlerRequestOptions} (whose + * `authInfo` reflects what trusted middleware established) and the inbound + * `Request`; returns the session key when identity is established, + * `undefined` when it is not. The SDK does **not** authenticate the + * returned value: it is used only as a bucketing key for the per-session + * cap, so the deployment must derive it from a trusted source (for + * example `authInfo.extra`), never from an unvalidated inbound header + * forwarded as-is. + * + * **Fail-closed contract:** when this resolver returns `undefined` for a + * listen request, the stream is refused pre-ack with + * `-32603 'Subscription limit reached'` rather than silently falling back + * to the handler-wide cap. A misconfigured deployment therefore cannot + * accidentally appear to be enforcing isolation that was never applied. + */ + getSessionId?: (options: McpHandlerRequestOptions, request: Request) => string | undefined; /** * SSE comment-frame keepalive interval for every SSE stream this handler * serves. In modern `auto` mode it starts after SSE upgrade. Set to `0` to disable. @@ -693,6 +728,17 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa ); } + // Fail-closed configuration guard: per-session attribution has no safe + // default source. If the cap is set without a resolver, there is no honest + // way to bucket a listen request. Reading `Mcp-Session-Id` off raw headers + // would claim isolation the SDK cannot vouch for. Reject at construction. + if (options.maxSubscriptionsPerSession !== undefined && options.getSessionId === undefined) { + throw new TypeError( + "The 'maxSubscriptionsPerSession' option requires a 'getSessionId' resolver to supply the trusted session key. " + + 'Without a resolver the SDK has no safe source for per-session attribution. The entry does not authenticate inbound headers.' + ); + } + /** Modern per-request instances with an exchange still in flight (close() tears these down). */ const inflight = new Set(); let closed = false; @@ -710,6 +756,7 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa const listenRouter = createListenRouter({ bus, maxSubscriptions: options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS, + ...(options.maxSubscriptionsPerSession !== undefined && { maxSubscriptionsPerSession: options.maxSubscriptionsPerSession }), keepAliveMs: options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS, onerror: reportError }); @@ -726,7 +773,12 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa const legacyHandler: LegacyHttpHandler | undefined = legacy === 'reject' ? undefined : createLegacyStatelessFallback(factory, reportError, options.keepAliveMs, maxRequestBodySize); - async function serveModern(route: InboundModernRoute, request: Request, authInfo: AuthInfo | undefined): Promise { + async function serveModern( + route: InboundModernRoute, + request: Request, + requestOptions: McpHandlerRequestOptions | undefined, + authInfo: AuthInfo | undefined + ): Promise { const claimedRevision = route.classification.revision; if (claimedRevision === undefined || !SUPPORTED_MODERN_PROTOCOL_VERSIONS.includes(claimedRevision)) { // The claim names a revision this endpoint does not serve (an @@ -811,7 +863,18 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa const capabilities = server.getCapabilities(); const serverInfo = serverIdentityOf(server); void product.close().catch(reportError); - return listenRouter.serve(route.message, request.signal, capabilities, serverInfo); + // Resolve the per-session bucketing key through the caller-supplied + // getSessionId only when the per-session cap is actually configured. + // The resolver is documented as ignored otherwise, so skipping it + // avoids a user-supplied throw turning an otherwise valid listen + // into a 500. Deployments decide the trust source for identity + // (typically a field stamped onto authInfo by their auth + // middleware); the SDK never derives the key from request headers + // directly. Construction already rejected the "cap configured but + // resolver missing" misconfiguration. + const sessionId = + options.maxSubscriptionsPerSession === undefined ? undefined : options.getSessionId?.(requestOptions ?? {}, request); + return listenRouter.serve(route.message, request.signal, capabilities, serverInfo, sessionId); } // SEP-2243 `Mcp-Param-*` server-side validation (pre-dispatch ladder @@ -984,7 +1047,7 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa return rejectionResponse(outcome, echoableRequestId(body)); } case 'modern': { - return await serveModern(outcome, request, authInfo); + return await serveModern(outcome, request, requestOptions, authInfo); } case 'legacy': { return await serveLegacyRoute(outcome, forwardRequest, authInfo, parsedBody); diff --git a/packages/server/src/server/listenRouter.ts b/packages/server/src/server/listenRouter.ts index e58807d5c1..aa6f19feee 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -45,6 +45,22 @@ export interface ListenRouterOptions { bus: ServerEventBus; /** Reject a new listen with `-32603` when this many subscriptions are already open (default 1024). */ maxSubscriptions?: number; + /** + * Reject a new listen with `-32603` when this many subscriptions are already + * open under the same resolved session key. The caller (the HTTP entry) is + * responsible for resolving the session key through a trusted path and + * passing it to {@linkcode serve}; this router does not read HTTP headers. + * + * Fail-closed: when `maxSubscriptionsPerSession` is set and the caller + * passes an undefined `sessionId` for a given listen, the stream is + * refused with `-32603` rather than silently falling back to the + * handler-wide cap. A misconfigured deployment cannot accidentally + * appear to be enforcing per-session isolation that was never applied. + * + * No default: when omitted the per-session guard is off and only the + * handler-wide {@linkcode maxSubscriptions} cap applies. + */ + maxSubscriptionsPerSession?: number; /** SSE comment-frame keepalive interval; `0` disables keepalive (default 15000). */ keepAliveMs?: number; /** Out-of-band error reporting (never alters the response). */ @@ -108,7 +124,13 @@ export interface ListenRouter { * graceful-close result's `_meta` (the spec's `SubscriptionsListenResultMetaObject` * extends `ResultMetaObject`, so the serverInfo SHOULD applies there too). */ - serve(message: JSONRPCRequest, signal: AbortSignal | undefined, capabilities: ServerCapabilities, serverInfo: Implementation): Response; + serve( + message: JSONRPCRequest, + signal: AbortSignal | undefined, + capabilities: ServerCapabilities, + serverInfo: Implementation, + sessionId?: string + ): Response; /** * Gracefully close every open subscription stream: emits the empty * `subscriptions/listen` JSON-RPC result (the spec's graceful-close @@ -122,21 +144,51 @@ export interface ListenRouter { export function createListenRouter(options: ListenRouterOptions): ListenRouter { const { bus, onerror } = options; const maxSubscriptions = options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS; + const maxSubscriptionsPerSession = options.maxSubscriptionsPerSession; const keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; const open = new Set<(graceful: boolean) => void>(); + // Per-session teardown handles, keyed on the transport-managed session id. + // Grown alongside `open` so the per-session cap can be enforced without a + // scan. A session's entry is removed when its last stream tears down. + const openPerSession = new Map void>>(); function serve( message: JSONRPCRequest, signal: AbortSignal | undefined, capabilities: ServerCapabilities, - serverInfo: Implementation + serverInfo: Implementation, + sessionId?: string ): Response { // Capacity guard, pre-ack: in-band -32603 on HTTP 200. if (open.size >= maxSubscriptions) { onerror?.(new Error(`subscriptions/listen refused: subscription limit reached (${maxSubscriptions})`)); return jsonRpcError(message.id, -32_603, 'Subscription limit reached'); } + if (maxSubscriptionsPerSession !== undefined) { + // Fail-closed: the per-session cap is only honest with a resolved + // key. The caller's resolver returned undefined for this listen, + // so there is no bucket to attribute it to. Reject rather than + // silently fall through to the handler-wide cap and misreport the + // guarantee. + if (sessionId === undefined) { + onerror?.( + new Error( + 'subscriptions/listen refused: per-session cap configured but no trusted session key was resolved for this request' + ) + ); + return jsonRpcError(message.id, -32_603, 'Subscription limit reached'); + } + const sessionOpen = openPerSession.get(sessionId); + if (sessionOpen !== undefined && sessionOpen.size >= maxSubscriptionsPerSession) { + onerror?.( + new Error( + `subscriptions/listen refused: per-session subscription limit reached (${maxSubscriptionsPerSession}, session ${sessionId})` + ) + ); + return jsonRpcError(message.id, -32_603, 'Subscription limit reached'); + } + } const filter = parseListenFilter(message); if (filter === undefined) { return jsonRpcError(message.id, -32_602, "Invalid params: 'notifications' is required and must be a valid SubscriptionFilter"); @@ -194,6 +246,13 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); abortCleanup?.(); open.delete(teardown); + if (sessionId !== undefined) { + const sessionOpen = openPerSession.get(sessionId); + if (sessionOpen !== undefined) { + sessionOpen.delete(teardown); + if (sessionOpen.size === 0) openPerSession.delete(sessionId); + } + } try { controller.close(); } catch { @@ -229,6 +288,14 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { keepAliveTimer = armSseKeepAlive(keepAliveMs, () => writeFrame(': keepalive\n\n')); open.add(teardown); + if (sessionId !== undefined) { + let sessionOpen = openPerSession.get(sessionId); + if (sessionOpen === undefined) { + sessionOpen = new Set(); + openPerSession.set(sessionId, sessionOpen); + } + sessionOpen.add(teardown); + } }, cancel() { // The client closed the SSE stream — the spec's HTTP cancel diff --git a/packages/server/test/server/createMcpHandlerListen.test.ts b/packages/server/test/server/createMcpHandlerListen.test.ts index 579e61d854..4ecc22ce72 100644 --- a/packages/server/test/server/createMcpHandlerListen.test.ts +++ b/packages/server/test/server/createMcpHandlerListen.test.ts @@ -26,15 +26,17 @@ const ENVELOPE = { [CLIENT_CAPABILITIES_META_KEY]: {} }; -function listenRequest(id: string | number, filter: Record): Request { +function listenRequest(id: string | number, filter: Record, sessionId?: string): Request { + const headers: Record = { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'mcp-protocol-version': MODERN_REVISION, + 'mcp-method': 'subscriptions/listen' + }; + if (sessionId !== undefined) headers['mcp-session-id'] = sessionId; return new Request('http://localhost/mcp', { method: 'POST', - headers: { - 'Content-Type': 'application/json', - Accept: 'application/json, text/event-stream', - 'mcp-protocol-version': MODERN_REVISION, - 'mcp-method': 'subscriptions/listen' - }, + headers, body: JSON.stringify({ jsonrpc: '2.0', id, @@ -206,6 +208,156 @@ describe('createMcpHandler — subscriptions/listen', () => { await handler.close(); }); + it('enforces maxSubscriptionsPerSession per resolved session key, independently of other sessions', async () => { + // Per-session cap: two sessions, cap of 1 each. Each session's second + // listen is refused with -32603; the other session is unaffected. The + // test resolver reads from the mcp-session-id header for shape, which + // is the deployment's trust choice. The SDK itself does not consult + // request headers. + const handler = createMcpHandler(trivialFactory(), { + keepAliveMs: 0, + maxSubscriptions: 1024, + maxSubscriptionsPerSession: 1, + getSessionId: (_opts, req) => req.headers.get('mcp-session-id') ?? undefined + }); + const sessionA = 'session-a'; + const sessionB = 'session-b'; + + const aFirst = await handler.fetch(listenRequest(1, { toolsListChanged: true }, sessionA)); + expect(aFirst.headers.get('Content-Type')).toBe('text/event-stream'); + + const aSecond = await handler.fetch(listenRequest(2, { toolsListChanged: true }, sessionA)); + expect(aSecond.headers.get('Content-Type')).toContain('application/json'); + const aBody = (await aSecond.json()) as { error: { code: number; message: string }; id: unknown }; + expect(aBody.error.code).toBe(-32_603); + expect(aBody.error.message).toBe('Subscription limit reached'); + expect(aBody.id).toBe(2); + + // Session B still has capacity even though session A is full. + const bFirst = await handler.fetch(listenRequest(3, { toolsListChanged: true }, sessionB)); + expect(bFirst.headers.get('Content-Type')).toBe('text/event-stream'); + + // Session B now also at its cap. + const bSecond = await handler.fetch(listenRequest(4, { toolsListChanged: true }, sessionB)); + const bBody = (await bSecond.json()) as { error: { code: number; message: string }; id: unknown }; + expect(bBody.error.code).toBe(-32_603); + expect(bBody.id).toBe(4); + + await aFirst.body!.cancel(); + await bFirst.body!.cancel(); + await handler.close(); + }); + + it('handler-wide maxSubscriptions still fires across sessions even when per-session cap has room', async () => { + // Three sessions, each well under a generous per-session cap, but the + // handler-wide cap of 2 fires on the third open regardless of session. + const handler = createMcpHandler(trivialFactory(), { + keepAliveMs: 0, + maxSubscriptions: 2, + maxSubscriptionsPerSession: 10, + getSessionId: (_opts, req) => req.headers.get('mcp-session-id') ?? undefined + }); + + const one = await handler.fetch(listenRequest(1, { toolsListChanged: true }, 'session-a')); + expect(one.headers.get('Content-Type')).toBe('text/event-stream'); + const two = await handler.fetch(listenRequest(2, { toolsListChanged: true }, 'session-b')); + expect(two.headers.get('Content-Type')).toBe('text/event-stream'); + + const three = await handler.fetch(listenRequest(3, { toolsListChanged: true }, 'session-c')); + expect(three.headers.get('Content-Type')).toContain('application/json'); + const body = (await three.json()) as { error: { code: number; message: string }; id: unknown }; + expect(body.error.code).toBe(-32_603); + expect(body.error.message).toBe('Subscription limit reached'); + expect(body.id).toBe(3); + + await one.body!.cancel(); + await two.body!.cancel(); + await handler.close(); + }); + + it('throws TypeError at construction when maxSubscriptionsPerSession is set without getSessionId', () => { + // Fail-closed configuration: per-session attribution has no safe + // default source. The entry refuses to construct a handler that would + // appear to enforce per-session isolation without a trusted resolver. + expect(() => createMcpHandler(trivialFactory(), { maxSubscriptionsPerSession: 1 })).toThrow(TypeError); + }); + + it('refuses pre-ack with -32603 when getSessionId returns undefined for a configured per-session cap', async () => { + // Fail-closed runtime contract: the resolver signaled it cannot + // attribute this listen (no trusted identity available), so the SDK + // refuses rather than silently applying only the handler-wide cap and + // misreporting the guarantee. + const handler = createMcpHandler(trivialFactory(), { + keepAliveMs: 0, + maxSubscriptionsPerSession: 1, + getSessionId: () => undefined + }); + const response = await handler.fetch(listenRequest(1, { toolsListChanged: true }, 'session-a')); + expect(response.headers.get('Content-Type')).toContain('application/json'); + const body = (await response.json()) as { error: { code: number; message: string }; id: unknown }; + expect(body.error.code).toBe(-32_603); + expect(body.error.message).toBe('Subscription limit reached'); + expect(body.id).toBe(1); + await handler.close(); + }); + + it('releases a per-session slot when the client cancels the listen stream', async () => { + // Slot accounting on cancel: the client drops the SSE stream, teardown + // decrements both the handler-wide and per-session buckets, and a + // subsequent listen on the same session is accepted. + const handler = createMcpHandler(trivialFactory(), { + keepAliveMs: 0, + maxSubscriptionsPerSession: 1, + getSessionId: (_opts, req) => req.headers.get('mcp-session-id') ?? undefined + }); + + const first = await handler.fetch(listenRequest(1, { toolsListChanged: true }, 'session-a')); + expect(first.headers.get('Content-Type')).toBe('text/event-stream'); + await first.body!.cancel(); + + const second = await handler.fetch(listenRequest(2, { toolsListChanged: true }, 'session-a')); + expect(second.headers.get('Content-Type')).toBe('text/event-stream'); + await second.body!.cancel(); + await handler.close(); + }); + + it('does not consume a per-session slot for a listen whose honored filter narrows to empty', async () => { + // Empty-honored listen (server declares no capabilities): the stream + // acks and immediately completes, never adding the teardown to the + // open set. A subsequent listen on the same session must still be + // accepted under a cap of 1. + const bareFactory = () => new McpServer({ name: 'listen-test-server', version: '1.0.0' }, { capabilities: {} }); + const handler = createMcpHandler(bareFactory, { + keepAliveMs: 0, + maxSubscriptionsPerSession: 1, + getSessionId: (_opts, req) => req.headers.get('mcp-session-id') ?? undefined + }); + + const emptyListen = await handler.fetch(listenRequest(1, { toolsListChanged: true }, 'session-a')); + // Drain to the server-side graceful close so the auto-completion runs. + await emptyListen.text(); + + const second = await handler.fetch(listenRequest(2, { toolsListChanged: true }, 'session-a')); + expect(second.headers.get('Content-Type')).toBe('text/event-stream'); + await second.body!.cancel(); + await handler.close(); + }); + + it('ignores getSessionId when maxSubscriptionsPerSession is unset, even if it would throw', async () => { + // The resolver is documented as ignored without the per-session cap. + // Verify it is not invoked, so a user-supplied throw cannot turn a + // legitimate listen into a 500 when the cap is off. + const getSessionId = vi.fn(() => { + throw new Error('resolver should not be called when maxSubscriptionsPerSession is unset'); + }); + const handler = createMcpHandler(trivialFactory(), { keepAliveMs: 0, getSessionId }); + const response = await handler.fetch(listenRequest(1, { toolsListChanged: true }, 'session-a')); + expect(response.headers.get('Content-Type')).toBe('text/event-stream'); + expect(getSessionId).not.toHaveBeenCalled(); + await response.body!.cancel(); + await handler.close(); + }); + it("rejects with -32602 when params.notifications is absent (spec marks 'notifications' REQUIRED)", async () => { const handler = createMcpHandler(trivialFactory(), { keepAliveMs: 0 }); const response = await handler.fetch(