mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix(admin-http-rpc): time out incomplete request bodies (#104564)
* fix(admin-http-rpc): time out incomplete request bodies * fix(admin-http-rpc): preserve timeout responses
This commit is contained in:
qingminlong
@@ -1,4 +1,6 @@
|
||||
// Admin Http Rpc tests cover handler plugin behavior.
|
||||
import { createServer, type Server } from "node:http";
|
||||
import { connect, type Socket } from "node:net";
|
||||
import { Readable } from "node:stream";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { handleAdminHttpRpcRequest } from "./handler.js";
|
||||
@@ -30,6 +32,22 @@ function createRequest(body: unknown, method = "POST") {
|
||||
return req as import("node:http").IncomingMessage;
|
||||
}
|
||||
|
||||
function createHangingRequest() {
|
||||
const req = new Readable({
|
||||
read() {
|
||||
// Keep the body open so the handler's request-body timeout owns settlement.
|
||||
},
|
||||
});
|
||||
Object.assign(req, {
|
||||
method: "POST",
|
||||
url: "/api/v1/admin/rpc",
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
},
|
||||
});
|
||||
return req as import("node:http").IncomingMessage;
|
||||
}
|
||||
|
||||
function createResponse() {
|
||||
const captured: CapturedResponse = {
|
||||
statusCode: 200,
|
||||
@@ -54,8 +72,12 @@ function createResponse() {
|
||||
}
|
||||
|
||||
async function invoke(body: unknown, method = "POST") {
|
||||
return invokeRequest(createRequest(body, method));
|
||||
}
|
||||
|
||||
async function invokeRequest(req: import("node:http").IncomingMessage) {
|
||||
const { res, captured } = createResponse();
|
||||
const handled = await handleAdminHttpRpcRequest(createRequest(body, method), res);
|
||||
const handled = await handleAdminHttpRpcRequest(req, res);
|
||||
return {
|
||||
handled,
|
||||
captured,
|
||||
@@ -63,6 +85,35 @@ async function invoke(body: unknown, method = "POST") {
|
||||
};
|
||||
}
|
||||
|
||||
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);
|
||||
});
|
||||
}
|
||||
|
||||
describe("admin-http-rpc plugin handler", () => {
|
||||
beforeEach(() => {
|
||||
dispatchGatewayMethod.mockReset();
|
||||
@@ -206,4 +257,81 @@ describe("admin-http-rpc plugin handler", () => {
|
||||
expect(result.captured.headers.allow).toBe("POST");
|
||||
expect(dispatchGatewayMethod).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("times out incomplete request bodies before dispatch", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const resultPromise = invokeRequest(createHangingRequest());
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
const result = await resultPromise;
|
||||
|
||||
expect(result.handled).toBe(true);
|
||||
expect(result.captured.statusCode).toBe(408);
|
||||
expect(result.json).toEqual({
|
||||
ok: false,
|
||||
error: {
|
||||
type: "invalid_request",
|
||||
message: "Request body timeout",
|
||||
},
|
||||
});
|
||||
expect(dispatchGatewayMethod).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("delivers a real HTTP 408 response before closing timed-out partial bodies", async () => {
|
||||
vi.useFakeTimers();
|
||||
let markRequestStarted: (() => void) | undefined;
|
||||
const requestStarted = new Promise<void>((resolve) => {
|
||||
markRequestStarted = resolve;
|
||||
});
|
||||
const server = createServer((req, res) => {
|
||||
void handleAdminHttpRpcRequest(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 /api/v1/admin/rpc HTTP/1.1",
|
||||
"Host: 127.0.0.1",
|
||||
"Content-Type: application/json",
|
||||
"Content-Length: 64",
|
||||
"Connection: close",
|
||||
"",
|
||||
"{",
|
||||
].join("\r\n"),
|
||||
);
|
||||
|
||||
await requestStarted;
|
||||
const responsePromise = readSocketResponse(socket);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
const response = await responsePromise;
|
||||
const [, rawBody = ""] = response.split("\r\n\r\n", 2);
|
||||
|
||||
expect(response).toContain("HTTP/1.1 408");
|
||||
expect(response).toContain("Connection: close");
|
||||
expect(JSON.parse(rawBody) as unknown).toEqual({
|
||||
ok: false,
|
||||
error: {
|
||||
type: "invalid_request",
|
||||
message: "Request body timeout",
|
||||
},
|
||||
});
|
||||
expect(dispatchGatewayMethod).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
socket?.destroy();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -6,10 +6,12 @@ import { randomUUID } from "node:crypto";
|
||||
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,
|
||||
WEBHOOK_BODY_READ_DEFAULTS,
|
||||
} from "openclaw/plugin-sdk/webhook-request-guards";
|
||||
import { isAdminHttpRpcAllowedMethod, listAdminHttpRpcAllowedMethods } from "./methods.js";
|
||||
|
||||
const DEFAULT_RPC_BODY_BYTES = 1024 * 1024;
|
||||
|
||||
const ErrorCodes = {
|
||||
AGENT_TIMEOUT: "AGENT_TIMEOUT",
|
||||
APPROVAL_NOT_FOUND: "APPROVAL_NOT_FOUND",
|
||||
@@ -43,6 +45,22 @@ type ParsedRequest = {
|
||||
params?: unknown;
|
||||
};
|
||||
|
||||
type RequestBodyLimitFailureCode =
|
||||
| "PAYLOAD_TOO_LARGE"
|
||||
| "REQUEST_BODY_TIMEOUT"
|
||||
| "CONNECTION_CLOSED";
|
||||
type ReadJsonBodyResult =
|
||||
| { ok: true; value: unknown }
|
||||
| {
|
||||
ok: false;
|
||||
status: number;
|
||||
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 };
|
||||
}
|
||||
@@ -82,33 +100,145 @@ function sendError(res: ServerResponse, status: number, error: { type: string; m
|
||||
async function readJsonBody(
|
||||
req: IncomingMessage,
|
||||
maxBytes: number,
|
||||
): Promise<{ ok: true; value: unknown } | { ok: false; status: number; message: string }> {
|
||||
const chunks: Buffer[] = [];
|
||||
let totalBytes = 0;
|
||||
try {
|
||||
for await (const chunk of req) {
|
||||
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
totalBytes += buffer.byteLength;
|
||||
if (totalBytes > maxBytes) {
|
||||
return { ok: false, status: 413, message: "Payload too large" };
|
||||
}
|
||||
chunks.push(buffer);
|
||||
}
|
||||
} catch {
|
||||
return { ok: false, status: 400, message: "failed to read request body" };
|
||||
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 = Buffer.concat(chunks).toString("utf8");
|
||||
if (!raw.trim()) {
|
||||
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) };
|
||||
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":
|
||||
return 413;
|
||||
case "REQUEST_BODY_TIMEOUT":
|
||||
return 408;
|
||||
case "CONNECTION_CLOSED":
|
||||
return 400;
|
||||
}
|
||||
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);
|
||||
});
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
const value = Number(raw);
|
||||
return Number.isSafeInteger(value) ? value : Number.MAX_SAFE_INTEGER;
|
||||
}
|
||||
|
||||
function closeRequestAfterResponse(req: IncomingMessage, res: ServerResponse): void {
|
||||
const once = (res as { once?: ServerResponse["once"] }).once;
|
||||
if (typeof once !== "function") {
|
||||
return;
|
||||
}
|
||||
res.setHeader("Connection", "close");
|
||||
once.call(res, "finish", () => {
|
||||
// Timeout/size failures must flush JSON first; destroying before finish drops
|
||||
// the HTTP response on real partial-body sockets.
|
||||
if (!req.destroyed) {
|
||||
req.destroy();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function readRpcRequestBody(body: unknown):
|
||||
| { ok: true; request: ParsedRequest }
|
||||
| {
|
||||
@@ -202,8 +332,15 @@ export async function handleAdminHttpRpcRequest(
|
||||
return true;
|
||||
}
|
||||
|
||||
const body = await readJsonBody(req, DEFAULT_RPC_BODY_BYTES);
|
||||
const body = await readJsonBody(
|
||||
req,
|
||||
WEBHOOK_BODY_READ_DEFAULTS.postAuth.maxBytes,
|
||||
WEBHOOK_BODY_READ_DEFAULTS.postAuth.timeoutMs,
|
||||
);
|
||||
if (!body.ok) {
|
||||
if (body.closeAfterResponse) {
|
||||
closeRequestAfterResponse(req, res);
|
||||
}
|
||||
sendError(res, body.status, {
|
||||
type: "invalid_request",
|
||||
message: body.message,
|
||||
|
||||
Reference in New Issue
Block a user