refactor(security): consolidate plugin JSON body reading onto canonical bounded reader (#124944)

* fix(security): harden canonical bounded JSON body reading * refactor(admin): use canonical bounded JSON body reader * refactor(telegram): use canonical bounded JSON body reader The assertion-safety baseline prune for extensions/telegram/src/miniapp/routes.ts (2 to 1) is explicitly approved. * test(security): cover canonical JSON body migrations * refactor(plugin-sdk): name response-first body profile * fix(telegram): flush miniapp body-limit responses before close
2026-08-28 13:26:04 -06:00 · 2026-08-16 22:01:34 -07:00
parent 1606969dca
commit 49b4775f30
9 changed files with 489 additions and 179 deletions
@@ -1304,7 +1304,7 @@ extensions/telegram/src/lane-delivery-text-deliverer.ts	1
 extensions/telegram/src/message-cache.ts	10
 extensions/telegram/src/miniapp/command.ts	1
 extensions/telegram/src/miniapp/init-data.ts	1
-extensions/telegram/src/miniapp/routes.ts	2
+extensions/telegram/src/miniapp/routes.ts	1
 extensions/telegram/src/monitor.ts	4
 extensions/telegram/src/network-config.ts	1
 extensions/telegram/src/network-errors.ts	8
@@ -14,6 +14,19 @@ vi.mock("openclaw/plugin-sdk/gateway-method-runtime", () => ({
  dispatchGatewayMethod,
 }));

+vi.mock("node:timers", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:timers")>();
+  return {
+    ...actual,
+    // The canonical reader deliberately captures Node timers. Route them through
+    // the test clock here so the 30-second response-flush contract stays fast.
+    setTimeout: ((callback: (...args: unknown[]) => void, delay?: number, ...args: unknown[]) =>
+      globalThis.setTimeout(callback, delay, ...args)) as typeof actual.setTimeout,
+    clearTimeout: ((timer: ReturnType<typeof globalThis.setTimeout> | undefined) =>
+      globalThis.clearTimeout(timer)) as typeof actual.clearTimeout,
+  };
+});
+
 type CapturedResponse = {
  statusCode: number;
  headers: Record<string, string | number | readonly string[]>;
@@ -48,6 +61,12 @@ function createHangingRequest() {
  return req as import("node:http").IncomingMessage;
 }

+function expectBodyReadListenersCleaned(req: import("node:http").IncomingMessage) {
+  for (const event of ["data", "end", "error", "close"] as const) {
+    expect(req.listenerCount(event), event).toBe(0);
+  }
+}
+
 function createResponse() {
  const captured: CapturedResponse = {
    statusCode: 200,
@@ -250,6 +269,20 @@ describe("admin-http-rpc plugin handler", () => {
    expect(dispatchGatewayMethod).not.toHaveBeenCalled();
  });

+  it.each([
+    ["", "request body must be JSON"],
+    ["{", "request body must be valid JSON"],
+  ])("preserves the invalid JSON response for %j", async (body, message) => {
+    const result = await invoke(body);
+
+    expect(result.captured.statusCode).toBe(400);
+    expect(result.json).toEqual({
+      ok: false,
+      error: { type: "invalid_request", message },
+    });
+    expect(dispatchGatewayMethod).not.toHaveBeenCalled();
+  });
+
  it("only accepts POST", async () => {
    const result = await invoke({ method: "status" }, "GET");

@@ -261,7 +294,8 @@ describe("admin-http-rpc plugin handler", () => {
  it("times out incomplete request bodies before dispatch", async () => {
    vi.useFakeTimers();
    try {
-      const resultPromise = invokeRequest(createHangingRequest());
+      const req = createHangingRequest();
+      const resultPromise = invokeRequest(req);
      await vi.advanceTimersByTimeAsync(30_000);
      const result = await resultPromise;

@@ -274,12 +308,77 @@ describe("admin-http-rpc plugin handler", () => {
          message: "Request body timeout",
        },
      });
+      expectBodyReadListenersCleaned(req);
      expect(dispatchGatewayMethod).not.toHaveBeenCalled();
    } finally {
      vi.useRealTimers();
    }
  });

+  it("settles an early client close and removes request-body listeners", async () => {
+    const req = createHangingRequest();
+    const resultPromise = invokeRequest(req);
+
+    req.emit("close");
+    const result = await resultPromise;
+
+    expect(result.captured.statusCode).toBe(400);
+    expect(result.json).toEqual({
+      ok: false,
+      error: {
+        type: "invalid_request",
+        message: "Connection closed",
+      },
+    });
+    expectBodyReadListenersCleaned(req);
+    expect(dispatchGatewayMethod).not.toHaveBeenCalled();
+  });
+
+  it("flushes a real HTTP 413 response before closing an oversized request", async () => {
+    const server = createServer((req, res) => {
+      void handleAdminHttpRpcRequest(req, res);
+    });
+    let socket: Socket | undefined;
+    try {
+      const port = await listen(server);
+      socket = connect({ host: "127.0.0.1", port });
+      await new Promise<void>((resolve) => {
+        socket?.once("connect", resolve);
+      });
+
+      socket.write(
+        [
+          "POST /api/v1/admin/rpc HTTP/1.1",
+          "Host: 127.0.0.1",
+          "Content-Type: application/json",
+          `Content-Length: ${1024 * 1024 + 1}`,
+          "Connection: keep-alive",
+          "",
+          "{",
+        ].join("\r\n"),
+      );
+
+      const response = await readSocketResponse(socket);
+      const [, rawBody = ""] = response.split("\r\n\r\n", 2);
+
+      expect(response).toContain("HTTP/1.1 413");
+      expect(response).toContain("Connection: close");
+      expect(JSON.parse(rawBody) as unknown).toEqual({
+        ok: false,
+        error: {
+          type: "invalid_request",
+          message: "Payload too large",
+        },
+      });
+      expect(dispatchGatewayMethod).not.toHaveBeenCalled();
+    } finally {
+      socket?.destroy();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    }
+  });
+
  it("delivers a real HTTP 408 response before closing timed-out partial bodies", async () => {
    vi.useFakeTimers();
    let markRequestStarted: (() => void) | undefined;
@@ -7,7 +7,7 @@ import type { IncomingMessage, ServerResponse } from "node:http";
 import { dispatchGatewayMethod } from "openclaw/plugin-sdk/gateway-method-runtime";
 import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
 import {
-  requestBodyErrorToText,
+  readJsonBodyWithLimit,
  WEBHOOK_BODY_READ_DEFAULTS,
 } from "openclaw/plugin-sdk/webhook-request-guards";
 import { isAdminHttpRpcAllowedMethod, listAdminHttpRpcAllowedMethods } from "./methods.js";
@@ -49,6 +49,7 @@ type RequestBodyLimitFailureCode =
  | "PAYLOAD_TOO_LARGE"
  | "REQUEST_BODY_TIMEOUT"
  | "CONNECTION_CLOSED";
+
 type ReadJsonBodyResult =
  | { ok: true; value: unknown }
  | {
@@ -57,9 +58,6 @@ type ReadJsonBodyResult =
      message: string;
      closeAfterResponse?: boolean;
    };
-type ReadRawBodyResult =
-  | { ok: true; raw: string }
-  | { ok: false; code: RequestBodyLimitFailureCode; closeAfterResponse?: boolean };

 function createError(code: string, message: string): RpcError {
  return { code, message };
@@ -97,32 +95,6 @@ function sendError(res: ServerResponse, status: number, error: { type: string; m
  sendJson(res, status, { ok: false, error });
 }

-async function readJsonBody(
-  req: IncomingMessage,
-  maxBytes: number,
-  timeoutMs: number,
-): Promise<ReadJsonBodyResult> {
-  const body = await readRawBodyWithLimit(req, maxBytes, timeoutMs);
-  if (!body.ok) {
-    return {
-      ok: false,
-      status: statusForBodyErrorCode(body.code),
-      message: requestBodyErrorToText(body.code),
-      closeAfterResponse: body.closeAfterResponse,
-    };
-  }
-
-  const raw = body.raw.trim();
-  if (!raw) {
-    return { ok: false, status: 400, message: "request body must be JSON" };
-  }
-  try {
-    return { ok: true, value: JSON.parse(raw) as unknown };
-  } catch {
-    return { ok: false, status: 400, message: "request body must be valid JSON" };
-  }
-}
-
 function statusForBodyErrorCode(code: RequestBodyLimitFailureCode): number {
  switch (code) {
    case "PAYLOAD_TOO_LARGE":
@@ -135,93 +107,32 @@ function statusForBodyErrorCode(code: RequestBodyLimitFailureCode): number {
  return 400;
 }

-async function readRawBodyWithLimit(
-  req: IncomingMessage,
-  maxBytes: number,
-  timeoutMs: number,
-): Promise<ReadRawBodyResult> {
-  const declaredLength = readDeclaredContentLength(req);
-  if (declaredLength !== null && declaredLength > maxBytes) {
-    req.pause();
-    return { ok: false, code: "PAYLOAD_TOO_LARGE", closeAfterResponse: true };
-  }
-
-  return await new Promise((resolve) => {
-    let done = false;
-    let ended = false;
-    let totalBytes = 0;
-    const chunks: Buffer[] = [];
-
-    const cleanup = () => {
-      req.removeListener("data", onData);
-      req.removeListener("end", onEnd);
-      req.removeListener("error", onError);
-      req.removeListener("close", onClose);
-      clearTimeout(timer);
-    };
-
-    const finish = (result: ReadRawBodyResult) => {
-      if (done) {
-        return;
-      }
-      done = true;
-      cleanup();
-      resolve(result);
-    };
-
-    const failAfterResponse = (code: RequestBodyLimitFailureCode) => {
-      req.pause();
-      finish({ ok: false, code, closeAfterResponse: true });
-    };
-
-    const timer = setTimeout(() => {
-      failAfterResponse("REQUEST_BODY_TIMEOUT");
-    }, timeoutMs);
-
-    const onData = (chunk: Buffer | string) => {
-      if (done) {
-        return;
-      }
-      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-      totalBytes += buffer.byteLength;
-      if (totalBytes > maxBytes) {
-        failAfterResponse("PAYLOAD_TOO_LARGE");
-        return;
-      }
-      chunks.push(buffer);
-    };
-
-    const onEnd = () => {
-      ended = true;
-      finish({ ok: true, raw: Buffer.concat(chunks).toString("utf8") });
-    };
-
-    const onError = () => {
-      finish({ ok: false, code: "CONNECTION_CLOSED" });
-    };
-
-    const onClose = () => {
-      if (done || ended) {
-        return;
-      }
-      finish({ ok: false, code: "CONNECTION_CLOSED" });
-    };
-
-    req.on("data", onData);
-    req.on("end", onEnd);
-    req.on("error", onError);
-    req.on("close", onClose);
+async function readAdminJsonBody(req: IncomingMessage): Promise<ReadJsonBodyResult> {
+  const body = await readJsonBodyWithLimit(req, {
+    // Admin responses are part of the client contract. The response-first profile
+    // defers destruction so closeRequestAfterResponse can flush the JSON error.
+    ...WEBHOOK_BODY_READ_DEFAULTS.postAuthResponseFirst,
+    emptyObjectOnEmpty: false,
  });
-}
-
-function readDeclaredContentLength(req: IncomingMessage): number | null {
-  const header = req.headers["content-length"];
-  const raw = Array.isArray(header) ? header[0] : header;
-  if (typeof raw !== "string" || !/^\d+$/.test(raw)) {
-    return null;
+  if (body.ok) {
+    return body;
  }
-  const value = Number(raw);
-  return Number.isSafeInteger(value) ? value : Number.MAX_SAFE_INTEGER;
+  if (body.code === "INVALID_JSON") {
+    return {
+      ok: false,
+      status: 400,
+      message:
+        body.error === "empty payload"
+          ? "request body must be JSON"
+          : "request body must be valid JSON",
+    };
+  }
+  return {
+    ok: false,
+    status: statusForBodyErrorCode(body.code),
+    message: body.error,
+    closeAfterResponse: body.code !== "CONNECTION_CLOSED",
+  };
 }

 function closeRequestAfterResponse(req: IncomingMessage, res: ServerResponse): void {
@@ -332,11 +243,7 @@ export async function handleAdminHttpRpcRequest(
    return true;
  }

-  const body = await readJsonBody(
-    req,
-    WEBHOOK_BODY_READ_DEFAULTS.postAuth.maxBytes,
-    WEBHOOK_BODY_READ_DEFAULTS.postAuth.timeoutMs,
-  );
+  const body = await readAdminJsonBody(req);
  if (!body.ok) {
    if (body.closeAfterResponse) {
      closeRequestAfterResponse(req, res);
@@ -1,5 +1,7 @@
 import crypto from "node:crypto";
-import type { IncomingMessage, ServerResponse } from "node:http";
+import { EventEmitter } from "node:events";
+import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
+import { connect, type Socket } from "node:net";
 import { Readable } from "node:stream";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry";
@@ -33,13 +35,25 @@ vi.mock("./url.js", async (importOriginal) => ({
  resolveTelegramMiniAppUrls,
 }));

+vi.mock("node:timers", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:timers")>();
+  return {
+    ...actual,
+    setTimeout: ((callback: (...args: unknown[]) => void, delay?: number, ...args: unknown[]) =>
+      globalThis.setTimeout(callback, delay, ...args)) as typeof actual.setTimeout,
+    clearTimeout: ((timer: ReturnType<typeof globalThis.setTimeout> | undefined) =>
+      globalThis.clearTimeout(timer)) as typeof actual.clearTimeout,
+  };
+});
+
 const { registerTelegramMiniAppRoutes } = await import("./routes.js");

 const BOT_TOKEN = "fixture";
+const AUTH_BODY_MAX_BYTES = 4096;
 let signedNonceSequence = 0;
 let launchTickets: TelegramMiniAppLaunchTickets;

-class MockResponse {
+class MockResponse extends EventEmitter {
  statusCode = 200;
  headers: Record<string, string> = {};
  body = "";
@@ -50,8 +64,14 @@ class MockResponse {
    return this;
  }

+  setHeader(name: string, value: string) {
+    this.headers[name] = value;
+    return this;
+  }
+
  end(body?: string) {
    this.body = body ?? "";
+    this.emit("finish");
    return this;
  }
 }
@@ -86,11 +106,69 @@ async function callRoute(params: {
  Object.defineProperty(req, "socket", {
    value: { remoteAddress: params.ip ?? "203.0.113.10" },
  });
+  return await callRouteRequest(params.route, req);
+}
+
+async function callRouteRequest(route: OpenClawPluginHttpRouteParams, req: IncomingMessage) {
  const res = new MockResponse() as ServerResponse & MockResponse;
-  await params.route.handler(req, res);
+  await route.handler(req, res);
  return res;
 }

+function createPendingAuthRequest(ip: string): IncomingMessage {
+  const req = new Readable({
+    read() {
+      // Keep the body open so the canonical reader owns timeout settlement.
+    },
+  }) as IncomingMessage;
+  req.method = "POST";
+  req.url = "/__openclaw_tg_miniapp/auth";
+  req.headers = { "content-type": "application/json" };
+  Object.defineProperty(req, "socket", { value: { remoteAddress: ip } });
+  return req;
+}
+
+function expectBodyReadListenersCleaned(req: IncomingMessage) {
+  for (const event of ["data", "end", "error", "close"] as const) {
+    expect(req.listenerCount(event), event).toBe(0);
+  }
+}
+
+async function listen(server: Server): Promise<number> {
+  await new Promise<void>((resolve) => {
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("expected TCP server address");
+  }
+  return address.port;
+}
+
+async function readSocketResponse(socket: Socket): Promise<string> {
+  const chunks: Buffer[] = [];
+  return await new Promise((resolve, reject) => {
+    let done = false;
+    const finish = () => {
+      if (done) {
+        return;
+      }
+      done = true;
+      resolve(Buffer.concat(chunks).toString("utf8"));
+    };
+    socket.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
+    socket.on("end", finish);
+    socket.on("close", finish);
+    socket.on("error", reject);
+  });
+}
+
+function createRouteServer(route: OpenClawPluginHttpRouteParams): Server {
+  return createServer((req, res) => {
+    void route.handler(req, res);
+  });
+}
+
 function config(allowFrom: string[] = ["123456"]): OpenClawConfig {
  return {
    channels: {
@@ -313,4 +391,154 @@ describe("registerTelegramMiniAppRoutes", () => {
    expect(last?.statusCode).toBe(429);
    expect(last?.body).toBe("Too many requests");
  });
+
+  it("keeps malformed JSON on the expired-link response", async () => {
+    const route = createRoute(config());
+    const res = await callRoute({
+      route,
+      method: "POST",
+      url: "/__openclaw_tg_miniapp/auth",
+      contentType: "application/json",
+      body: "{",
+      ip: "203.0.113.49",
+    });
+
+    expect(res.statusCode).toBe(401);
+    expect(res.body).toBe("This link expired. Reopen the dashboard from your bot chat.");
+    expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+  });
+
+  it("rejects an oversized auth body and closes the request", async () => {
+    const route = createRoute(config());
+    const req = Readable.from(["x".repeat(4097)]) as IncomingMessage;
+    req.method = "POST";
+    req.url = "/__openclaw_tg_miniapp/auth";
+    req.headers = { "content-type": "application/json" };
+    Object.defineProperty(req, "socket", { value: { remoteAddress: "203.0.113.50" } });
+
+    const res = await callRouteRequest(route, req);
+
+    expect(res.statusCode).toBe(413);
+    expect(res.body).toBe("Payload too large");
+    expect(req.destroyed).toBe(true);
+    expectBodyReadListenersCleaned(req);
+    expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+  });
+
+  it("flushes a real HTTP 413 response before closing an oversized auth request", async () => {
+    const server = createRouteServer(createRoute(config()));
+    let socket: Socket | undefined;
+    try {
+      const port = await listen(server);
+      socket = connect({ host: "127.0.0.1", port });
+      await new Promise<void>((resolve) => {
+        socket?.once("connect", resolve);
+      });
+
+      socket.write(
+        [
+          "POST /__openclaw_tg_miniapp/auth HTTP/1.1",
+          "Host: 127.0.0.1",
+          "Content-Type: application/json",
+          `Content-Length: ${AUTH_BODY_MAX_BYTES + 1}`,
+          "Connection: keep-alive",
+          "",
+          "{",
+        ].join("\r\n"),
+      );
+
+      const response = await readSocketResponse(socket);
+      const [, body = ""] = response.split("\r\n\r\n", 2);
+
+      expect(response).toContain("HTTP/1.1 413");
+      expect(response).toContain("Connection: close");
+      expect(body).toBe("Payload too large");
+      expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+    } finally {
+      socket?.destroy();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    }
+  });
+
+  it("settles an early client close without leaking request-body listeners", async () => {
+    const route = createRoute(config());
+    const req = createPendingAuthRequest("203.0.113.51");
+    const responsePromise = callRouteRequest(route, req);
+
+    req.emit("close");
+    const res = await responsePromise;
+
+    expect(res.statusCode).toBe(400);
+    expect(res.body).toBe("Connection closed");
+    expectBodyReadListenersCleaned(req);
+    expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+  });
+
+  it("times out a slow auth body and closes the request", async () => {
+    const route = createRoute(config());
+    const req = createPendingAuthRequest("203.0.113.52");
+    try {
+      const res = await callRouteRequest(route, req);
+
+      expect(res.statusCode).toBe(408);
+      expect(res.body).toBe("Request body timeout");
+      expect(req.destroyed).toBe(true);
+      expectBodyReadListenersCleaned(req);
+      expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+    } finally {
+      req.destroy();
+    }
+  }, 10_000);
+
+  it("flushes a real HTTP 408 response before closing a stalled auth request", async () => {
+    vi.useFakeTimers();
+    let markRequestStarted: (() => void) | undefined;
+    const requestStarted = new Promise<void>((resolve) => {
+      markRequestStarted = resolve;
+    });
+    const route = createRoute(config());
+    const server = createServer((req, res) => {
+      void route.handler(req, res);
+      markRequestStarted?.();
+    });
+    let socket: Socket | undefined;
+    try {
+      const port = await listen(server);
+      socket = connect({ host: "127.0.0.1", port });
+      await new Promise<void>((resolve) => {
+        socket?.once("connect", resolve);
+      });
+
+      socket.write(
+        [
+          "POST /__openclaw_tg_miniapp/auth HTTP/1.1",
+          "Host: 127.0.0.1",
+          "Content-Type: application/json",
+          "Content-Length: 64",
+          "Connection: keep-alive",
+          "",
+          "{",
+        ].join("\r\n"),
+      );
+
+      await requestStarted;
+      const responsePromise = readSocketResponse(socket);
+      await vi.advanceTimersByTimeAsync(5_000);
+      const response = await responsePromise;
+      const [, body = ""] = response.split("\r\n\r\n", 2);
+
+      expect(response).toContain("HTTP/1.1 408");
+      expect(response).toContain("Connection: close");
+      expect(body).toBe("Request body timeout");
+      expect(issueDeviceBootstrapToken).not.toHaveBeenCalled();
+    } finally {
+      vi.useRealTimers();
+      socket?.destroy();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    }
+  });
 });
@@ -7,6 +7,8 @@ import {
  issueDeviceBootstrapToken,
 } from "openclaw/plugin-sdk/device-bootstrap";
 import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry";
+import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
+import { readJsonWebhookBodyOrReject } from "openclaw/plugin-sdk/webhook-request-guards";
 import { resolveTelegramAccount } from "../accounts.js";
 import { validateTelegramMiniAppInitData } from "./init-data.js";
 import type { TelegramMiniAppLaunchTickets } from "./launch-ticket.js";
@@ -89,20 +91,28 @@ async function handleAuth(
    return;
  }

-  const body = await readJsonBody(req);
-  if (body === "too-large") {
-    sendText(res, 413, "Payload too large");
+  const body = await readJsonWebhookBodyOrReject({
+    req,
+    res,
+    maxBytes: MAX_BODY_BYTES,
+    profile: "pre-auth",
+    emptyObjectOnEmpty: false,
+    invalidJsonMessage: TELEGRAM_MINIAPP_EXPIRED_MESSAGE,
+    invalidJsonStatusCode: 401,
+  });
+  if (!body.ok) {
    return;
  }
-  if (!body) {
+  const authBody = parseAuthBody(body.value);
+  if (!authBody) {
    sendText(res, 401, TELEGRAM_MINIAPP_EXPIRED_MESSAGE);
    return;
  }
-  const accountId = normalizeAccountId(body.accountId ?? DEFAULT_ACCOUNT_ID);
+  const accountId = normalizeAccountId(authBody.accountId ?? DEFAULT_ACCOUNT_ID);
  const cfg = currentConfig(api);
  const account = resolveTelegramAccount({ cfg, accountId });
  const validated = validateTelegramMiniAppInitData({
-    initData: body.initData,
+    initData: authBody.initData,
    botToken: account.token,
  });
  if (!validated) {
@@ -123,7 +133,7 @@ async function handleAuth(
  }
  if (
    !launchTickets.consume({
-      ticket: body.launchTicket,
+      ticket: authBody.launchTicket,
      accountId,
      userId: validated.userId,
    })
@@ -153,36 +163,20 @@ function currentConfig(api: OpenClawPluginApi): OpenClawConfig {
  return (api.runtime.config?.current?.() ?? api.config) as OpenClawConfig;
 }

-async function readJsonBody(
-  req: IncomingMessage,
-): Promise<{ initData: string; launchTicket: string; accountId?: string } | "too-large" | null> {
-  const chunks: Buffer[] = [];
-  let total = 0;
-  for await (const chunk of req) {
-    const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-    total += buffer.length;
-    if (total > MAX_BODY_BYTES) {
-      return "too-large";
-    }
-    chunks.push(buffer);
-  }
-  try {
-    const parsed = JSON.parse(Buffer.concat(chunks).toString("utf8")) as {
-      initData?: unknown;
-      launchTicket?: unknown;
-      accountId?: unknown;
-    };
-    if (typeof parsed.initData !== "string" || typeof parsed.launchTicket !== "string") {
-      return null;
-    }
-    return {
-      initData: parsed.initData,
-      launchTicket: parsed.launchTicket,
-      ...(typeof parsed.accountId === "string" ? { accountId: parsed.accountId } : {}),
-    };
-  } catch {
+function parseAuthBody(
+  value: unknown,
+): { initData: string; launchTicket: string; accountId?: string } | null {
+  if (!isRecord(value)) {
    return null;
  }
+  if (typeof value.initData !== "string" || typeof value.launchTicket !== "string") {
+    return null;
+  }
+  return {
+    initData: value.initData,
+    launchTicket: value.launchTicket,
+    ...(typeof value.accountId === "string" ? { accountId: value.accountId } : {}),
+  };
 }

 function consumeRateLimit(ip: string): boolean {
@@ -177,6 +177,11 @@ describe("http body limits", () => {
      headers: { "content-length": "9999" },
      maxBytes: 128,
    },
+    {
+      name: "declared unsafe-integer content-length remains oversized",
+      headers: { "content-length": "999999999999999999999999" },
+      maxBytes: 128,
+    },
  ])("$name", async ({ chunks, headers, maxBytes }) => {
    await expectReadPayloadTooLarge({ chunks, headers, maxBytes });
  });
@@ -289,5 +294,40 @@ describe("http body limits", () => {
      message: "RequestBodyConnectionClosed",
      statusCode: 400,
    });
+    for (const event of ["data", "end", "error", "close"] as const) {
+      expect(req.listenerCount(event), event).toBe(0);
+    }
+  });
+
+  it("classifies request stream errors as a closed connection", async () => {
+    const req = createMockRequest({ emitEnd: false });
+    const promise = readJsonBodyWithLimit(req, { maxBytes: 128 });
+    queueMicrotask(() => req.emit("error", new Error("socket reset")));
+    await expect(promise).resolves.toEqual({
+      ok: false,
+      code: "CONNECTION_CLOSED",
+      error: "Connection closed",
+    });
+  });
+
+  it("can defer destructive limit cleanup until a response flushes", async () => {
+    const req = createMockRequest({
+      headers: { "content-length": "129" },
+      emitEnd: false,
+    });
+    const pause = vi.fn();
+    req.pause = pause;
+
+    await expectRequestBodyLimitError(
+      readRequestBodyWithLimit(req, { maxBytes: 128, destroyOnLimit: false }),
+      {
+        code: "PAYLOAD_TOO_LARGE",
+        message: "PayloadTooLarge",
+        statusCode: 413,
+      },
+    );
+
+    expect(req.destroyed).toBe(false);
+    expect(pause).toHaveBeenCalledOnce();
  });
 });
@@ -86,16 +86,18 @@ function parseContentLengthHeader(req: IncomingMessage): number | null {
    return null;
  }
  const parsed = parseStrictNonNegativeInteger(raw);
-  if (parsed === undefined) {
-    return null;
+  if (parsed !== undefined) {
+    return parsed;
  }
-  return parsed;
+  return /^\d+$/.test(raw.trim()) ? Number.MAX_SAFE_INTEGER : null;
 }

 export type ReadRequestBodyOptions = {
  maxBytes: number;
  timeoutMs?: number;
  encoding?: BufferEncoding;
+  /** Pause instead of destroying on size/timeout failures so a caller can flush a response first. */
+  destroyOnLimit?: boolean;
 };

 type RequestBodyLimitValues = {
@@ -140,6 +142,19 @@ function advanceRequestBodyChunk(
  };
 }

+function stopRequestBodyAfterLimit(req: IncomingMessage, destroyOnLimit: boolean): void {
+  if (req.destroyed) {
+    return;
+  }
+  if (destroyOnLimit) {
+    // Limit violations are expected user input; destroying with an Error causes
+    // an async 'error' event which can crash the process if no listener remains.
+    req.destroy();
+    return;
+  }
+  req.pause();
+}
+
 type ReadResponsePrefixResult = {
  buffer: Buffer;
  size: number;
@@ -344,15 +359,12 @@ export async function readRequestBodyWithLimit(
 ): Promise<string> {
  const { maxBytes, timeoutMs } = resolveRequestBodyLimitValues(options);
  const encoding = options.encoding ?? "utf-8";
+  const destroyOnLimit = options.destroyOnLimit !== false;

  const declaredLength = parseContentLengthHeader(req);
  if (declaredLength !== null && declaredLength > maxBytes) {
    const error = new RequestBodyLimitError({ code: "PAYLOAD_TOO_LARGE" });
-    if (!req.destroyed) {
-      // Limit violations are expected user input; destroying with an Error causes
-      // an async 'error' event which can crash the process if no listener remains.
-      req.destroy();
-    }
+    stopRequestBodyAfterLimit(req, destroyOnLimit);
    throw error;
  }

@@ -385,9 +397,7 @@ export async function readRequestBodyWithLimit(

    const timer = setNodeTimeout(() => {
      const error = new RequestBodyLimitError({ code: "REQUEST_BODY_TIMEOUT" });
-      if (!req.destroyed) {
-        req.destroy();
-      }
+      stopRequestBodyAfterLimit(req, destroyOnLimit);
      fail(error);
    }, timeoutMs);

@@ -399,9 +409,7 @@ export async function readRequestBodyWithLimit(
      totalBytes = progress.totalBytes;
      if (progress.exceeded) {
        const error = new RequestBodyLimitError({ code: "PAYLOAD_TOO_LARGE" });
-        if (!req.destroyed) {
-          req.destroy();
-        }
+        stopRequestBodyAfterLimit(req, destroyOnLimit);
        fail(error);
        return;
      }
@@ -470,8 +478,8 @@ export async function readJsonBodyWithLimit(
    }
    return {
      ok: false,
-      code: "INVALID_JSON",
-      error: formatErrorMessage(error),
+      code: "CONNECTION_CLOSED",
+      error: requestBodyErrorToText("CONNECTION_CLOSED"),
    };
  }
 }
@@ -19,6 +19,7 @@ import {
 type MockIncomingMessage = IncomingMessage & {
  destroyed?: boolean;
  destroy: () => MockIncomingMessage;
+  pause: () => MockIncomingMessage;
 };

 function createMockRequest(params: {
@@ -35,6 +36,7 @@ function createMockRequest(params: {
    req.destroyed = true;
    return req;
  }) as MockIncomingMessage["destroy"];
+  req.pause = (() => req) as MockIncomingMessage["pause"];

  if (params.chunks) {
    void Promise.resolve().then(() => {
@@ -36,6 +36,11 @@ export const WEBHOOK_BODY_READ_DEFAULTS = Object.freeze({
    maxBytes: 1024 * 1024,
    timeoutMs: 30_000,
  },
+  postAuthResponseFirst: {
+    maxBytes: 1024 * 1024,
+    timeoutMs: 30_000,
+    destroyOnLimit: false,
+  },
 });

 /** Default in-flight concurrency limits for webhook request pipelines. */
@@ -79,17 +84,21 @@ function resolveWebhookBodyReadLimits(params: {
 }

 function respondWebhookBodyReadError(params: {
+  req: IncomingMessage;
  res: ServerResponse;
  code: string;
  invalidMessage?: string;
+  invalidStatusCode?: number;
 }): { ok: false } {
-  const { res, code, invalidMessage } = params;
+  const { req, res, code, invalidMessage, invalidStatusCode } = params;
  if (code === "PAYLOAD_TOO_LARGE") {
+    closeRequestAfterResponse(req, res);
    res.statusCode = 413;
    res.end(requestBodyErrorToText("PAYLOAD_TOO_LARGE"));
    return { ok: false };
  }
  if (code === "REQUEST_BODY_TIMEOUT") {
+    closeRequestAfterResponse(req, res);
    res.statusCode = 408;
    res.end(requestBodyErrorToText("REQUEST_BODY_TIMEOUT"));
    return { ok: false };
@@ -99,11 +108,24 @@ function respondWebhookBodyReadError(params: {
    res.end(requestBodyErrorToText("CONNECTION_CLOSED"));
    return { ok: false };
  }
-  res.statusCode = 400;
+  res.statusCode = invalidStatusCode ?? 400;
  res.end(invalidMessage ?? "Bad Request");
  return { ok: false };
 }

+function closeRequestAfterResponse(req: IncomingMessage, res: ServerResponse): void {
+  const once = Reflect.get(res, "once");
+  if (typeof once !== "function") {
+    return;
+  }
+  res.setHeader("Connection", "close");
+  once.call(res, "finish", () => {
+    if (!req.destroyed) {
+      req.destroy();
+    }
+  });
+}
+
 /** Create an in-memory limiter that caps concurrent webhook handlers per key. */
 export function createWebhookInFlightLimiter(options?: {
  /** Maximum concurrent handlers allowed for one key. */
@@ -321,17 +343,22 @@ export async function readWebhookBodyOrReject(params: {
  });

  try {
-    const raw = await readRequestBodyWithLimit(params.req, limits);
+    const raw = await readRequestBodyWithLimit(params.req, {
+      ...limits,
+      destroyOnLimit: false,
+    });
    return { ok: true, value: raw };
  } catch (error) {
    if (isRequestBodyLimitError(error)) {
      return respondWebhookBodyReadError({
+        req: params.req,
        res: params.res,
        code: error.code,
        invalidMessage: params.invalidBodyMessage,
      });
    }
    return respondWebhookBodyReadError({
+      req: params.req,
      res: params.res,
      code: "INVALID_BODY",
      invalidMessage: params.invalidBodyMessage ?? formatErrorMessage(error),
@@ -355,6 +382,8 @@ export async function readJsonWebhookBodyOrReject(params: {
  emptyObjectOnEmpty?: boolean;
  /** Response body for malformed JSON. */
  invalidJsonMessage?: string;
+  /** Response status for malformed JSON. */
+  invalidJsonStatusCode?: number;
 }): Promise<{ ok: true; value: unknown } | { ok: false }> {
  const limits = resolveWebhookBodyReadLimits({
    maxBytes: params.maxBytes,
@@ -365,13 +394,16 @@ export async function readJsonWebhookBodyOrReject(params: {
    maxBytes: limits.maxBytes,
    timeoutMs: limits.timeoutMs,
    emptyObjectOnEmpty: params.emptyObjectOnEmpty,
+    destroyOnLimit: false,
  });
  if (body.ok) {
    return { ok: true, value: body.value };
  }
  return respondWebhookBodyReadError({
+    req: params.req,
    res: params.res,
    code: body.code,
    invalidMessage: params.invalidJsonMessage,
+    invalidStatusCode: params.invalidJsonStatusCode,
  });
 }