diff --git a/docs/cli.md b/docs/cli.md index 1e93ef191d..44b577af8c 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -322,8 +322,8 @@ rea mcp --http --port 8765 `--host` and `--port` default to `MCP_HOST` and `MCP_PORT` from the environment, then to `127.0.0.1` and an ephemeral port. The bound URL is written to stderr as -one JSON log line (`"msg":"MCP transport listening"`); nothing is written to -stdout. `rea mcp --http --help` lists the accepted options. The listener serves +one JSON log line (`"msg":"MCP transport listening"`) whatever `REA_LOG_LEVEL` +is set to; nothing is written to stdout. `rea mcp --http --help` lists the accepted options. The listener serves one client at a time: a session ends with the client's `DELETE`, and a client that connects while another session is open takes it over, so a client that exited without ending its session needs no restart. The listener runs until diff --git a/docs/mcp-contracts.md b/docs/mcp-contracts.md index 47cc5895b6..d9b2508ffb 100644 --- a/docs/mcp-contracts.md +++ b/docs/mcp-contracts.md @@ -427,7 +427,8 @@ stdio is the default transport and the only one that client registrations, streamable HTTP transport at `/mcp`; it replaces stdio for that process rather than running beside it. Host and port come from the flags, then `MCP_HOST` and `MCP_PORT`, then `127.0.0.1` and an ephemeral port. The bound URL is announced -as one JSON log line on stderr (`"msg":"MCP transport listening"` with `url`); +as one JSON log line on stderr (`"msg":"MCP transport listening"` with `url`) +at every `REA_LOG_LEVEL`, because an ephemeral port is otherwise unreachable; stdout stays unused. The HTTP server is stateful and serves one client at a time on one server @@ -439,11 +440,19 @@ and the shared active target, and the next client opens a fresh session against them. The newest client replaces an open session: an `initialize` that arrives while a session is open ends that session instead of being refused, because a client that exits without `DELETE` must not wedge the listener; the replaced -client's open streams are closed and its later requests are refused. Requests -that carry no valid session id are refused with `400` or `404` as the SDK -transport defines. On a loopback bind, requests whose `Host` header is outside -the SDK loopback allowlist are refused with `403`; a non-loopback bind relaxes -that check and logs the relaxation. Request bodies are bounded by the SDK's +client's open streams are closed and its later requests are refused. Only an +`initialize` that carries the streamable headers (`Content-Type: +application/json` and an `Accept` naming both `application/json` and +`text/event-stream`) and passes the Host and Origin checks can replace a +session; any other sessionless POST goes to the live transport, which refuses +it without disturbing the session. Requests that carry no valid session id are +refused with `400` or `404` as the SDK transport defines. A bind is loopback by +address class, not spelling: any `127.0.0.0/8` address, `::1`, an IPv4-mapped +loopback and `localhost` names. On such a bind, requests whose `Host` header is +outside the SDK loopback allowlist plus the bind's own hostname are refused with +`403`, and a browser `Origin` outside that allowlist is refused the same way (a +missing `Origin` passes); a non-loopback bind relaxes both checks and logs the +relaxation. Request bodies are bounded by the SDK's 4 MiB limit, and the response budget keeps its stdio-derived floor. The listener ends on SIGINT or SIGTERM only, not on stdin EOF, and exits `130` or `143` after cleanup settles. diff --git a/docs/specs/agent-environments/http-transport/spec.md b/docs/specs/agent-environments/http-transport/spec.md index 95d06b67fb..1d3205cfa7 100644 --- a/docs/specs/agent-environments/http-transport/spec.md +++ b/docs/specs/agent-environments/http-transport/spec.md @@ -16,7 +16,7 @@ client registration keep stdio. dispatcher without changing bare `mcp`, `mcp doctor` or `mcp add`. - Resolve host and port from the flags, then `MCP_HOST`/`MCP_PORT` from the environment snapshot, then `127.0.0.1` and port `0`; announce the bound URL as - one pino JSON line on stderr and leave stdout unused. + one pino JSON line on stderr at every log level and leave stdout unused. - Use `WebStandardStreamableHTTPServerTransport` in stateful mode around one server instance built once. `DELETE` ends the session on its transport without running the instance's shutdown hooks, so the same instance keeps its @@ -29,8 +29,13 @@ client registration keep stdio. - Bridge node:http to web Request/Response in `src/server/http/`, streaming response chunks as they are produced and turning a client disconnect into a request abort and body cancellation. -- Validate the `Host` header with the SDK loopback allowlist for loopback binds; - relax it for other binds and log that relaxation. +- Classify a loopback bind by address class (`127.0.0.0/8`, `::1`, IPv4-mapped + loopback, `localhost` names), validate the `Host` and `Origin` headers with + the SDK loopback allowlist plus the bind's own hostname on such binds, and + relax both for other binds and log that relaxation. +- Let only an `initialize` that carries the streamable headers and passes those + checks replace an open session; refuse requests once the slot is closed, join + an arming session during shutdown, and keep close stages retryable. - End the HTTP listener on SIGINT/SIGTERM only, never on stdin EOF, and exit with the signal status (`130`/`143`) after cleanup settles. - Export `serveStreamableHttp({ host, port, createMcpServer, routes, logger })` diff --git a/docs/specs/agent-environments/http-transport/verification.md b/docs/specs/agent-environments/http-transport/verification.md index e261f2001b..322fd844fc 100644 --- a/docs/specs/agent-environments/http-transport/verification.md +++ b/docs/specs/agent-environments/http-transport/verification.md @@ -155,3 +155,44 @@ re-runs, not by repeating the full projects on an idle host. The error-level log line that names the bind cause was observed in earlier runs, not asserted by `mainErrors.test.ts`. Response budgets keep the stdio-derived floor and were not re-measured over HTTP. Windows was not run. + +## Automated review findings on the PR (2026-10-18) + +Codex and Greptile left eight inline findings on morluto/rea#4955 at +46dd8f0e; each was checked against the code and addressed in one follow-up +commit, with a regression test where the behaviour is observable: + +- A late `initialize` could arm a fresh transport after shutdown (Codex, P1): + the slot now records that it is closed, `close()` joins an arming session + before ending it, and every request after that is refused with `503`. Test: + an initialize racing `close()` gets `503`, the shutdown hook fires once, and + the next request gets `503`. +- A failed `close()` was memoized forever (both reviewers, P1): the close runs + three stages once each and a failed stage is retried by the next call. Test: + the instance's first `close()` rejects, the second call resolves and the port + is released. +- Loopback aliases skipped Host validation (both reviewers, P2 and P1): a bind + is now loopback by address class (`127.0.0.0/8`, `::1`, IPv4-mapped loopback, + `localhost` names) after the same normalization the SDK applies to the Host + header, and the allowlist adds the bind's own hostname so requests to + `127.0.0.2` or `localhost.` are still served. Test: twelve spellings. +- A sessionless POST could close the live session before any header check + (Greptile, P1): on loopback binds the route validates `Origin` with the same + allowlist as `Host`, and only a POST carrying `Content-Type: + application/json` and an `Accept` naming both `application/json` and + `text/event-stream` may replace a session; others reach the live transport, + which refuses them. Test: a `text/plain` initialize and one with a foreign + `Origin` are refused (4xx and 403) while the connected client keeps echoing. +- The bound URL disappeared at quiet log levels (Greptile, P2): the listening + line is written through a child logger pinned to `info`. Test: a `warn` + logger still receives the line with the URL. +- `rea mcp -h --http` reached the ordinary CLI (Greptile, P2): the dispatcher + predicate accepts any flag after the entry. Test: the dispatcher case. + +Checks after the change: `oxfmt --check .` clean; `tsc --noEmit` clean; +`oxlint` on the touched sources clean after extracting the staged close into +its own function; `npx knip` no findings; `npm run jscpd` 3.35 % duplicated +lines; `scan:todos` none; `verify:test-discovery` 1,583 files once; `vitest +run` over the listener and bridge adapters tests 14 passed in 2 files; after +`npm run build:cached`, `vitest run` over the dispatcher, option, streamable +HTTP, shutdown and main-error boundary tests 22 passed in 5 files. diff --git a/scripts/rea.mjs b/scripts/rea.mjs index d0f5bc42a9..326676a976 100755 --- a/scripts/rea.mjs +++ b/scripts/rea.mjs @@ -15,10 +15,11 @@ process.env.REA_PACKAGE_VERSION = packageJson.version; const isMcpEntry = args[0] === "--mcp" || args[0] === "mcp"; const isMcpMode = isMcpEntry && args.length === 1; const isMcpDoctorMode = args[0] === "mcp" && args[1] === "doctor"; -// Only flags may follow the entry: `mcp add --http` stays Incur's `mcp add`. +// Only flags may follow the entry: `mcp add --http` stays Incur's `mcp add`, +// while `mcp -h --http` reaches the HTTP entry's own help. const isMcpHttpMode = isMcpEntry && - (args[1]?.startsWith("--") ?? false) && + (args[1]?.startsWith("-") ?? false) && args.slice(1).includes("--http"); /** Exit status of a writer terminated by SIGPIPE (128 + 13). */ const SIGPIPE_EXIT_STATUS = 141; diff --git a/src/server/http/serveStreamableHttp.test.ts b/src/server/http/serveStreamableHttp.test.ts index 0be59585ef..c21bb2bb4c 100644 --- a/src/server/http/serveStreamableHttp.test.ts +++ b/src/server/http/serveStreamableHttp.test.ts @@ -216,14 +216,108 @@ it("reports a port that cannot be bound as a typed listen error naming the cause await expect(failed).rejects.toThrow(/EADDRINUSE/u); }); -it("keeps the SDK loopback allowlist only for loopback binds", () => { - expect(hostHeaderAllowlist("127.0.0.1")).toEqual([ - "localhost", - "127.0.0.1", - "[::1]", +it("keeps the SDK loopback allowlist, plus the bind's own hostname, for every loopback spelling", () => { + const sdk = ["localhost", "127.0.0.1", "[::1]"]; + expect(hostHeaderAllowlist("127.0.0.1")).toEqual(sdk); + expect(hostHeaderAllowlist("localhost")).toEqual(sdk); + expect(hostHeaderAllowlist("LOCALHOST")).toEqual(sdk); + expect(hostHeaderAllowlist("::1")).toEqual(sdk); + expect(hostHeaderAllowlist("0:0:0:0:0:0:0:1")).toEqual(sdk); + expect(hostHeaderAllowlist("127.0.0.2")).toEqual([...sdk, "127.0.0.2"]); + expect(hostHeaderAllowlist("localhost.")).toEqual([...sdk, "localhost."]); + expect(hostHeaderAllowlist("::ffff:127.0.0.1")).toEqual([ + ...sdk, + "[::ffff:7f00:1]", ]); - expect(hostHeaderAllowlist("localhost")).toBeDefined(); - expect(hostHeaderAllowlist("::1")).toBeDefined(); expect(hostHeaderAllowlist("192.0.2.10")).toBeUndefined(); + expect(hostHeaderAllowlist("0.0.0.0")).toBeUndefined(); expect(hostHeaderAllowlist("::")).toBeUndefined(); + expect(hostHeaderAllowlist("example.com")).toBeUndefined(); + expect(hostHeaderAllowlist("not a host")).toBeUndefined(); +}); + +it("refuses requests once the slot is closed and joins an initialize that is still arming", async () => { + let shutdowns = 0; + const slot = new McpSessionSlot( + createEchoServer(() => { + shutdowns += 1; + }), + undefined, + ); + const arming = slot.handle(initializeRequest()); + await slot.close(); + expect((await arming).status).toBe(503); + expect(shutdowns).toBe(1); + expect((await slot.handle(initializeRequest())).status).toBe(503); +}); + +it("retries unfinished cleanup after a close stage fails instead of replaying the failure", async () => { + let failures = 1; + const { served } = await start({ + createMcpServer: () => { + const server = createEchoServer(() => undefined); + const original = server.close.bind(server); + server.close = async () => { + if (failures > 0) { + failures -= 1; + throw new Error("provider cleanup failed"); + } + await original(); + }; + return server; + }, + }); + await expect(served.close()).rejects.toThrow("provider cleanup failed"); + await expect(served.close()).resolves.toBeUndefined(); + await expect(rawHttpRequest(new URL(served.url))).rejects.toMatchObject({ + code: "ECONNREFUSED", + }); +}); + +it("leaves a sessionless POST without the streamable headers or with a foreign Origin to refusal, keeping the session", async () => { + const { served } = await start(); + const { client } = await connect(served.url); + const url = new URL(served.url); + const body = await initializeRequest().text(); + const host = `127.0.0.1:${url.port}`; + const plain = await rawHttpRequest(url, { + method: "POST", + headers: { + host, + "content-type": "text/plain", + accept: "application/json, text/event-stream", + }, + body, + }); + expect(plain.status).toBeGreaterThanOrEqual(400); + expect(plain.status).toBeLessThan(500); + const foreign = await rawHttpRequest(url, { + method: "POST", + headers: { + host, + origin: "http://attacker.example", + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body, + }); + expect(foreign.status).toBe(403); + expect(await echo(client, "still here")).toEqual([ + { type: "text", text: "still here" }, + ]); + await client.close(); +}); + +it("announces the bound URL even when the log level hides info lines", async () => { + const logs: string[] = []; + const logger = pino({ level: "warn" }, { write: (line) => logs.push(line) }); + const { served } = await start({ logger }); + expect( + logs.map((line) => JSON.parse(line) as Record), + ).toContainEqual( + expect.objectContaining({ + msg: "MCP transport listening", + url: served.url, + }), + ); }); diff --git a/src/server/http/serveStreamableHttp.ts b/src/server/http/serveStreamableHttp.ts index 2a2fdbdc05..6a396d1d69 100644 --- a/src/server/http/serveStreamableHttp.ts +++ b/src/server/http/serveStreamableHttp.ts @@ -2,6 +2,7 @@ import { hostHeaderValidationResponse, isInitializeRequest, localhostAllowedHostnames, + originValidationResponse, readRequestBody, WebStandardStreamableHTTPServerTransport, type McpServer, @@ -15,6 +16,7 @@ import { type IncomingMessage, type ServerResponse, } from "node:http"; +import { isIPv4, isIPv6 } from "node:net"; import type { Logger } from "pino"; import { toWebRequest, writeWebResponse } from "./nodeHttpBridge.js"; @@ -23,14 +25,6 @@ import { StreamableHttpListenError } from "./streamableHttpListenError.js"; /** MCP endpoint path shared by `rea mcp --http` and agent environment servers. */ export const MCP_HTTP_PATH = "/mcp"; -/** Bind literals that keep the SDK loopback Host allowlist. */ -const LOOPBACK_HOSTS: ReadonlySet = new Set([ - "127.0.0.1", - "localhost", - "::1", - "[::1]", -]); - /** One plain HTTP route served beside the MCP endpoint as web Request to Response. */ export interface StreamableHttpRoute { readonly method: string; @@ -55,9 +49,41 @@ export interface StreamableHttpServer { readonly close: () => Promise; } -/** Host header allowlist for a bind, or undefined when validation is relaxed. */ -export const hostHeaderAllowlist = (host: string): string[] | undefined => - LOOPBACK_HOSTS.has(host) ? localhostAllowedHostnames() : undefined; +/** The bind host as the SDK compares Host headers: lowercase, IPv6 bracketed and canonical. */ +const normalizedHostname = (host: string): string | undefined => { + try { + return new URL(`http://${isIPv6(host) ? `[${host}]` : host}`).hostname; + } catch { + return undefined; + } +}; +const isLoopbackIpv4 = (address: string): boolean => + isIPv4(address) && address.startsWith("127."); +/** Loopback by address class and localhost names, so no spelling of a local bind skips validation. */ +const isLoopbackHostname = (hostname: string): boolean => { + if ( + hostname === "localhost" || + hostname === "localhost." || + hostname.endsWith(".localhost") || + hostname.endsWith(".localhost.") + ) + return true; + if (isLoopbackIpv4(hostname)) return true; + if (!hostname.startsWith("[") || !hostname.endsWith("]")) return false; + const inner = hostname.slice(1, -1); + if (inner === "::1") return true; + if (!inner.startsWith("::ffff:")) return false; + const mapped = inner.slice("::ffff:".length); + if (isLoopbackIpv4(mapped)) return true; + const hex = /^([0-9a-f]{1,4}):[0-9a-f]{1,4}$/u.exec(mapped); + return hex !== null && Number.parseInt(hex[1] ?? "0", 16) >> 8 === 0x7f; +}; +/** Host and Origin allowlist for a loopback bind, including the bind's own hostname; undefined relaxes validation. */ +export const hostHeaderAllowlist = (host: string): string[] | undefined => { + const hostname = normalizedHostname(host); + if (hostname === undefined || !isLoopbackHostname(hostname)) return undefined; + return [...new Set([...localhostAllowedHostnames(), hostname])]; +}; const jsonError = (status: number, code: number, message: string): Response => Response.json( @@ -72,10 +98,27 @@ const describe = (cause: unknown): string => cause instanceof Error ? cause.message : String(cause); const NO_SHUTDOWN_HOOK = (): void => undefined; +const SHUTTING_DOWN = (): Response => + jsonError(503, -32000, "Server shutting down"); + +/** The headers the SDK requires of a streamable POST; a request without them is left to the transport's refusal. */ +const hasStreamablePostHeaders = (request: Request): boolean => { + const accept = request.headers.get("accept") ?? ""; + const contentType = request.headers.get("content-type") ?? ""; + return ( + accept.includes("application/json") && + accept.includes("text/event-stream") && + /^application\/json\b/iu.test(contentType.trim()) + ); +}; /** Only a new client's initialize travels without a session id; peek within the SDK body bound. */ const isInitializePost = async (request: Request): Promise => { - if (request.method !== "POST" || request.headers.has("mcp-session-id")) + if ( + request.method !== "POST" || + request.headers.has("mcp-session-id") || + !hasStreamablePostHeaders(request) + ) return false; const probe = request.clone(); try { @@ -102,6 +145,7 @@ interface LiveSession { export class McpSessionSlot { #live: LiveSession | undefined; #arming: Promise | undefined; + #closed = false; /** The instance's own close hook, captured once so every session end restores exactly it. */ readonly #shutdownHook: () => void; @@ -114,25 +158,34 @@ export class McpSessionSlot { /** Route one MCP request; a new client's initialize replaces a session its predecessor left open. */ async handle(request: Request): Promise { + if (this.#closed) return SHUTTING_DOWN(); const open = this.#live; if ( open?.transport.sessionId !== undefined && (await isInitializePost(request)) ) { + if (this.#closed) return SHUTTING_DOWN(); this.logger?.info("MCP HTTP session replaced by a new client"); await open.transport.close(); } - return (await this.#current()).handleRequest(request); + const transport = await this.#current(); + // Shutdown may have closed the slot while this request was arming it. + if (this.#closed) return SHUTTING_DOWN(); + return transport.handleRequest(request); } - /** End the live session as part of server shutdown, so the instance's onclose runs. */ + /** End the live session as part of server shutdown, so the instance's onclose runs; later requests are refused. */ async close(): Promise { + this.#closed = true; + await this.#arming?.catch(() => undefined); const open = this.#live; this.#live = undefined; await open?.shutdown(); } #current(): Promise { + if (this.#closed) + return Promise.reject(new Error("MCP HTTP session slot is closed")); if (this.#live !== undefined) return Promise.resolve(this.#live.transport); this.#arming ??= this.#arm().finally(() => { this.#arming = undefined; @@ -194,10 +247,12 @@ export const serveStreamableHttp = async ( let stopped = false; const route = async (request: Request): Promise => { if (stopped) return jsonError(503, -32000, "Server shutting down"); + // One loopback allowlist gates Host (DNS rebinding) and Origin (browser pages); a missing Origin passes. const refused = allowedHosts === undefined ? undefined - : hostHeaderValidationResponse(request, allowedHosts); + : (hostHeaderValidationResponse(request, allowedHosts) ?? + originValidationResponse(request, allowedHosts)); if (refused !== undefined) return refused; const { pathname } = new URL(request.url); if (pathname === MCP_HTTP_PATH) return slot.handle(request); @@ -249,13 +304,12 @@ export const serveStreamableHttp = async ( { host }, "MCP HTTP Host header validation relaxed for a non-loopback bind", ); - logger?.info( - { transport: "streamable-http", url }, - "MCP transport listening", - ); - let closing: Promise | undefined; - const close = (): Promise => { - closing ??= (async () => { + // The URL is the only way to reach an ephemeral port, so the announcement ignores the log threshold. + logger + ?.child({}, { level: "info" }) + .info({ transport: "streamable-http", url }, "MCP transport listening"); + const close = stagedClose([ + async () => { stopped = true; httpServer.closeAllConnections(); await new Promise((resolve, reject) => { @@ -263,10 +317,32 @@ export const serveStreamableHttp = async ( error === undefined ? resolve() : reject(error), ); }); - await slot.close(); - await instance.close(); - })(); - return closing; - }; + }, + () => slot.close(), + () => instance.close(), + ]); return { url, close }; }; + +/** Run close stages in order once each; a failed stage is retried by the next call instead of replaying its rejection. */ +const stagedClose = ( + stages: readonly (() => Promise)[], +): (() => Promise) => { + const done = stages.map(() => false); + let closing: Promise | undefined; + return () => { + if (closing !== undefined) return closing; + const attempt = (async () => { + for (const [index, stage] of stages.entries()) { + if (done[index]) continue; + await stage(); + done[index] = true; + } + })(); + closing = attempt; + void attempt.catch(() => { + if (closing === attempt) closing = undefined; + }); + return attempt; + }; +}; diff --git a/tests/boundary/cli/dispatcher.test.ts b/tests/boundary/cli/dispatcher.test.ts index 33fab705af..316a2cff21 100644 --- a/tests/boundary/cli/dispatcher.test.ts +++ b/tests/boundary/cli/dispatcher.test.ts @@ -100,20 +100,16 @@ describe("executable dispatcher", () => { adapter: "cli", args: ["--mcp", "extra"], }); - await expect(invoke(["mcp", "--http", "--port", "0"])).resolves.toEqual({ - adapter: "mcp-http", - args: ["--http", "--port", "0"], - }); - await expect(invoke(["--mcp", "--http"])).resolves.toEqual({ - adapter: "mcp-http", - args: ["--http"], - }); - await expect( - invoke(["mcp", "--host", "127.0.0.1", "--http"]), - ).resolves.toEqual({ - adapter: "mcp-http", - args: ["--host", "127.0.0.1", "--http"], - }); + for (const args of [ + ["mcp", "--http", "--port", "0"], + ["--mcp", "--http"], + ["mcp", "-h", "--http"], + ["mcp", "--host", "127.0.0.1", "--http"], + ]) + await expect(invoke(args)).resolves.toEqual({ + adapter: "mcp-http", + args: args.slice(1), + }); await expect(invoke(["mcp", "doctor", "--http"])).resolves.toEqual({ adapter: "mcp-doctor", args: ["--http"],