fix(sse): end the legacy SSE stream at the resolved context's expiresAt

src/routes/mcp.ts

@@ -1150,6 +1150,21 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = 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<MCPPubSubRoutesOptions> = 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)) {

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<boolean> {
+  let timer: NodeJS.Timeout | undefined
+  try {
+    return await Promise.race([
+      once(stream, 'end').then(() => true),
+      new Promise<boolean>((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)
+  })
+})