diff --git a/scripts/e2e/kitchen-sink-rpc-walk.mjs b/scripts/e2e/kitchen-sink-rpc-walk.mjs index 71ea85442962..1203cb8a1955 100644 --- a/scripts/e2e/kitchen-sink-rpc-walk.mjs +++ b/scripts/e2e/kitchen-sink-rpc-walk.mjs @@ -28,6 +28,10 @@ const INSTALL_TIMEOUT_MS = readPositiveInt( ); const RPC_TIMEOUT_MS = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_RPC_CALL_MS, 60000); const FETCH_TIMEOUT_MS = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_RPC_FETCH_MS, 10000); +const FETCH_BODY_MAX_BYTES = readPositiveInt( + process.env.OPENCLAW_KITCHEN_SINK_RPC_FETCH_BODY_BYTES, + 1024 * 1024, +); const MAX_RSS_MIB = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_MAX_RSS_MIB, 2048); const GATEWAY_TEARDOWN_GRACE_MS = 10000; const GATEWAY_TEARDOWN_KILL_GRACE_MS = 2000; @@ -461,6 +465,7 @@ function isRetryableTransientNetworkError(error, seen = new Set()) { export async function fetchJson(url, options = {}) { const attempts = Math.max(1, options.attempts ?? 3); const timeoutMs = Math.max(1, options.timeoutMs ?? FETCH_TIMEOUT_MS); + const maxBodyBytes = Math.max(1, options.maxBodyBytes ?? FETCH_BODY_MAX_BYTES); let lastError; for (let attempt = 1; attempt <= attempts; attempt += 1) { const controller = new AbortController(); @@ -480,7 +485,10 @@ export async function fetchJson(url, options = {}) { (options.fetchImpl ?? fetch)(url, { signal: controller.signal }), timeoutPromise, ]); - const text = await Promise.race([response.text(), timeoutPromise]); + const text = await Promise.race([ + readBoundedResponseText(response, maxBodyBytes), + timeoutPromise, + ]); let body = null; try { body = text ? JSON.parse(text) : null; @@ -503,6 +511,31 @@ export async function fetchJson(url, options = {}) { throw lastError ?? new Error(`fetch ${url} failed`); } +export async function readBoundedResponseText(response, byteLimit = FETCH_BODY_MAX_BYTES) { + const reader = response.body?.getReader?.(); + if (!reader) { + return await response.text(); + } + const chunks = []; + let totalBytes = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) { + break; + } + const chunk = Buffer.from(value); + totalBytes += chunk.byteLength; + if (totalBytes > byteLimit) { + await reader.cancel().catch(() => undefined); + throw Object.assign(new Error(`fetch response body exceeded ${byteLimit} bytes`), { + code: "ETOOBIG", + }); + } + chunks.push(chunk); + } + return Buffer.concat(chunks, totalBytes).toString("utf8"); +} + function configureKitchenSink(env, port) { const configPath = env.OPENCLAW_CONFIG_PATH; const config = fs.existsSync(configPath) ? readJson(configPath) : {}; diff --git a/test/scripts/kitchen-sink-rpc-walk.test.ts b/test/scripts/kitchen-sink-rpc-walk.test.ts index 1be77e0a2ebc..f8765a672069 100644 --- a/test/scripts/kitchen-sink-rpc-walk.test.ts +++ b/test/scripts/kitchen-sink-rpc-walk.test.ts @@ -24,6 +24,7 @@ import { hasChildExited, makeEnv, readPositiveInt, + readBoundedResponseText, runCommand, sampleProcess, sampleWindowsProcessByPort, @@ -620,6 +621,29 @@ describe("kitchen-sink RPC process sampling", () => { expect(fetchImpl).toHaveBeenCalledTimes(2); }); + it("bounds HTTP probe response bodies", async () => { + const fetchImpl = vi + .fn() + .mockResolvedValue(new Response("x".repeat(1025), { status: 200 })); + + await expect( + fetchJson("http://127.0.0.1:19680/healthz", { + attempts: 1, + fetchImpl, + maxBodyBytes: 1024, + }), + ).rejects.toMatchObject({ + code: "ETOOBIG", + message: "fetch response body exceeded 1024 bytes", + }); + }); + + it("reads bounded response streams", async () => { + await expect( + readBoundedResponseText(new Response('{"status":"live"}'), 1024), + ).resolves.toBe('{"status":"live"}'); + }); + it("times out stalled HTTP probe response bodies", async () => { vi.useFakeTimers(); const fetchImpl = vi.fn().mockResolvedValue({