mirror of
https://github.com/morluto/rea.git
synced 2026-10-26 17:35:11 +00:00
fix(mcp): enforce initialization and discovery contracts
This commit is contained in:
1 parent
4c776d5918
commit
6eb9d09268
8 files changed
+188
-22
No files matched your search
@@ -702,7 +702,7 @@ the bound address is local; names are added only after classifying the actual IP
|
||||
On such a bind, requests whose `Host` header is outside the SDK loopback allowlist plus
|
||||
the configured name and the bound address 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. A reply sent before a request body was read (a refused `Host` or `Origin`, an unknown route, the startup `503`) closes the connection after the reply, so an unread upload cannot keep a socket open. Request targets are taken verbatim as origin-form paths (`//host/path` is a path, not another origin); an absolute-form target, a target containing a raw fragment delimiter (`#`), or one whose path URL parsing would rewrite (dot segments such as `/public/../mcp`), is refused with `400`, and a method Fetch does not support with `405`. Percent-encoded `%23` remains valid query data. Only a lone `initialize` can replace an open session; a batch that carries one is left to the transport's refusal. Scoped
|
||||
passes); a non-loopback bind relaxes both checks and logs the relaxation. A reply sent before a request body was read (a refused `Host` or `Origin`, an unknown route, the startup `503`) closes the connection after the reply, so an unread upload cannot keep a socket open. Request targets are taken verbatim as origin-form paths (`//host/path` is a path, not another origin); an absolute-form target, a target containing a raw fragment delimiter (`#`), or one whose path URL parsing would rewrite (dot segments such as `/public/../mcp`), is refused with `400`, and a method Fetch does not support with `405`. Percent-encoded `%23` remains valid query data. Only an `initialize` object can replace an open session. A sessionless array containing initialize, including a one-element array, is refused with HTTP `400` before session creation or handover. Scoped
|
||||
IPv6 literals (`fe80::1%eth0`) are refused before binding, because no URL can
|
||||
name them. Listener errors reach the runtime's connection-loss handler only
|
||||
after a successful bind; a bind failure prints the listen-failure message alone. Request bodies are bounded by the SDK's
|
||||
@@ -835,6 +835,11 @@ temporary source files are declared as `launchesProcess` and `writesFilesystem`,
|
||||
including for a single JavaScript source. This internal analysis does not execute
|
||||
the supplied application code.
|
||||
|
||||
`analyze_web_bundle` also declares `launchesProcess` and `writesFilesystem`:
|
||||
analysis of captured sources starts an owned worker and transfers source through
|
||||
private temporary files. Its network and session effects remain declared, and
|
||||
the worker does not execute the captured application code.
|
||||
|
||||
## Static Ghidra bytes and source mappings
|
||||
|
||||
`read_bytes` returns initialized analysis memory, including loader fixups, rather
|
||||
|
||||
@@ -61,6 +61,11 @@ The server advertises `listChanged: true`. Updating a registered guided prompt
|
||||
emits `notifications/prompts/list_changed`, allowing clients to refresh their
|
||||
cached prompt catalog.
|
||||
|
||||
Schema-invalid parameters to `prompts/get`, `prompts/list`,
|
||||
`completion/complete`, and `tools/list` return JSON-RPC `-32602` with field
|
||||
diagnostics. Correct the named fields and retry on the same connection.
|
||||
Unexpected handler failures retain JSON-RPC `-32603`.
|
||||
|
||||
## Session-aware completion
|
||||
|
||||
MCP `completion/complete` completes guided-prompt arguments; it does not define
|
||||
|
||||
@@ -113,6 +113,19 @@ const declaresForeignAuthority = (name: string): boolean => {
|
||||
};
|
||||
|
||||
describe("tool effects", () => {
|
||||
it("declares the web bundle worker's private file and child process effects", () => {
|
||||
const contract = TOOL_CONTRACTS.find(
|
||||
({ name }) => name === "analyze_web_bundle",
|
||||
);
|
||||
expect(contract?.effects).toMatchObject({
|
||||
mutatesSession: true,
|
||||
accessesNetwork: true,
|
||||
launchesProcess: true,
|
||||
writesFilesystem: true,
|
||||
executesForeignCode: false,
|
||||
mutatesTarget: false,
|
||||
});
|
||||
});
|
||||
it("audits every public tool exactly once", () => {
|
||||
expect(Object.keys(TOOL_EFFECTS).sort()).toEqual(
|
||||
TOOL_CONTRACTS.map(({ name }) => name).sort(),
|
||||
|
||||
@@ -626,7 +626,11 @@ export const TOOL_EFFECTS: Readonly<Record<string, ToolEffects>> = {
|
||||
project_managed_application_graph: sessionEvidence,
|
||||
list_browser_targets: browserEvidence,
|
||||
inspect_web_page: browserEvidence,
|
||||
analyze_web_bundle: browserEvidence,
|
||||
analyze_web_bundle: effects({
|
||||
...browserEvidence,
|
||||
launchesProcess: true,
|
||||
writesFilesystem: true,
|
||||
}),
|
||||
observe_web_session: browserEvidence,
|
||||
discover_webmcp_tools: browserEvidence,
|
||||
compare_web_captures: sessionEvidence,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
|
||||
import {
|
||||
specTypeSchemas,
|
||||
isCallToolResult,
|
||||
isJSONRPCErrorResponse,
|
||||
isJSONRPCRequest,
|
||||
@@ -313,6 +314,28 @@ class ErrorEvidenceTransport implements Transport {
|
||||
if (transport.hasPerRequestStream !== undefined)
|
||||
this.hasPerRequestStream = transport.hasPerRequestStream;
|
||||
transport.onmessage = (message, extra) => {
|
||||
// The pinned SDK classifies generic method schema failures as internal
|
||||
// errors. Validate these wire requests with its canonical schemas before
|
||||
// dispatch, leaving callback failures and negotiated-era checks to it.
|
||||
if (isJSONRPCRequest(message)) {
|
||||
const schema = discoveryRequestSchema(message.method);
|
||||
const result = schema?.["~standard"].validate(message);
|
||||
if (result?.issues !== undefined) {
|
||||
void this.send({
|
||||
jsonrpc: "2.0",
|
||||
id: message.id,
|
||||
error: {
|
||||
code: -32602,
|
||||
message: `Invalid ${message.method} request: ${JSON.stringify(result.issues)}`,
|
||||
},
|
||||
}).catch((cause: unknown) =>
|
||||
this.onerror?.(
|
||||
cause instanceof Error ? cause : new Error(String(cause)),
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (
|
||||
isJSONRPCRequest(message) &&
|
||||
message.method === "tools/call" &&
|
||||
@@ -697,3 +720,18 @@ class ErrorEvidenceTransport implements Transport {
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const discoveryRequestSchema = (method: string) => {
|
||||
switch (method) {
|
||||
case "prompts/get":
|
||||
return specTypeSchemas.GetPromptRequest;
|
||||
case "prompts/list":
|
||||
return specTypeSchemas.ListPromptsRequest;
|
||||
case "completion/complete":
|
||||
return specTypeSchemas.CompleteRequest;
|
||||
case "tools/list":
|
||||
return specTypeSchemas.ListToolsRequest;
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
@@ -793,27 +793,52 @@ it("refuses raw fragments without changing an encoded query value", async () =>
|
||||
});
|
||||
});
|
||||
|
||||
it("lets only a lone initialize replace a session, not a batch that carries one", async () => {
|
||||
it.each([1, 2])(
|
||||
"refuses a %i-message initialize batch without replacing a session",
|
||||
async (size) => {
|
||||
const { served } = await start();
|
||||
const { client } = await connect(served.url);
|
||||
const url = new URL(served.url);
|
||||
const single = JSON.parse(await initializeRequest().text()) as unknown;
|
||||
const batch = await rawHttpRequest(url, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
host: `127.0.0.1:${url.port}`,
|
||||
"content-type": "application/json",
|
||||
accept: "application/json, text/event-stream",
|
||||
},
|
||||
body: JSON.stringify(
|
||||
size === 1
|
||||
? [single]
|
||||
: [single, { jsonrpc: "2.0", method: "notifications/initialized" }],
|
||||
),
|
||||
});
|
||||
expect(batch.status).toBeGreaterThanOrEqual(400);
|
||||
expect(batch.status).toBeLessThan(500);
|
||||
expect(await echo(client, "not evicted")).toEqual([
|
||||
{ type: "text", text: "not evicted" },
|
||||
]);
|
||||
await client.close();
|
||||
},
|
||||
);
|
||||
|
||||
it("refuses an initialize batch before opening the first session", async () => {
|
||||
const { served } = await start();
|
||||
const { client } = await connect(served.url);
|
||||
const url = new URL(served.url);
|
||||
const single = JSON.parse(await initializeRequest().text()) as unknown;
|
||||
const batch = await rawHttpRequest(url, {
|
||||
const response = await rawHttpRequest(url, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
host: `127.0.0.1:${url.port}`,
|
||||
"content-type": "application/json",
|
||||
accept: "application/json, text/event-stream",
|
||||
},
|
||||
body: JSON.stringify([
|
||||
single,
|
||||
{ jsonrpc: "2.0", method: "notifications/initialized" },
|
||||
]),
|
||||
body: JSON.stringify([single]),
|
||||
});
|
||||
expect(batch.status).toBeGreaterThanOrEqual(400);
|
||||
expect(batch.status).toBeLessThan(500);
|
||||
expect(await echo(client, "not evicted")).toEqual([
|
||||
{ type: "text", text: "not evicted" },
|
||||
expect(response.status).toBe(400);
|
||||
const { client } = await connect(served.url);
|
||||
expect(await echo(client, "valid initialization")).toEqual([
|
||||
{ type: "text", text: "valid initialization" },
|
||||
]);
|
||||
await client.close();
|
||||
});
|
||||
|
||||
@@ -198,24 +198,27 @@ const hasStreamablePostHeaders = (request: Request): boolean => {
|
||||
};
|
||||
|
||||
/** Only a new client's initialize travels without a session id; peek within the SDK body bound. */
|
||||
const isInitializePost = async (request: Request): Promise<boolean> => {
|
||||
const classifyInitializePost = async (
|
||||
request: Request,
|
||||
): Promise<"initialize" | "batch" | undefined> => {
|
||||
if (
|
||||
request.method !== "POST" ||
|
||||
request.headers.has("mcp-session-id") ||
|
||||
!hasStreamablePostHeaders(request)
|
||||
)
|
||||
return false;
|
||||
return undefined;
|
||||
const probe = request.clone();
|
||||
try {
|
||||
const read = await readRequestBody(probe);
|
||||
if (read.tooLarge) return false;
|
||||
if (read.tooLarge) return undefined;
|
||||
const body: unknown = JSON.parse(read.text);
|
||||
// The transport accepts only a lone initialize, so only that may evict a session.
|
||||
return Array.isArray(body)
|
||||
? body.length === 1 && isInitializeRequest(body[0])
|
||||
: isInitializeRequest(body);
|
||||
// SDK 2.3.1 accepts a one-message array. REA's session handover contract
|
||||
// requires an object, regardless of whether a session is already open.
|
||||
if (Array.isArray(body))
|
||||
return body.some(isInitializeRequest) ? "batch" : undefined;
|
||||
return isInitializeRequest(body) ? "initialize" : undefined;
|
||||
} catch {
|
||||
return false;
|
||||
return undefined;
|
||||
} finally {
|
||||
// The clone is one branch of a tee; awaiting its cancel would wait for the original to drain.
|
||||
void probe.body?.cancel().catch(() => undefined);
|
||||
@@ -336,12 +339,19 @@ export class McpSessionSlot {
|
||||
|
||||
async #handle(request: Request): Promise<Response> {
|
||||
if (this.#closed) return SHUTTING_DOWN();
|
||||
const initialization = await classifyInitializePost(request);
|
||||
if (initialization === "batch")
|
||||
return jsonError(
|
||||
400,
|
||||
-32600,
|
||||
"Initialize must be a single JSON-RPC object, not a batch",
|
||||
);
|
||||
if (this.#draining) await this.#live?.transport.close();
|
||||
if (this.#closed) return SHUTTING_DOWN();
|
||||
const open = this.#live;
|
||||
if (
|
||||
open?.transport.sessionId !== undefined &&
|
||||
(await isInitializePost(request)) &&
|
||||
initialization === "initialize" &&
|
||||
// A client that left while its initialize was probed would evict the session for nobody.
|
||||
!request.signal.aborted
|
||||
) {
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
import {
|
||||
STDIO_DEFAULT_MAX_BUFFER_SIZE,
|
||||
type McpServer,
|
||||
type JSONRPCMessage,
|
||||
} from "@modelcontextprotocol/server";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
@@ -60,6 +61,71 @@ afterEach(async () => {
|
||||
});
|
||||
|
||||
describe("guided prompts over MCP", () => {
|
||||
it("classifies malformed discovery parameters separately from handler failures", async () => {
|
||||
const session = fixtureSession();
|
||||
const server = createServer({ kind: "session", session });
|
||||
const [peer, transport] = InMemoryTransport.createLinkedPair();
|
||||
resources.push(server, peer, session);
|
||||
await server.connect(transport);
|
||||
await peer.start();
|
||||
let id = 0;
|
||||
const request = (method: string, params: Record<string, unknown>) =>
|
||||
new Promise<JSONRPCMessage>((resolve, reject) => {
|
||||
id += 1;
|
||||
const requestId = id;
|
||||
peer.onmessage = (message) => {
|
||||
if ("id" in message && message.id === requestId) resolve(message);
|
||||
};
|
||||
void peer
|
||||
.send({ jsonrpc: "2.0", id: requestId, method, params })
|
||||
.catch(reject);
|
||||
});
|
||||
await request("initialize", {
|
||||
protocolVersion: "2025-06-18",
|
||||
capabilities: {},
|
||||
clientInfo: { name: "raw-parameter-test", version: "1" },
|
||||
});
|
||||
await peer.send({ jsonrpc: "2.0", method: "notifications/initialized" });
|
||||
for (const [method, params] of [
|
||||
[
|
||||
"prompts/get",
|
||||
{ name: "investigate_feature", arguments: { feature: 5 } },
|
||||
],
|
||||
["prompts/get", { name: "investigate_feature", arguments: null }],
|
||||
["prompts/list", { cursor: 5 }],
|
||||
[
|
||||
"completion/complete",
|
||||
{
|
||||
ref: { type: "ref/tool", name: "binary_session" },
|
||||
argument: { name: "foo", value: "" },
|
||||
},
|
||||
],
|
||||
["tools/list", { cursor: 5 }],
|
||||
] as const) {
|
||||
expect(await request(method, params)).toMatchObject({
|
||||
error: { code: -32602, message: expect.stringContaining("params") },
|
||||
});
|
||||
}
|
||||
expect(
|
||||
await request("prompts/get", {
|
||||
name: "investigate_feature",
|
||||
arguments: { feature: "offline search" },
|
||||
}),
|
||||
).toHaveProperty("result.messages");
|
||||
expect(await request("prompts/list", {})).toHaveProperty("result.prompts");
|
||||
expect(
|
||||
await request("completion/complete", {
|
||||
ref: { type: "ref/prompt", name: "investigate_feature" },
|
||||
argument: { name: "provider", value: "" },
|
||||
}),
|
||||
).toHaveProperty("result.completion");
|
||||
server.server.setRequestHandler("prompts/list", async () => {
|
||||
throw new Error("injected internal failure");
|
||||
});
|
||||
expect(await request("prompts/list", {})).toMatchObject({
|
||||
error: { code: -32603, message: "injected internal failure" },
|
||||
});
|
||||
});
|
||||
it("lists and renders all workflows without changing the tool inventory", async () => {
|
||||
const session = fixtureSession();
|
||||
const client = await connect(createServer({ kind: "session", session }));
|
||||
|
||||
Reference in new issue
Block a user