fix(sse): close GET /mcp streams when the opening token expires

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.

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<MCPPubSubRoutesOptions> = 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))
         }

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
   /**

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<void> {
+  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)
+  })
+})