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..5a9a46c7 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,20 +611,45 @@ 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' + } + + if (closeAfterMs !== undefined) { + const deadline = Date.now() + closeAfterMs + let maxDurationTimer: NodeJS.Timeout | undefined + const closeStream = () => { 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) - 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/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..3ea11e44 --- /dev/null +++ b/test/sse-token-expiry.test.ts @@ -0,0 +1,188 @@ +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 { + createTestAuthConfig, + createTestJWT, + setupMockAgent, + 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 +} + +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 waitForEnd(stream) + + 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('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() + }) + 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 waitForEnd(stream) + }) +})