diff --git a/src/routes/mcp.ts b/src/routes/mcp.ts index 4d30abc..167b1fc 100644 --- a/src/routes/mcp.ts +++ b/src/routes/mcp.ts @@ -1150,6 +1150,21 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync = async return } + // Access is only checked when the stream opens, so the stream ends + // when the identity that opened it expires. One that has already + // expired would be closed at once and reopened in a tight loop. + const authContext = await authContextFrom(request) + const expiresAt = authContext?.expiresAt instanceof Date ? authContext.expiresAt.getTime() : undefined + if (expiresAt !== undefined && Number.isFinite(expiresAt) && expiresAt <= Date.now()) { + // Shaped like every other 401 when this plugin runs OAuth, so the + // client finds the metadata it needs to re-authorize. + if (opts.authorization?.enabled) { + reply.header('WWW-Authenticate', generateWWWAuthenticateHeader(opts.authorization, { error: 'invalid_token', error_description: 'The access token expired' })) + } + reply.code(401).type('application/json').send({ error: 'Unauthorized: the access token expired' }) + return + } + request.log.info({ sessionId }, 'Handling SSE request') // Find or create the session before taking the reply over, so a store @@ -1221,9 +1236,9 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync = async // 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. // 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 + // identity that opened it: close when it expires (the token's exp, or the + // resolver's expiresAt) and the client reconnects with a fresh one. + const msUntilTokenExp = expiresAt !== undefined && Number.isFinite(expiresAt) ? Math.max(0, expiresAt - 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)) { diff --git a/test/sse-resolved-auth-expiry.test.ts b/test/sse-resolved-auth-expiry.test.ts new file mode 100644 index 0000000..00de016 --- /dev/null +++ b/test/sse-resolved-auth-expiry.test.ts @@ -0,0 +1,75 @@ +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 { MemorySessionStore } from '../src/stores/memory-session-store.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: number): Promise { + let timer: NodeJS.Timeout | undefined + try { + return await Promise.race([ + once(stream, 'end').then(() => true), + new Promise((resolve) => { + timer = setTimeout(() => resolve(false), ms) + }) + ]) + } finally { + clearTimeout(timer) + } +} + +async function buildApp (t: TestContext, expiresInMs: number) { + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { + enableSSE: true, + resolveAuthorizationContext: async () => ({ + userId: 'alice', + authorizationServer: 'https://app.example.com', + expiresAt: new Date(Date.now() + expiresInMs) + }) + }) + await app.ready() + return app +} + +describe('legacy SSE GET honours the resolved authorization context expiresAt', () => { + test('the stream closes when the resolved expiresAt passes', async (t: TestContext) => { + const app = await buildApp(t, 300) + + 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() + const ended = await waitForEnd(stream, 1500) + if (!ended) stream.destroy() + t.assert.ok(ended, 'the stream outlived the resolved expiresAt') + }) + + test('an already expired context is answered with 401 and no session', async (t: TestContext) => { + const app = await buildApp(t, -1000) + const create = t.mock.method(MemorySessionStore.prototype, 'create') + + const response = await app.inject({ + method: 'GET', + url: '/mcp', + payloadAsStream: true, + headers: { accept: 'text/event-stream' } + }) + response.stream().destroy() + + t.assert.strictEqual(response.statusCode, 401) + t.assert.strictEqual(response.headers['content-type'], 'application/json; charset=utf-8') + t.assert.strictEqual(response.headers['mcp-session-id'], undefined) + t.assert.strictEqual(create.mock.callCount(), 0) + }) +})