From 97c7145791d3c0e08f98643254240c2bd0c88144 Mon Sep 17 00:00:00 2001 From: FilipBl4gojevic <84703395+FilipBl4gojevic@users.noreply.github.com> Date: Fri, 31 Jul 2026 14:54:11 +0200 Subject: [PATCH] fix(builder): reap idle /mcp sessions and cap sessions per identity - close the anonymous session-exhaustion vector --- packages/builder/src/index.ts | 112 +++++++++++++ .../streamable-http-gating.test.ts | 153 +++++++++++++++++- 2 files changed, 262 insertions(+), 3 deletions(-) diff --git a/packages/builder/src/index.ts b/packages/builder/src/index.ts index 8b919b5..1561fd6 100644 --- a/packages/builder/src/index.ts +++ b/packages/builder/src/index.ts @@ -54,6 +54,79 @@ const transports = new Map(); const sessionIdentities = new Map(); const streamableTransports = new Map(); const streamableIdentities = new Map(); +// Last activity per /mcp session, bumped on every request that resolves to a +// session (initialize, and every subsequent handleRequest). The SSE pair +// self-limits - a dropped socket fires res.on('close') and drains its maps +// with no help from us. /mcp has no such signal: a caller can POST +// initialize, take the session id, and drop the connection without ever +// sending DELETE. Nothing in the SDK notices - the transport, its McpServer, +// and its 24 registered tools sit in memory until process exit. This map +// plus the reaper below is what bounds that. +const streamableLastSeen = new Map(); + +// Upper bound on concurrent /mcp sessions per identity. A caller that forgets +// to DELETE self-bounds instead of leaking sessions forever, and a flood of +// anon: sessions from one source cannot grow past this many concurrent +// ones. A GLOBAL cap is the reverse-proxy/network tier's job (clientIpOf's +// trust-model docstring in auth.ts is explicit about that split - this +// server trusts exactly one upstream proxy hop); this is the app-layer +// bound, same philosophy as the per-identity rate limiter in +// vector/rate-limiter.ts. Exported for visibility/tests; production tuning +// is via the env var, since index.ts is an executable entrypoint (importing +// it boots a real server as a side effect) rather than something a test can +// import and reconfigure in-process. +export const MAX_SESSIONS_PER_IDENTITY = parseInt(process.env.VECTOR_MCP_MAX_SESSIONS_PER_IDENTITY || '32'); + +// Idle bound: a session with no request for this long is reaped on the next +// sweep, regardless of whether DELETE ever arrives. Default 10 minutes - +// generous for a real client mid-conversation, short enough that an +// abandoned session does not outlive the process. +const STREAMABLE_SESSION_IDLE_MS = parseInt(process.env.VECTOR_MCP_SESSION_IDLE_MS || String(10 * 60 * 1000)); +// Sweep cadence, kept separate from the idle bound so a hermetic test can +// shrink both independently without waiting anywhere near the real idle +// window. +const STREAMABLE_SESSION_SWEEP_MS = parseInt(process.env.VECTOR_MCP_SESSION_SWEEP_MS || '60000'); + +/** + * Closes a /mcp session and drains it from every map, from any trigger other + * than the SDK's own DELETE handling (the idle reaper, and the per-identity + * cap's eviction of the oldest session). transport.close() drives our own + * transport.onclose callback below (which already deletes the same three + * keys) - the explicit deletes here are belt-and-suspenders so a caller of + * this function can rely on the maps being drained even if that callback + * were ever missing or delayed. + */ +async function closeStreamableSession(sessionId: string): Promise { + const transport = streamableTransports.get(sessionId); + if (transport) { + try { + await transport.close(); + } catch (err) { + console.error(`Error closing Streamable HTTP session ${sessionId}:`, (err as Error).message); + } + } + streamableTransports.delete(sessionId); + streamableIdentities.delete(sessionId); + streamableLastSeen.delete(sessionId); +} + +// The idle reaper. unref()'d so it never holds the process open on its own +// (matches every other background timer in this process - there are none +// today, but the convention is set here for the next one), and cleared on +// the http server's own 'close' event so a test that spawns and kills many +// server processes never leaks a live interval into a process that outlives +// the test itself. +const streamableReapInterval = setInterval(() => { + const now = Date.now(); + const idleSessionIds: string[] = []; + for (const [sessionId, lastSeen] of streamableLastSeen) { + if (now - lastSeen > STREAMABLE_SESSION_IDLE_MS) idleSessionIds.push(sessionId); + } + for (const sessionId of idleSessionIds) { + void closeStreamableSession(sessionId); + } +}, STREAMABLE_SESSION_SWEEP_MS); +streamableReapInterval.unref(); const httpServer = createServer(async (req, res) => { const url = new URL(req.url!, `http://localhost:${PORT}`); @@ -123,6 +196,7 @@ const httpServer = createServer(async (req, res) => { res.end('This session belongs to a different identity.'); return; } + streamableLastSeen.set(sessionId, Date.now()); await transport.handleRequest(req, res); return; } @@ -131,6 +205,25 @@ const httpServer = createServer(async (req, res) => { res.end('Missing mcp-session-id header'); return; } + // Per-identity session cap: evict the identity's oldest session before + // creating one more, so a caller that keeps calling initialize without + // ever sending DELETE cannot grow this identity's session count past + // MAX_SESSIONS_PER_IDENTITY. + const identitySessionIds = [...streamableIdentities.entries()] + .filter(([, identity]) => identity === auth.identity) + .map(([sid]) => sid); + if (identitySessionIds.length > 0 && identitySessionIds.length >= MAX_SESSIONS_PER_IDENTITY) { + let oldestSessionId = identitySessionIds[0]; + let oldestLastSeen = streamableLastSeen.get(oldestSessionId) ?? 0; + for (const sid of identitySessionIds) { + const lastSeen = streamableLastSeen.get(sid) ?? 0; + if (lastSeen < oldestLastSeen) { + oldestLastSeen = lastSeen; + oldestSessionId = sid; + } + } + await closeStreamableSession(oldestSessionId); + } // No session header on a POST: treat as an initialization request. The SDK // enforces the rest (a non-initialize body without a session is rejected by // the transport itself; the unstored transport is then garbage). @@ -139,17 +232,32 @@ const httpServer = createServer(async (req, res) => { onsessioninitialized: (sid) => { streamableTransports.set(sid, transport); streamableIdentities.set(sid, auth.identity); + streamableLastSeen.set(sid, Date.now()); }, onsessionclosed: (sid) => { streamableTransports.delete(sid); streamableIdentities.delete(sid); + streamableLastSeen.delete(sid); }, }); transport.onclose = () => { + // Fires on every close, not just DELETE. handleDeleteRequest (the + // SDK's own webStandardStreamableHttp.js) calls onsessionclosed above + // AND this callback in the same request, so on that path the two are + // redundant with each other. But onsessionclosed is wired into the SDK + // exclusively from that DELETE handler - it is never invoked for an + // abrupt disconnect, and never invoked by a direct transport.close() + // call. The idle reaper and the per-identity cap eviction above both + // call transport.close() directly (via closeStreamableSession), which + // drives only this callback. This is therefore the one drain path + // every teardown route - DELETE, idle reaper, and cap eviction alike - + // actually shares; onsessionclosed only ever adds coverage for DELETE, + // which this callback already has. const sid = transport.sessionId; if (sid) { streamableTransports.delete(sid); streamableIdentities.delete(sid); + streamableLastSeen.delete(sid); } }; const session = createMcpServer(auth.identity); @@ -169,3 +277,7 @@ httpServer.on('error', (err: Error) => { console.error('Fatal error in HTTP server:', err.message); process.exit(1); }); + +httpServer.on('close', () => { + clearInterval(streamableReapInterval); +}); diff --git a/packages/builder/test/integration/streamable-http-gating.test.ts b/packages/builder/test/integration/streamable-http-gating.test.ts index 14369c9..c977fd9 100644 --- a/packages/builder/test/integration/streamable-http-gating.test.ts +++ b/packages/builder/test/integration/streamable-http-gating.test.ts @@ -12,9 +12,21 @@ import type { RawServerHandle } from '../setup.ts'; // auth-gating.test.ts covers. Same two things at stake, same reason no unit // test can see them: the exact status-code/header wiring against a real // spawned server (401 checked before session lookup, 404 on an unknown -// session, 403 on cross-identity), and whether streamableTransports / -// streamableIdentities actually drain on both cleanup paths -// (onsessionclosed - the DELETE path - and transport.onclose). +// session, 403 on cross-identity - on both POST and GET), and whether +// streamableTransports/streamableIdentities/streamableLastSeen actually +// drain on every teardown path: DELETE (onsessionclosed, then +// transport.onclose - redundant with each other on that one path), the idle +// reaper, and the per-identity session cap's eviction (the latter two only +// ever go through transport.onclose - see index.ts's own comment on that +// callback for why onsessionclosed does not cover them). The reaper and the +// cap exist because the SSE pair self-limits on an abrupt disconnect +// (res.on('close') drains its maps with no help from us) and /mcp does not: +// nothing tells the server a caller dropped its socket without DELETE, so an +// un-reaped session - transport, McpServer, and its 24 tools - would +// otherwise sit in memory until process exit. Both bounds are exercised +// below with env-shrunk thresholds (VECTOR_MCP_SESSION_IDLE_MS / +// VECTOR_MCP_SESSION_SWEEP_MS / VECTOR_MCP_MAX_SESSIONS_PER_IDENTITY), never +// the real multi-minute defaults. // // Mixed transport strategy, deliberately: // - Case 1 uses the real SDK Client + StreamableHTTPClientTransport, to prove @@ -213,6 +225,23 @@ describe('streamable HTTP gating - MCP_AUTH_TOKENS set', () => { assert.ok(!r.body.includes(TOKEN_ALICE) && !r.body.includes(TOKEN_BOB), 'no token value echoed in the 401 body'); }); + test('a real, valid session id with no Authorization header is still rejected with 401, not 403 or 404 (auth checked before session lookup)', async () => { + // Ordering pin, same class as auth-gating.test.ts's "POST /messages with + // no Authorization header is rejected with 401, not 400 or 404": uses a + // session id that genuinely exists (not a bogus one), so a passing result + // cannot be explained away as "it 404'd because the session doesn't + // exist" - it specifically proves the auth check runs, and short-circuits, + // before the session map is ever consulted. + const init = record(await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest())); + assert.equal(init.status, 200); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId); + + const r = record(await postMcp(ctx.port, { 'mcp-session-id': sessionId! }, toolsListRequest())); + assert.equal(r.status, 401); + assert.equal(r.headers['www-authenticate'], 'Bearer'); + }); + test('cross-identity: a valid token for a different identity than the session owner is rejected with 403; the owner is accepted', async () => { const init = record(await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest())); assert.equal(init.status, 200); @@ -257,6 +286,27 @@ describe('streamable HTTP gating - MCP_AUTH_TOKENS set', () => { noSession.destroy(); }); + test('GET with a valid session but a different identity token is rejected with 403, independent of the POST cross-identity path', async () => { + // The cross-identity test above only ever exercises POST. The route's + // identity check runs before transport.handleRequest() regardless of + // method, so this pins that it is not accidentally POST-only wiring. + const init = record(await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest())); + assert.equal(init.status, 200); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId); + + // GET's error responses (unlike its 200 open-stream response) are short, + // res.end()-terminated bodies - safe to read to completion with + // rawRequest, same as the POST cross-identity case. + const crossed = record(await rawRequest(ctx.port, 'GET', '/mcp', { + Authorization: `Bearer ${TOKEN_BOB}`, + 'mcp-session-id': sessionId!, + Accept: 'text/event-stream', + })); + assert.equal(crossed.status, 403); + assert.equal(crossed.body, 'This session belongs to a different identity.'); + }); + test('DELETE terminates the session; a follow-up request with that session id 404s (the maps drain via onsessionclosed)', async () => { const init = record(await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest())); assert.equal(init.status, 200); @@ -277,6 +327,12 @@ describe('streamable HTTP gating - MCP_AUTH_TOKENS set', () => { assert.equal(followUp.status, 404); }); + test('DELETE with no session id is rejected with 400', async () => { + const r = record(await rawRequest(ctx.port, 'DELETE', '/mcp', { Authorization: `Bearer ${TOKEN_ALICE}` })); + assert.equal(r.status, 400); + assert.equal(r.body, 'Missing mcp-session-id header'); + }); + test('no token value leaked into any captured response body, response header, or stderr', () => { const stderr = ctx.stderr(); for (const body of seenBodies) { @@ -378,3 +434,94 @@ describe('streamable HTTP gating - per-IP separation for anonymous callers', () assert.equal(crossedIdentity.status, 403, 'a session opened under one anonymous identity must reject a request whose recomputed identity differs'); }); }); + +describe('streamable HTTP gating - idle session reaper', () => { + let ctx: RawServerHandle; + + before(async () => { + // Idle bound and sweep cadence both shrunk far below their real defaults + // (10 minutes / 60 seconds) so this test proves the reaper in under a + // second instead of sleeping anywhere near the production window. The + // worst case for when a sweep can first observe this session as idle is + // roughly idleMs + sweepMs after creation (a sweep tick can land just + // before the session is created); the wait below is 2x that worst case. + ctx = await spawnServer({ ...AUTH_ENV, VECTOR_MCP_SESSION_IDLE_MS: '400', VECTOR_MCP_SESSION_SWEEP_MS: '100' }); + }); + + after(async () => { + await stopRawServer(ctx); + }); + + test('a session idle past the threshold is reaped on the next sweep; the maps drain the same way DELETE does', async () => { + const init = await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest()); + assert.equal(init.status, 200); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId); + + // No request of any kind against this session in between - lastSeen never + // bumps, so it goes idle starting from the moment initialize returned. + await new Promise((resolveWait) => setTimeout(resolveWait, 1000)); + + const followUp = await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': sessionId! }, + toolsListRequest(), + ); + assert.equal(followUp.status, 404, 'an idle-reaped session must 404 - streamableTransports/streamableIdentities drained, same observable proof as the DELETE case'); + }); +}); + +describe('streamable HTTP gating - per-identity session cap', () => { + let ctx: RawServerHandle; + + before(async () => { + // Cap shrunk to 2 so the third session for the same identity is enough to + // force an eviction, well within a hermetic test's budget. + ctx = await spawnServer({ ...AUTH_ENV, VECTOR_MCP_MAX_SESSIONS_PER_IDENTITY: '2' }); + }); + + after(async () => { + await stopRawServer(ctx); + }); + + test('opening a session beyond the per-identity cap evicts the oldest; the newest sessions survive', async () => { + const first = await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest()); + assert.equal(first.status, 200); + const firstId = first.headers['mcp-session-id'] as string | undefined; + assert.ok(firstId); + + const second = await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest()); + assert.equal(second.status, 200); + const secondId = second.headers['mcp-session-id'] as string | undefined; + assert.ok(secondId); + assert.notEqual(secondId, firstId); + + // alice is now at the cap (2 live sessions). A third session for the SAME + // identity must evict the oldest (first) before it is created. + const third = await postMcp(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}` }, initializeRequest()); + assert.equal(third.status, 200); + const thirdId = third.headers['mcp-session-id'] as string | undefined; + assert.ok(thirdId); + + const firstFollowUp = await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': firstId! }, + toolsListRequest(), + ); + assert.equal(firstFollowUp.status, 404, 'the oldest session must have been evicted once the identity hit its session cap'); + + const secondFollowUp = await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': secondId! }, + toolsListRequest(), + ); + assert.equal(secondFollowUp.status, 200, 'the second-oldest session must still be alive'); + + const thirdFollowUp = await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': thirdId! }, + toolsListRequest(), + ); + assert.equal(thirdFollowUp.status, 200, 'the newest session must still be alive'); + }); +});