From 035c16a085ff2365cbd57cd9c407e3401e2758b6 Mon Sep 17 00:00:00 2001 From: FilipBl4gojevic <84703395+FilipBl4gojevic@users.noreply.github.com> Date: Fri, 31 Jul 2026 14:27:29 +0200 Subject: [PATCH] feat(builder): Streamable HTTP endpoint at /mcp - dual-stack with the deprecated SSE pair, same auth and session invariants --- package.json | 2 +- packages/builder/src/index.ts | 59 +++ .../streamable-http-gating.test.ts | 380 ++++++++++++++++++ 3 files changed, 440 insertions(+), 1 deletion(-) create mode 100644 packages/builder/test/integration/streamable-http-gating.test.ts diff --git a/package.json b/package.json index 5721d19..24a901b 100644 --- a/package.json +++ b/package.json @@ -15,7 +15,7 @@ "test:smoke": "npm run build && node --import tsx --test packages/builder/test/smoke/*.test.ts", "test:smoke:signer": "npm run build && node --import tsx --test packages/signer/test/smoke/*.test.ts", "test:integration": "npm run build && node --import tsx --test packages/builder/test/integration/*.test.ts", - "test:integration:auth": "npm run build && node --import tsx --test packages/builder/test/integration/auth-gating.test.ts packages/builder/test/integration/rate-limit-gating.test.ts", + "test:integration:auth": "npm run build && node --import tsx --test packages/builder/test/integration/auth-gating.test.ts packages/builder/test/integration/rate-limit-gating.test.ts packages/builder/test/integration/streamable-http-gating.test.ts", "test:integration:signer": "node --import tsx --test packages/signer/test/integration/*.test.ts" }, "devDependencies": { diff --git a/packages/builder/src/index.ts b/packages/builder/src/index.ts index 49bad5c..8b919b5 100644 --- a/packages/builder/src/index.ts +++ b/packages/builder/src/index.ts @@ -3,6 +3,7 @@ import { fileURLToPath } from 'url'; import { dirname, resolve } from 'path'; import { existsSync } from 'fs'; import { createServer } from 'http'; +import { randomUUID } from 'node:crypto'; // Get directory name for ES module const __filename = fileURLToPath(import.meta.url); @@ -25,6 +26,7 @@ if (existsSync(envPath)) { import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; +import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { registerVectorTools } from "./vector/index.js"; import { loadAuthConfig, resolveIdentity, clientIpOf } from "./auth.js"; @@ -50,6 +52,8 @@ if (authConfig.enabled) { const PORT = parseInt(process.env.PORT || '3000'); const transports = new Map(); const sessionIdentities = new Map(); +const streamableTransports = new Map(); +const streamableIdentities = new Map(); const httpServer = createServer(async (req, res) => { const url = new URL(req.url!, `http://localhost:${PORT}`); @@ -96,6 +100,61 @@ const httpServer = createServer(async (req, res) => { return; } await transport.handlePostMessage(req, res); + } else if (url.pathname === '/mcp') { + // Modern Streamable HTTP transport (MCP spec: SSE is deprecated). Same auth + // gate and the same invariant as the SSE pair: a valid token for identity A + // must not be able to drive identity B's session. + const auth = resolveIdentity(req.headers.authorization, authConfig, clientIpOf(req)); + if (!auth.ok) { + res.writeHead(401, { 'Content-Type': 'text/plain', 'WWW-Authenticate': 'Bearer' }); + res.end(auth.reason); + return; + } + const sessionId = req.headers['mcp-session-id'] as string | undefined; + if (sessionId) { + const transport = streamableTransports.get(sessionId); + if (!transport) { + res.writeHead(404, { 'Content-Type': 'text/plain' }); + res.end('Session not found'); + return; + } + if (streamableIdentities.get(sessionId) !== auth.identity) { + res.writeHead(403, { 'Content-Type': 'text/plain' }); + res.end('This session belongs to a different identity.'); + return; + } + await transport.handleRequest(req, res); + return; + } + if (req.method !== 'POST') { + res.writeHead(400, { 'Content-Type': 'text/plain' }); + res.end('Missing mcp-session-id header'); + return; + } + // 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). + const transport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: (sid) => { + streamableTransports.set(sid, transport); + streamableIdentities.set(sid, auth.identity); + }, + onsessionclosed: (sid) => { + streamableTransports.delete(sid); + streamableIdentities.delete(sid); + }, + }); + transport.onclose = () => { + const sid = transport.sessionId; + if (sid) { + streamableTransports.delete(sid); + streamableIdentities.delete(sid); + } + }; + const session = createMcpServer(auth.identity); + await session.connect(transport); + await transport.handleRequest(req, res); } else { res.writeHead(404); res.end('Not found'); diff --git a/packages/builder/test/integration/streamable-http-gating.test.ts b/packages/builder/test/integration/streamable-http-gating.test.ts new file mode 100644 index 0000000..14369c9 --- /dev/null +++ b/packages/builder/test/integration/streamable-http-gating.test.ts @@ -0,0 +1,380 @@ +import { describe, test, before, after } from 'node:test'; +import assert from 'node:assert/strict'; +import { request as httpRequest } from 'node:http'; +import type { IncomingHttpHeaders, IncomingMessage } from 'node:http'; +import { Client } from '@modelcontextprotocol/sdk/client/index.js'; +import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; +import { spawnServer, stopRawServer } from '../setup.ts'; +import type { RawServerHandle } from '../setup.ts'; + +// Covers the /mcp Streamable HTTP route in packages/builder/src/index.ts - the +// modern, dual-stack sibling of the deprecated /sse + /messages pair that +// 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). +// +// Mixed transport strategy, deliberately: +// - Case 1 uses the real SDK Client + StreamableHTTPClientTransport, to prove +// a real client can drive this route end to end (not just that our own raw +// assertions are self-consistent). +// - Every other case uses raw node:http, for the same reason auth-gating.test.ts +// does: this file's point is exact status codes and exact headers, plus - +// for the cross-identity case - presenting a *different* Authorization +// header than the one that opened the session, and reusing one session's id +// under another identity's headers. StreamableHTTPClientTransport fixes its +// Authorization header per transport instance and manages mcp-session-id +// internally, so it cannot express either scenario. +// +// One wrinkle the SSE pair never had: the route does not set +// enableJsonResponse, so a POST that carries a JSON-RPC *request* (as opposed +// to a bare notification) gets back a 200 with Content-Type: text/event-stream +// and the JSON-RPC response embedded in a single "data: ..." line, not a +// plain application/json body. extractJsonRpc() below unwraps that. Responses +// this route writes itself - 401/403/404/400 - are always short plain-text +// bodies (res.end(literal) before the transport is ever touched), never SSE. + +const TOKEN_ALICE = 'tok-alice-shttp-3f8b21ec'; +const TOKEN_BOB = 'tok-bob-shttp-9a06d4c7'; +const AUTH_ENV = { MCP_AUTH_TOKENS: `alice:${TOKEN_ALICE},bob:${TOKEN_BOB}` }; + +const RATE_LIMIT_PATTERN = /^Rate limit exceeded\. Retry after \d+ms\.$/; + +interface RawResponse { + status: number; + headers: IncomingHttpHeaders; + body: string; +} + +function rawRequest(port: number, method: string, path: string, headers: Record, body?: string): Promise { + return new Promise((resolve, reject) => { + const payload = body !== undefined ? Buffer.from(body, 'utf8') : undefined; + const req = httpRequest({ + host: '127.0.0.1', + port, + path, + method, + headers: payload ? { ...headers, 'Content-Length': String(payload.length) } : headers, + }, (res: IncomingMessage) => { + let buf = ''; + res.setEncoding('utf8'); + res.on('data', (chunk: string) => { buf += chunk; }); + res.on('end', () => resolve({ status: res.statusCode ?? 0, headers: res.headers, body: buf })); + res.on('error', reject); + }); + req.on('error', reject); + if (payload) req.end(payload); else req.end(); + }); +} + +// Every POST this file sends carries a real JSON-RPC body, so the required +// Accept/Content-Type pair (the transport 406s/415s a POST missing either +// half - see webStandardStreamableHttp.js's handlePostRequest) is applied +// once here rather than at every call site. +function postMcp(port: number, headers: Record, payload: unknown): Promise { + return rawRequest(port, 'POST', '/mcp', { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + ...headers, + }, JSON.stringify(payload)); +} + +interface StreamResponse extends RawResponse { + /** Ends the underlying request/response once the caller is done with the open stream. */ + destroy: () => void; +} + +/** + * Raw GET /mcp. Resolves as soon as the response headers arrive - for a 200 + * that leaves an open SSE stream behind (the caller destroys it when done); + * for a 4xx the body is a short literal string index.ts writes directly, but + * this file never needs to inspect it, only the status/headers. + */ +function getMcpStream(port: number, headers: Record): Promise { + return new Promise((resolve, reject) => { + const req = httpRequest({ host: '127.0.0.1', port, path: '/mcp', method: 'GET', headers }, (res: IncomingMessage) => { + res.resume(); // drain so the socket doesn't back up; body content is never asserted on here + resolve({ status: res.statusCode ?? 0, headers: res.headers, body: '', destroy: () => req.destroy() }); + }); + req.on('error', reject); + req.end(); + }); +} + +/** + * Unwraps the single JSON-RPC response this file's POSTs always carry. + * enableJsonResponse is not set on the transport (matching the brief's + * route code verbatim), so a POST carrying a request - as opposed to a bare + * notification, which gets 202 with no body - answers with a + * text/event-stream body containing exactly one "data: " line, not a + * plain application/json body. + */ +function extractJsonRpc(headers: IncomingHttpHeaders, body: string): any { + const contentType = headers['content-type'] ?? ''; + if (contentType.includes('text/event-stream')) { + const m = body.match(/^data: (.*)$/m); + assert.ok(m, `expected an SSE "data:" line in a text/event-stream body, got: ${body}`); + return JSON.parse(m![1]); + } + if (contentType.includes('application/json') && body.length > 0) { + return JSON.parse(body); + } + return undefined; +} + +let rpcId = 1; +function nextId(): number { + return rpcId++; +} + +function initializeRequest() { + return { + jsonrpc: '2.0', + id: nextId(), + method: 'initialize', + params: { + protocolVersion: '2025-06-18', + capabilities: {}, + clientInfo: { name: 'streamable-http-gating-test', version: '1.0.0' }, + }, + }; +} + +function toolsListRequest() { + return { jsonrpc: '2.0', id: nextId(), method: 'tools/list' }; +} + +// Same fast-fail probe rate-limit-gating.test.ts established: lucidForAddress +// (packages/builder/src/vector/build.ts) validates changeAddress via +// getAddressDetails() and throws before any provider call, so this is +// in-process only - no network, no real timing dependency - whether or not +// the rate limiter admits the call. +const PROBE_TOOL_ARGS = { changeAddress: 'addr1garbage', recipientAddress: 'addr1garbage', amount: 1 }; +function toolsCallRequest() { + return { + jsonrpc: '2.0', + id: nextId(), + method: 'tools/call', + params: { name: 'vector_build_send_apex', arguments: PROBE_TOOL_ARGS }, + }; +} + +async function connectStreamable(port: number, token?: string): Promise { + const transport = new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${port}/mcp`), + token ? { requestInit: { headers: { Authorization: `Bearer ${token}` } } } : undefined, + ); + const client = new Client({ name: 'streamable-http-gating-test', version: '1.0.0' }); + await client.connect(transport); + return client; +} + +// Every response the auth-enabled block observes, scanned once at the end for +// a leaked token - same style as auth-gating.test.ts's record()/seenBodies. +const seenBodies: string[] = []; +const seenHeaderDumps: string[] = []; +function record(r: T): T { + seenBodies.push(r.body); + seenHeaderDumps.push(JSON.stringify(r.headers)); + return r; +} + +describe('streamable HTTP gating - MCP_AUTH_TOKENS set', () => { + let ctx: RawServerHandle; + + before(async () => { + ctx = await spawnServer(AUTH_ENV); + }); + + after(async () => { + await stopRawServer(ctx); + }); + + test('SDK client with a valid token connects over Streamable HTTP and lists exactly 24 tools', async () => { + const client = await connectStreamable(ctx.port, TOKEN_ALICE); + try { + const result = await client.listTools(); + assert.equal(result.tools.length, 24); + } finally { + await client.close(); + } + }); + + test('no token: the SDK client rejects on connect, and a raw POST initialize is rejected with 401 and no token echo', async () => { + const transport = new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${ctx.port}/mcp`)); + const client = new Client({ name: 'streamable-http-gating-test', version: '1.0.0' }); + await assert.rejects(() => client.connect(transport)); + + const r = record(await postMcp(ctx.port, {}, initializeRequest())); + assert.equal(r.status, 401); + assert.equal(r.headers['www-authenticate'], 'Bearer'); + assert.ok(!r.body.includes(TOKEN_ALICE) && !r.body.includes(TOKEN_BOB), 'no token value echoed in the 401 body'); + }); + + 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); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId, 'expected an mcp-session-id response header on the initialize response'); + + const crossed = record(await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_BOB}`, 'mcp-session-id': sessionId! }, + toolsListRequest(), + )); + assert.equal(crossed.status, 403); + assert.equal(crossed.body, 'This session belongs to a different identity.'); + + const own = record(await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': sessionId! }, + toolsListRequest(), + )); + assert.equal(own.status, 200); + const payload = extractJsonRpc(own.headers, own.body); + assert.equal(payload.result.tools.length, 24); + }); + + test('GET opens the standalone SSE stream for the session owner; GET with no session id is rejected with 400', async () => { + 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 opened = record(await getMcpStream(ctx.port, { + Authorization: `Bearer ${TOKEN_ALICE}`, + 'mcp-session-id': sessionId!, + Accept: 'text/event-stream', + })); + assert.equal(opened.status, 200); + assert.ok(opened.headers['content-type']?.includes('text/event-stream')); + opened.destroy(); + + const noSession = record(await getMcpStream(ctx.port, { Authorization: `Bearer ${TOKEN_ALICE}`, Accept: 'text/event-stream' })); + assert.equal(noSession.status, 400); + noSession.destroy(); + }); + + 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); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId); + + const del = record(await rawRequest(ctx.port, 'DELETE', '/mcp', { + Authorization: `Bearer ${TOKEN_ALICE}`, + 'mcp-session-id': sessionId!, + })); + assert.equal(del.status, 200); + + const followUp = record(await postMcp( + ctx.port, + { Authorization: `Bearer ${TOKEN_ALICE}`, 'mcp-session-id': sessionId! }, + toolsListRequest(), + )); + assert.equal(followUp.status, 404); + }); + + test('no token value leaked into any captured response body, response header, or stderr', () => { + const stderr = ctx.stderr(); + for (const body of seenBodies) { + assert.ok(!body.includes(TOKEN_ALICE), 'a response body must never contain the alice token'); + assert.ok(!body.includes(TOKEN_BOB), 'a response body must never contain the bob token'); + } + for (const dump of seenHeaderDumps) { + assert.ok(!dump.includes(TOKEN_ALICE), 'a response header must never contain the alice token'); + assert.ok(!dump.includes(TOKEN_BOB), 'a response header must never contain the bob token'); + } + assert.ok(!stderr.includes(TOKEN_ALICE), 'server stderr must never contain the alice token'); + assert.ok(!stderr.includes(TOKEN_BOB), 'server stderr must never contain the bob token'); + }); +}); + +describe('streamable HTTP gating - per-IP separation for anonymous callers', () => { + let ctx: RawServerHandle; + + before(async () => { + // No MCP_AUTH_TOKENS: every caller is admitted, bucketed by clientIpOf(req) + // (rightmost X-Forwarded-For entry). Tiny budget so three probes exhaust it + // well within this test's lifetime. + ctx = await spawnServer({ MCP_AUTH_TOKENS: '', VECTOR_RATE_LIMIT_PER_MINUTE: '3' }); + }); + + after(async () => { + await stopRawServer(ctx); + }); + + test('the forgeable leftmost X-Forwarded-For entry does not matter; only the rightmost, proxy-appended entry buckets the rate limit', async () => { + // Identical leftmost entry ("6.6.6.6") in both groups - the assertion + // below on this fact is the point of the test: if the leftmost entry + // mattered, both groups would land in the same bucket. Only the rightmost + // entry differs, and only the rightmost entry may legitimately be trusted + // (see clientIpOf's docstring in auth.ts) - the trusted proxy appends it, + // everything left of that append point is client-supplied and forgeable. + const XFF_BUDGET_IP = '6.6.6.6, 203.0.113.50'; + const XFF_OTHER_IP = '6.6.6.6, 198.51.100.7'; + assert.equal( + XFF_BUDGET_IP.split(',')[0].trim(), + XFF_OTHER_IP.split(',')[0].trim(), + 'sanity: the leftmost (forgeable) XFF entry must be identical across both probe groups', + ); + + // A "probe" is a fresh session (own initialize) plus one tool call under + // the given X-Forwarded-For - a session's identity is fixed at + // initialize time, so the follow-up call must repeat the same header or + // it would 403 as a cross-identity request instead of ever reaching the + // rate limiter (see the bonus assertion at the end of this test). + async function probe(xff: string): Promise<{ sessionId: string; text: string }> { + const init = await postMcp(ctx.port, { 'X-Forwarded-For': xff }, initializeRequest()); + assert.equal(init.status, 200, `probe initialize under X-Forwarded-For "${xff}" must succeed`); + const sessionId = init.headers['mcp-session-id'] as string | undefined; + assert.ok(sessionId, 'expected an mcp-session-id response header on the initialize response'); + + const call = await postMcp(ctx.port, { 'X-Forwarded-For': xff, 'mcp-session-id': sessionId! }, toolsCallRequest()); + assert.equal(call.status, 200, `probe tool call under X-Forwarded-For "${xff}" must reach the tool handler`); + const payload = extractJsonRpc(call.headers, call.body); + const content = payload.result.content as Array<{ type: string; text?: string }>; + return { sessionId: sessionId!, text: content?.[0]?.text ?? '' }; + } + + // Three probes under 203.0.113.50 consume the whole budget (VECTOR_RATE_LIMIT_PER_MINUTE=3). + // Each probe opens its OWN session - proven separately below - but all three + // share one rate-limit bucket because limiterFor() keys on identity + // (anon:203.0.113.50), not on session id. + for (let i = 1; i <= 3; i++) { + const { text } = await probe(XFF_BUDGET_IP); + assert.ok(!RATE_LIMIT_PATTERN.test(text), `probe ${i} under 203.0.113.50 must not be rate-limited yet, got: ${text}`); + } + + // The 4th probe's tool call is refused - the bucket is exhausted. + const fourth = await probe(XFF_BUDGET_IP); + assert.match(fourth.text, RATE_LIMIT_PATTERN, 'the 4th probe under 203.0.113.50 must be rate-limited'); + + // A distinct rightmost IP gets its own, unexhausted budget - proving the + // buckets are separate despite the identical leftmost entry above. + const distinct = await probe(XFF_OTHER_IP); + assert.ok(!RATE_LIMIT_PATTERN.test(distinct.text), `a distinct rightmost IP must have its own budget, got: ${distinct.text}`); + + // Bonus invariant proof (per the task brief): the 198.51.100.7 probe's own + // session id, presented with its own consistent XFF, is still accepted. + // Presented instead with the budget IP's XFF, it recomputes to a + // different identity (anon:203.0.113.50) and hits the same + // cross-identity guard the token-based case exercises - it does not + // silently fall through to the exhausted bucket, or to any bucket at all. + const ownIdentity = await postMcp( + ctx.port, + { 'X-Forwarded-For': XFF_OTHER_IP, 'mcp-session-id': distinct.sessionId }, + toolsListRequest(), + ); + assert.equal(ownIdentity.status, 200, "the OTHER-IP session's own identity, presented consistently, is still accepted"); + + const crossedIdentity = await postMcp( + ctx.port, + { 'X-Forwarded-For': XFF_BUDGET_IP, 'mcp-session-id': distinct.sessionId }, + toolsListRequest(), + ); + assert.equal(crossedIdentity.status, 403, 'a session opened under one anonymous identity must reject a request whose recomputed identity differs'); + }); +});