diff --git a/src/gateway/mcp-http.request.ts b/src/gateway/mcp-http.request.ts index a6b3ddf50bbb..bfee0c8cba8b 100644 --- a/src/gateway/mcp-http.request.ts +++ b/src/gateway/mcp-http.request.ts @@ -14,7 +14,25 @@ import { isLoopbackAddress } from "./net.js"; import { checkBrowserOrigin } from "./origin-check.js"; const MAX_MCP_BODY_BYTES = 1_048_576; +const DEFAULT_MCP_BODY_TIMEOUT_MS = 30_000; const MCP_HTTP_BODY_TOO_LARGE_CODE = "ETOOBIG"; +const MCP_HTTP_BODY_TIMEOUT_CODE = "ETIMEDOUT"; +const MCP_HTTP_BODY_CLOSED_CODE = "ECONNRESET"; + +function readPositiveIntEnv(name: string, fallback: number): number { + const raw = process.env[name]?.trim(); + if (!raw) { + return fallback; + } + if (!/^\d+$/u.test(raw)) { + throw new Error(`${name} must be a positive integer. Got: ${JSON.stringify(raw)}`); + } + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new Error(`${name} must be a positive integer. Got: ${JSON.stringify(raw)}`); + } + return parsed; +} function shouldLogMcpLoopbackHttp(): boolean { return ( @@ -173,33 +191,40 @@ export function validateMcpLoopbackRequest(params: { return { senderIsOwner }; } -export async function readMcpHttpBody(req: IncomingMessage): Promise { +export async function readMcpHttpBody( + req: IncomingMessage, + options: { maxBytes?: number; timeoutMs?: number } = {}, +): Promise { return await new Promise((resolve, reject) => { + const maxBytes = Math.max(1, Math.floor(options.maxBytes ?? MAX_MCP_BODY_BYTES)); + const timeoutMs = Math.max(1, Math.floor(options.timeoutMs ?? DEFAULT_MCP_BODY_TIMEOUT_MS)); const chunks: Buffer[] = []; let received = 0; let settled = false; // Remove listeners on every terminal path; oversized bodies keep the error // listener briefly so Node can deliver the pause/error safely. - const cleanup = (options?: { keepErrorListener?: boolean }) => { + const cleanup = (cleanupOptions?: { keepErrorListener?: boolean }) => { req.off("data", onData); req.off("end", onEnd); - if (options?.keepErrorListener !== true) { + req.off("close", onClose); + if (cleanupOptions?.keepErrorListener !== true) { req.off("error", onError); } + clearTimeout(timeout); }; - const rejectOnce = (error: Error, options?: { keepErrorListener?: boolean }) => { + const rejectOnce = (error: Error, rejectOptions?: { keepErrorListener?: boolean }) => { if (settled) { return; } settled = true; - cleanup(options); + cleanup(rejectOptions); reject(error); }; const onData = (chunk: Buffer) => { received += chunk.length; - if (received > MAX_MCP_BODY_BYTES) { + if (received > maxBytes) { req.pause(); - rejectOnce(createMcpHttpBodyTooLargeError(), { keepErrorListener: true }); + rejectOnce(createMcpHttpBodyTooLargeError(maxBytes), { keepErrorListener: true }); return; } chunks.push(chunk); @@ -215,18 +240,40 @@ export async function readMcpHttpBody(req: IncomingMessage): Promise { const onError = (error: Error) => { rejectOnce(error); }; + const onClose = () => { + rejectOnce(createMcpHttpBodyClosedError()); + }; + const timeout = setTimeout(() => { + req.pause(); + rejectOnce(createMcpHttpBodyTimeoutError(), { keepErrorListener: true }); + }, timeoutMs); + timeout.unref?.(); + req.on("data", onData); req.on("end", onEnd); + req.on("close", onClose); req.on("error", onError); }); } -function createMcpHttpBodyTooLargeError(): Error & { code: string } { - return Object.assign(new Error(`Request body exceeds ${MAX_MCP_BODY_BYTES} bytes`), { +function createMcpHttpBodyTooLargeError(maxBytes: number): Error & { code: string } { + return Object.assign(new Error(`Request body exceeds ${maxBytes} bytes`), { code: MCP_HTTP_BODY_TOO_LARGE_CODE, }); } +function createMcpHttpBodyTimeoutError(): Error & { code: string } { + return Object.assign(new Error("Request body timed out"), { + code: MCP_HTTP_BODY_TIMEOUT_CODE, + }); +} + +function createMcpHttpBodyClosedError(): Error & { code: string } { + return Object.assign(new Error("Request body connection closed"), { + code: MCP_HTTP_BODY_CLOSED_CODE, + }); +} + export function isMcpHttpBodyTooLargeError(error: unknown): error is Error & { code: string } { return ( typeof error === "object" && @@ -235,6 +282,18 @@ export function isMcpHttpBodyTooLargeError(error: unknown): error is Error & { c ); } +export function isMcpHttpBodyTimeoutError(error: unknown): error is Error & { code: string } { + return ( + typeof error === "object" && + error !== null && + (error as { code?: unknown }).code === MCP_HTTP_BODY_TIMEOUT_CODE + ); +} + +export function resolveMcpHttpBodyTimeoutMs(): number { + return readPositiveIntEnv("OPENCLAW_MCP_LOOPBACK_BODY_TIMEOUT_MS", DEFAULT_MCP_BODY_TIMEOUT_MS); +} + export function resolveMcpRequestContext( req: IncomingMessage, cfg: OpenClawConfig, diff --git a/src/gateway/mcp-http.test.ts b/src/gateway/mcp-http.test.ts index 92a7902e2723..903a4d75f78a 100644 --- a/src/gateway/mcp-http.test.ts +++ b/src/gateway/mcp-http.test.ts @@ -185,6 +185,75 @@ async function sendChunkedOversizedBody(params: { }); } +async function sendStalledBody(params: { + port: number; + token: string; +}): Promise<{ status: number | undefined; body: string; closed: boolean }> { + return await new Promise((resolve, reject) => { + let sawResponse = false; + let closed = false; + let settled = false; + const req = request( + { + hostname: "127.0.0.1", + port: params.port, + path: "/mcp", + method: "POST", + headers: { + authorization: `Bearer ${params.token}`, + "content-type": "application/json", + "transfer-encoding": "chunked", + }, + }, + (res) => { + sawResponse = true; + let body = ""; + res.setEncoding("utf8"); + res.on("data", (chunk) => { + body += chunk; + }); + res.on("end", () => { + const waitForClose = new Promise((closeResolve) => { + if (closed) { + closeResolve(); + return; + } + req.once("close", () => closeResolve()); + setTimeout(closeResolve, 250).unref(); + }); + void waitForClose.then(() => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + resolve({ status: res.statusCode, body, closed }); + }); + }); + }, + ); + const timeout = setTimeout(() => { + if (settled) { + return; + } + settled = true; + req.destroy(); + reject(new Error("stalled body test timed out")); + }, 2_000); + req.on("close", () => { + closed = true; + }); + req.on("error", (error) => { + if (!sawResponse && !settled) { + settled = true; + clearTimeout(timeout); + reject(error); + } + }); + req.write("{"); + }); +} + async function startLoopbackServerForTest(port = 0) { server = await startMcpLoopbackServer(port); const runtime = getActiveMcpLoopbackRuntime(); @@ -974,6 +1043,35 @@ describe("mcp loopback server", () => { }); }); + it("times out stalled request bodies and closes uploads after flushing 408", async () => { + const previousTimeout = process.env.OPENCLAW_MCP_LOOPBACK_BODY_TIMEOUT_MS; + process.env.OPENCLAW_MCP_LOOPBACK_BODY_TIMEOUT_MS = "20"; + try { + server = await startMcpLoopbackServer(0); + const runtime = getActiveMcpLoopbackRuntime(); + if (!runtime) { + throw new Error("expected active MCP loopback runtime"); + } + + const response = await sendStalledBody({ + port: server.port, + token: runtime.ownerToken, + }); + + expect(response).toEqual({ + status: 408, + body: '{"error":"request_body_timeout"}', + closed: true, + }); + } finally { + if (previousTimeout === undefined) { + delete process.env.OPENCLAW_MCP_LOOPBACK_BODY_TIMEOUT_MS; + } else { + process.env.OPENCLAW_MCP_LOOPBACK_BODY_TIMEOUT_MS = previousTimeout; + } + } + }); + it("rejects cross-origin browser requests before auth", async () => { await expectBrowserToolsListStatus({ origin: "https://evil.example", diff --git a/src/gateway/mcp-http.ts b/src/gateway/mcp-http.ts index 990bac166a87..3f25bd0a9506 100644 --- a/src/gateway/mcp-http.ts +++ b/src/gateway/mcp-http.ts @@ -19,7 +19,9 @@ import { import { jsonRpcError, type JsonRpcRequest } from "./mcp-http.protocol.js"; import { isMcpHttpBodyTooLargeError, + isMcpHttpBodyTimeoutError, readMcpHttpBody, + resolveMcpHttpBodyTimeoutMs, resolveMcpRequestContext, validateMcpLoopbackRequest, } from "./mcp-http.request.js"; @@ -152,7 +154,7 @@ export async function startMcpLoopbackServer(port = 0): Promise<{ void (async () => { let parsed: JsonRpcRequest | JsonRpcRequest[] | undefined; try { - const body = await readMcpHttpBody(req); + const body = await readMcpHttpBody(req, { timeoutMs: resolveMcpHttpBodyTimeoutMs() }); parsed = parseMcpJsonBody(body); const cfg = getRuntimeConfig(); const requestContext = resolveMcpRequestContext(req, cfg, auth); @@ -237,6 +239,11 @@ export async function startMcpLoopbackServer(port = 0): Promise<{ res.end(JSON.stringify({ error: "payload_too_large" }), () => { req.destroy(); }); + } else if (isMcpHttpBodyTimeoutError(error)) { + res.writeHead(408, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: "request_body_timeout" }), () => { + req.destroy(); + }); } else if (isMcpJsonParseError(error)) { res.writeHead(400, { "Content-Type": "application/json" }); res.end(JSON.stringify(jsonRpcError(null, -32700, "Parse error")));