mirror of
https://github.com/morluto/rea.git
synced 2026-11-07 07:16:13 +00:00
fix(mcp): compare the media type proper and bind before creating the server
Addresses the second automated review round: a sessionless POST replaces a session only when its media type proper is application/json, and the server instance is created after the listener binds so a bind failure leaves nothing to clean up and a factory failure releases the listener. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
b0ec63cc03
commit
9ed8f06f75
4 files changed
+88
-26
No files matched your search
@@ -18,7 +18,7 @@ client registration keep stdio.
|
||||
environment snapshot, then `127.0.0.1` and port `0`; announce the bound URL as
|
||||
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
|
||||
server instance built once, after the listener has bound. `DELETE` ends the session on its transport without
|
||||
running the instance's shutdown hooks, so the same instance keeps its
|
||||
providers and serves the next client on a fresh transport; the hooks run when
|
||||
the listener closes.
|
||||
|
||||
@@ -196,3 +196,11 @@ 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.
|
||||
|
||||
A second automated round on the fix (Codex, two P2 findings) was addressed in
|
||||
the next commit: the replacement guard compares the media type proper, so
|
||||
`application/json-patch+json` no longer counts as JSON (test: such an
|
||||
initialize is refused while the connected client keeps echoing), and the
|
||||
server instance is created only after the listener has bound, so a bind
|
||||
failure has nothing to clean up and a factory failure releases the listener
|
||||
and rethrows (test: a throwing factory rejects without announcing a URL).
|
||||
@@ -321,3 +321,41 @@ it("announces the bound URL even when the log level hides info lines", async ()
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("does not let a JSON-suffixed media type replace the session", async () => {
|
||||
const { served } = await start();
|
||||
const { client } = await connect(served.url);
|
||||
const url = new URL(served.url);
|
||||
const patch = await rawHttpRequest(url, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
host: `127.0.0.1:${url.port}`,
|
||||
"content-type": "application/json-patch+json",
|
||||
accept: "application/json, text/event-stream",
|
||||
},
|
||||
body: await initializeRequest().text(),
|
||||
});
|
||||
expect(patch.status).toBeGreaterThanOrEqual(400);
|
||||
expect(patch.status).toBeLessThan(500);
|
||||
expect(await echo(client, "unchanged")).toEqual([
|
||||
{ type: "text", text: "unchanged" },
|
||||
]);
|
||||
await client.close();
|
||||
});
|
||||
|
||||
it("releases the listener when the server factory fails after the port is bound", async () => {
|
||||
const logs: string[] = [];
|
||||
const logger = pino({ level: "debug" }, { write: (line) => logs.push(line) });
|
||||
const seen: string[] = [];
|
||||
const failing = serveStreamableHttp({
|
||||
host: "127.0.0.1",
|
||||
port: 0,
|
||||
logger,
|
||||
createMcpServer: () => {
|
||||
throw new Error("provider unavailable");
|
||||
},
|
||||
});
|
||||
await expect(failing).rejects.toThrow("provider unavailable");
|
||||
for (const line of logs) seen.push((JSON.parse(line) as { msg: string }).msg);
|
||||
expect(seen).not.toContain("MCP transport listening");
|
||||
});
|
||||
@@ -105,10 +105,12 @@ const SHUTTING_DOWN = (): Response =>
|
||||
const hasStreamablePostHeaders = (request: Request): boolean => {
|
||||
const accept = request.headers.get("accept") ?? "";
|
||||
const contentType = request.headers.get("content-type") ?? "";
|
||||
// The media type proper, without parameters: `application/json-patch+json` is not JSON here.
|
||||
const essence = (contentType.split(";")[0] ?? "").trim().toLowerCase();
|
||||
return (
|
||||
accept.includes("application/json") &&
|
||||
accept.includes("text/event-stream") &&
|
||||
/^application\/json\b/iu.test(contentType.trim())
|
||||
essence === "application/json"
|
||||
);
|
||||
};
|
||||
|
||||
@@ -240,10 +242,9 @@ export const serveStreamableHttp = async (
|
||||
): Promise<StreamableHttpServer> => {
|
||||
const { host, port, logger, routes = [] } = options;
|
||||
const allowedHosts = hostHeaderAllowlist(host);
|
||||
const instance = await options.createMcpServer({ era: "legacy" });
|
||||
protocolOf(instance).onerror = (error) =>
|
||||
logger?.warn({ error: error.message }, "MCP HTTP request rejected");
|
||||
const slot = new McpSessionSlot(instance, logger);
|
||||
// The instance is created after the listener binds, so a bind failure leaves nothing to clean up.
|
||||
let slot: McpSessionSlot | undefined;
|
||||
let origin = "";
|
||||
let stopped = false;
|
||||
const route = async (request: Request): Promise<Response> => {
|
||||
if (stopped) return jsonError(503, -32000, "Server shutting down");
|
||||
@@ -255,7 +256,10 @@ export const serveStreamableHttp = async (
|
||||
originValidationResponse(request, allowedHosts));
|
||||
if (refused !== undefined) return refused;
|
||||
const { pathname } = new URL(request.url);
|
||||
if (pathname === MCP_HTTP_PATH) return slot.handle(request);
|
||||
if (pathname === MCP_HTTP_PATH)
|
||||
return slot === undefined
|
||||
? jsonError(503, -32000, "Server starting")
|
||||
: slot.handle(request);
|
||||
const match = routes.find(
|
||||
(candidate) =>
|
||||
candidate.path === pathname && candidate.method === request.method,
|
||||
@@ -265,15 +269,6 @@ export const serveStreamableHttp = async (
|
||||
: match.handle(request);
|
||||
};
|
||||
const httpServer = createHttpServer();
|
||||
try {
|
||||
httpServer.listen(port, host);
|
||||
await once(httpServer, "listening");
|
||||
} catch (cause: unknown) {
|
||||
await instance.close().catch(() => undefined);
|
||||
throw new StreamableHttpListenError(cause);
|
||||
}
|
||||
const url = endpointUrl(httpServer.address());
|
||||
const origin = new URL(url).origin;
|
||||
const handleRequest = async (
|
||||
request: IncomingMessage,
|
||||
response: ServerResponse,
|
||||
@@ -299,6 +294,35 @@ export const serveStreamableHttp = async (
|
||||
void handleRequest(request, response);
|
||||
});
|
||||
httpServer.on("error", (error) => options.onerror?.(error));
|
||||
const closeListener = async (): Promise<void> => {
|
||||
stopped = true;
|
||||
httpServer.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
httpServer.close((error) =>
|
||||
error === undefined ? resolve() : reject(error),
|
||||
);
|
||||
});
|
||||
};
|
||||
try {
|
||||
httpServer.listen(port, host);
|
||||
await once(httpServer, "listening");
|
||||
} catch (cause: unknown) {
|
||||
throw new StreamableHttpListenError(cause);
|
||||
}
|
||||
const url = endpointUrl(httpServer.address());
|
||||
origin = new URL(url).origin;
|
||||
let instance: McpServer | Server;
|
||||
try {
|
||||
instance = await options.createMcpServer({ era: "legacy" });
|
||||
} catch (cause: unknown) {
|
||||
// No instance exists yet, so the listener is the only resource to release.
|
||||
await closeListener().catch(() => undefined);
|
||||
throw cause;
|
||||
}
|
||||
protocolOf(instance).onerror = (error) =>
|
||||
logger?.warn({ error: error.message }, "MCP HTTP request rejected");
|
||||
const live = new McpSessionSlot(instance, logger);
|
||||
slot = live;
|
||||
if (allowedHosts === undefined)
|
||||
logger?.warn(
|
||||
{ host },
|
||||
@@ -309,16 +333,8 @@ export const serveStreamableHttp = async (
|
||||
?.child({}, { level: "info" })
|
||||
.info({ transport: "streamable-http", url }, "MCP transport listening");
|
||||
const close = stagedClose([
|
||||
async () => {
|
||||
stopped = true;
|
||||
httpServer.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
httpServer.close((error) =>
|
||||
error === undefined ? resolve() : reject(error),
|
||||
);
|
||||
});
|
||||
},
|
||||
() => slot.close(),
|
||||
closeListener,
|
||||
() => live.close(),
|
||||
() => instance.close(),
|
||||
]);
|
||||
return { url, close };
|
||||
|
||||
Reference in new issue
Block a user