From df3d11d129bc1540155a551fcd1cb0cd05c09f77 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 13 Oct 2026 13:19:03 +0200 Subject: [PATCH 1/3] fix(sse): close GET /mcp streams when the opening token expires Access is checked only when a long-lived SSE stream opens, so a stream kept receiving notifications after the bearer token that opened it had expired. When authorization is enabled and the token carries an `exp`, schedule an unref'd timer that ends the stream at exp, cleared when the stream closes. The client reconnects with a fresh token and resumes with Last-Event-ID. The existing sseMaxConnectionMs option (SEP-1699) already provides a configurable maximum stream lifetime; the stream now closes at min(exp, sseMaxConnectionMs). Delays beyond setTimeout's 2^31-1 ms limit are skipped instead of firing immediately. Fixes #254 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UF4S3w7iA83V9DVsasLJzM --- README.md | 2 + src/routes/mcp.ts | 25 ++++++-- src/types.ts | 4 +- test/sse-token-expiry.test.ts | 111 ++++++++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 7 deletions(-) create mode 100644 test/sse-token-expiry.test.ts diff --git a/README.md b/README.md index 4e027cbe..8c4efc16 100644 --- a/README.md +++ b/README.md @@ -815,6 +815,8 @@ await app.register(mcpPlugin, { }) ``` +When authorization is enabled, the access token is only checked when a `GET /mcp` stream opens. So that a stream never outlives that token, the server closes it at the token's `exp` (or after `sseMaxConnectionMs`, whichever comes first). The client then reconnects with a fresh token and resumes with `Last-Event-ID`. Tokens without an `exp` claim are bounded only by `sseMaxConnectionMs`, so set it if your authorization server issues them or if you want revocation and scope changes to take effect within a bounded time. + ## Redis Configuration for Horizontal Scaling The plugin supports Redis-backed session management and message broadcasting for horizontal scaling across multiple server instances. diff --git a/src/routes/mcp.ts b/src/routes/mcp.ts index 92e49a8d..01933548 100644 --- a/src/routes/mcp.ts +++ b/src/routes/mcp.ts @@ -22,6 +22,8 @@ import type { JsonSchemaValidator } from '../validation/json-schema-validator.ts import { processMessage, checkRequestId } from '../handlers.ts' import { isStdioRequest } from '../stdio-marker.ts' +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1 + interface MCPPubSubRoutesOptions { enableSSE: boolean opts: MCPPluginOptions @@ -609,19 +611,30 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync = async // SEP-1699: servers may end an SSE stream whenever they like, turning the // stream into a polling channel. The client reconnects with Last-Event-ID // on GET and we replay whatever it missed, so closing here loses nothing. - let maxDurationTimer: NodeJS.Timeout | undefined - if (opts.sseMaxConnectionMs) { - maxDurationTimer = setTimeout(() => { + // Access is only checked when the stream opens, so never let it outlive the + // token that opened it: close at exp and the client reconnects with a fresh one. + const tokenExp = request.tokenPayload?.exp + const msUntilTokenExp = typeof tokenExp === 'number' ? Math.max(0, tokenExp * 1000 - Date.now()) : undefined + let closeAfterMs = opts.sseMaxConnectionMs || undefined + let closeReason = 'Closing SSE stream to let the client poll; it may resume with Last-Event-ID' + if (msUntilTokenExp !== undefined && (closeAfterMs === undefined || msUntilTokenExp < closeAfterMs)) { + closeAfterMs = msUntilTokenExp + closeReason = 'Closing SSE stream because the token that opened it expired; it may reconnect with a fresh token' + } + + // setTimeout fires immediately on delays past 2^31-1 ms (~24.8 days) + if (closeAfterMs !== undefined && closeAfterMs <= MAX_TIMER_DELAY_MS) { + const maxDurationTimer = setTimeout(() => { app.log.info({ sessionId: session.id, - afterMs: opts.sseMaxConnectionMs - }, 'Closing SSE stream to let the client poll; it may resume with Last-Event-ID') + afterMs: closeAfterMs + }, closeReason) try { reply.raw.end() } catch { // already gone } - }, opts.sseMaxConnectionMs) + }, closeAfterMs) maxDurationTimer.unref() reply.raw.on('close', () => clearTimeout(maxDurationTimer)) diff --git a/src/types.ts b/src/types.ts index f2f59667..70b427f3 100644 --- a/src/types.ts +++ b/src/types.ts @@ -290,7 +290,9 @@ export interface MCPPluginOptions { /** * Close an SSE stream after this many milliseconds so the client falls back to * polling and reconnects with `Last-Event-ID` (SEP-1699). Omit to keep streams - * open indefinitely, which stays valid. + * open indefinitely, which stays valid. With authorization enabled, a stream is + * also closed when the token that opened it expires (its `exp` claim), whichever + * comes first, so the client reconnects with a fresh token. */ sseMaxConnectionMs?: number /** diff --git a/test/sse-token-expiry.test.ts b/test/sse-token-expiry.test.ts new file mode 100644 index 00000000..fa0a596e --- /dev/null +++ b/test/sse-token-expiry.test.ts @@ -0,0 +1,111 @@ +import { test, describe } from 'node:test' +import type { TestContext } from 'node:test' +import { once } from 'node:events' +import Fastify from 'fastify' +import mcpPlugin from '../src/index.ts' +import { + createTestAuthConfig, + createTestJWT, + setupMockAgent, + generateMockJWKSResponse +} from './auth-test-utils.ts' + +function countRefedTimeouts (): number { + return process.getActiveResourcesInfo().filter(r => r === 'Timeout').length +} + +describe('SSE streams close when the opening token expires', () => { + test('a stream opened with a short-lived JWT is closed shortly after exp', async (t: TestContext) => { + const restoreMock = setupMockAgent({ + 'https://auth.example.com/.well-known/jwks.json': generateMockJWKSResponse() + }) + t.after(restoreMock) + + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { + enableSSE: true, + // The token expires well before this, so exp wins + sseMaxConnectionMs: 60_000, + authorization: createTestAuthConfig() + }) + await app.ready() + + const exp = Math.floor(Date.now() / 1000) + 2 + const token = createTestJWT({ exp }) + + const response = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { + accept: 'text/event-stream', + authorization: `Bearer ${token}` + } + }) + t.assert.strictEqual(response.statusCode, 200) + + const stream = response.stream() + stream.resume() + await once(stream, 'end', { signal: AbortSignal.timeout(5000) }) + + const closedAt = Date.now() + t.assert.ok(closedAt >= exp * 1000 - 50, `closed ${exp * 1000 - closedAt}ms before exp`) + t.assert.ok(closedAt < exp * 1000 + 1000, `closed ${closedAt - exp * 1000}ms after exp`) + }) + + test('the expiry timer does not keep the process alive', async (t: TestContext) => { + const restoreMock = setupMockAgent({ + 'https://auth.example.com/.well-known/jwks.json': generateMockJWKSResponse() + }) + t.after(restoreMock) + + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { + enableSSE: true, + authorization: createTestAuthConfig() + }) + await app.ready() + + const token = createTestJWT({ exp: Math.floor(Date.now() / 1000) + 3600 }) + // Warm up so the JWKS fetch does not leave timers behind during the measurement + const warmup = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream', authorization: `Bearer ${token}` } + }) + warmup.stream().destroy() + + const before = countRefedTimeouts() + const response = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream', authorization: `Bearer ${token}` } + }) + t.assert.strictEqual(response.statusCode, 200) + t.assert.strictEqual(countRefedTimeouts(), before) + response.stream().destroy() + }) + + test('without authorization, sseMaxConnectionMs alone still applies', async (t: TestContext) => { + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { enableSSE: true, sseMaxConnectionMs: 50 }) + await app.ready() + + const response = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream' } + }) + t.assert.strictEqual(response.statusCode, 200) + + const stream = response.stream() + stream.resume() + await once(stream, 'end', { signal: AbortSignal.timeout(5000) }) + }) +}) From a09147b14173881d3a61116eba6427c8d922e409 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 13 Oct 2026 13:28:10 +0200 Subject: [PATCH 2/3] test: keep the event loop alive while waiting for SSE close The close timer and AbortSignal.timeout are both unref'd, so in CI the event loop could drain before the stream ended and node:test cancelled the suite. Wait with a ref'd timer instead. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UF4S3w7iA83V9DVsasLJzM --- test/sse-token-expiry.test.ts | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/test/sse-token-expiry.test.ts b/test/sse-token-expiry.test.ts index fa0a596e..e7f36928 100644 --- a/test/sse-token-expiry.test.ts +++ b/test/sse-token-expiry.test.ts @@ -10,6 +10,22 @@ import { generateMockJWKSResponse } from './auth-test-utils.ts' +// The stream's close timer is unref'd, so a ref'd timer has to keep the event +// loop alive while the test waits for it to fire +async function waitForEnd (stream: NodeJS.ReadableStream, ms = 5000): Promise { + let timer: NodeJS.Timeout | undefined + try { + await Promise.race([ + once(stream, 'end'), + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`stream did not end within ${ms}ms`)), ms) + }) + ]) + } finally { + clearTimeout(timer) + } +} + function countRefedTimeouts (): number { return process.getActiveResourcesInfo().filter(r => r === 'Timeout').length } @@ -47,7 +63,7 @@ describe('SSE streams close when the opening token expires', () => { const stream = response.stream() stream.resume() - await once(stream, 'end', { signal: AbortSignal.timeout(5000) }) + await waitForEnd(stream) const closedAt = Date.now() t.assert.ok(closedAt >= exp * 1000 - 50, `closed ${exp * 1000 - closedAt}ms before exp`) @@ -106,6 +122,6 @@ describe('SSE streams close when the opening token expires', () => { const stream = response.stream() stream.resume() - await once(stream, 'end', { signal: AbortSignal.timeout(5000) }) + await waitForEnd(stream) }) }) From c71bfab36b47c74d963b525dddbdce2a239f276c Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 14 Oct 2026 12:09:22 +0200 Subject: [PATCH 3/3] fix(sse): close streams for tokens expiring past the setTimeout limit A token whose exp was more than 2^31-1 ms (~24.8 days) away skipped the close timer entirely, so the stream outlived the token. Wait in capped, unref'd steps until the deadline instead, and clear the pending timer on close. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UF4S3w7iA83V9DVsasLJzM --- src/routes/mcp.ts | 24 +++++++++++--- test/sse-token-expiry.test.ts | 61 +++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 5 deletions(-) diff --git a/src/routes/mcp.ts b/src/routes/mcp.ts index 01933548..5a9a46c7 100644 --- a/src/routes/mcp.ts +++ b/src/routes/mcp.ts @@ -622,9 +622,10 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync = async closeReason = 'Closing SSE stream because the token that opened it expired; it may reconnect with a fresh token' } - // setTimeout fires immediately on delays past 2^31-1 ms (~24.8 days) - if (closeAfterMs !== undefined && closeAfterMs <= MAX_TIMER_DELAY_MS) { - const maxDurationTimer = setTimeout(() => { + if (closeAfterMs !== undefined) { + const deadline = Date.now() + closeAfterMs + let maxDurationTimer: NodeJS.Timeout | undefined + const closeStream = () => { app.log.info({ sessionId: session.id, afterMs: closeAfterMs @@ -634,8 +635,21 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync = async } catch { // already gone } - }, closeAfterMs) - maxDurationTimer.unref() + } + // setTimeout fires immediately on delays past 2^31-1 ms (~24.8 days), so + // wait in capped steps until the deadline is reached + const scheduleClose = () => { + const remaining = Math.max(0, deadline - Date.now()) + maxDurationTimer = setTimeout(() => { + if (Date.now() >= deadline) { + closeStream() + } else { + scheduleClose() + } + }, Math.min(remaining, MAX_TIMER_DELAY_MS)) + maxDurationTimer.unref() + } + scheduleClose() reply.raw.on('close', () => clearTimeout(maxDurationTimer)) } diff --git a/test/sse-token-expiry.test.ts b/test/sse-token-expiry.test.ts index e7f36928..3ea11e44 100644 --- a/test/sse-token-expiry.test.ts +++ b/test/sse-token-expiry.test.ts @@ -1,6 +1,7 @@ import { test, describe } from 'node:test' import type { TestContext } from 'node:test' import { once } from 'node:events' +import { setImmediate as tick } from 'node:timers/promises' import Fastify from 'fastify' import mcpPlugin from '../src/index.ts' import { @@ -70,6 +71,66 @@ describe('SSE streams close when the opening token expires', () => { t.assert.ok(closedAt < exp * 1000 + 1000, `closed ${closedAt - exp * 1000}ms after exp`) }) + test('a stream opened with a token that expires past the setTimeout limit still closes at exp', async (t: TestContext) => { + const restoreMock = setupMockAgent({ + 'https://auth.example.com/.well-known/jwks.json': generateMockJWKSResponse() + }) + t.after(restoreMock) + + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { + enableSSE: true, + authorization: createTestAuthConfig() + }) + await app.ready() + + const thirtyDaysMs = 30 * 24 * 60 * 60 * 1000 + const exp = Math.floor(Date.now() / 1000) + thirtyDaysMs / 1000 + const token = createTestJWT({ exp }) + + // Warm up so the JWKS is fetched before timers are mocked + const warmup = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream', authorization: `Bearer ${token}` } + }) + warmup.stream().destroy() + + t.mock.timers.enable({ apis: ['setTimeout', 'Date'], now: Date.now() }) + + const response = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream', authorization: `Bearer ${token}` } + }) + t.assert.strictEqual(response.statusCode, 200) + + const stream = response.stream() + let ended = false + stream.on('end', () => { ended = true }) + stream.resume() + t.after(() => stream.destroy()) + + async function flush () { + for (let i = 0; i < 5; i++) await tick() + } + + await flush() + t.assert.strictEqual(ended, false, 'stream closed immediately') + + // Past the 2^31-1 ms setTimeout limit but still before exp + t.mock.timers.tick(exp * 1000 - Date.now() - 1000) + await flush() + t.assert.strictEqual(ended, false, 'stream closed before exp') + + t.mock.timers.tick(2000) + await flush() + t.assert.strictEqual(ended, true, 'stream still open after exp') + }) + test('the expiry timer does not keep the process alive', async (t: TestContext) => { const restoreMock = setupMockAgent({ 'https://auth.example.com/.well-known/jwks.json': generateMockJWKSResponse()