fix(gateway): timeout stalled mcp bodies

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<string> {
+export async function readMcpHttpBody(
+  req: IncomingMessage,
+  options: { maxBytes?: number; timeoutMs?: number } = {},
+): Promise<string> {
   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<string> {
     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,

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<void>((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",

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")));