ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-28 05:16:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(qa-channel): bound bus JSON response reads (#99169)

Browse Source 
Co-authored-by: NIO <nocodet@mail.com>
This commit is contained in:
NIO
2026-07-07 01:02:42 +08:00
committed by GitHub
parent 2e967ea61d
commit db334b3cf7
2 changed files with 157 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/qa-channel/src/bus-client.test.ts
+118 -2
View File
@@ -1,8 +1,10 @@
// Qa Channel tests cover bus client plugin behavior.
import { createServer } from "node:http";
import { createServer, type Server } from "node:http";
import { afterEach, describe, expect, it } from "vitest";
import { buildQaTarget, getQaBusState, parseQaTarget, pollQaBus } from "./bus-client.js";
const OVERSIZED_RESPONSE_BYTES = 18 * 1024 * 1024;
async function startJsonServer(
handler: (req: { url?: string | undefined }) => { statusCode?: number; body: string },
) {
@@ -34,6 +36,83 @@ async function startJsonServer(
};
}
async function listenLoopbackServer(server: Server): Promise<number> {
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => {
server.off("error", reject);
resolve();
});
});
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("test server failed to bind");
}
return address.port;
}
function createOversizedJsonServer(pathname: string): { server: Server; closed: Promise<number> } {
let resolveClosed: (sentBytes: number) => void = () => {};
const closed = new Promise<number>((resolve) => {
resolveClosed = resolve;
});
const server = createServer((req, res) => {
if (req.url !== pathname) {
res.writeHead(404, { "content-type": "application/json; charset=utf-8" });
res.end(JSON.stringify({ error: `unexpected path: ${req.url}` }));
return;
}
let sentBytes = 0;
let stopped = false;
let prefixSent = false;
const prefixChunk = Buffer.from('{"payload":"');
const bodyChunk = Buffer.alloc(64 * 1024, 0x61);
const suffixChunk = Buffer.from('"}');
const writeBuffer = (buffer: Buffer) => {
sentBytes += buffer.length;
if (!res.write(buffer)) {
res.once("drain", writeChunks);
return false;
}
return true;
};
const writeChunks = () => {
if (!prefixSent) {
prefixSent = true;
if (!writeBuffer(prefixChunk)) {
return;
}
}
while (true) {
if (stopped) {
return;
}
if (sentBytes + bodyChunk.length + suffixChunk.length >= OVERSIZED_RESPONSE_BYTES) {
break;
}
if (!writeBuffer(bodyChunk)) {
return;
}
}
if (!stopped) {
sentBytes += suffixChunk.length;
res.end(suffixChunk);
}
};
res.writeHead(200, { "content-type": "application/json; charset=utf-8", connection: "close" });
res.on("close", () => {
stopped = true;
resolveClosed(sentBytes);
});
req.on("aborted", () => {
stopped = true;
res.destroy();
});
writeChunks();
});
return { server, closed };
}
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> {
let timer: NodeJS.Timeout | undefined;
try {
@@ -83,7 +162,29 @@ describe("qa-bus client", () => {
cursor: 0,
timeoutMs: 0,
}),
).rejects.toThrow(SyntaxError);
).rejects.toThrow("qa-bus /v1/poll: malformed JSON response");
});
it("bounds oversized poll responses and closes the stream early", async () => {
const oversized = createOversizedJsonServer("/v1/poll");
const port = await listenLoopbackServer(oversized.server);
stops.push(async () => {
oversized.server.closeAllConnections?.();
await new Promise<void>((resolve, reject) => {
oversized.server.close((error) => (error ? reject(error) : resolve()));
});
});
await expect(
pollQaBus({
baseUrl: `http://127.0.0.1:${port}`,
accountId: "acct-a",
cursor: 0,
timeoutMs: 0,
}),
).rejects.toThrow("qa-bus /v1/poll: JSON response exceeds 16777216 bytes");
const sentBytes = await oversized.closed;
expect(sentBytes).toBeLessThan(OVERSIZED_RESPONSE_BYTES);
});
it("rejects immediately when a poll request is aborted", async () => {
@@ -151,4 +252,19 @@ describe("qa-bus client", () => {
events: [],
});
});
it("bounds oversized qa-bus state responses", async () => {
const oversized = createOversizedJsonServer("/v1/state");
const port = await listenLoopbackServer(oversized.server);
stops.push(async () => {
oversized.server.closeAllConnections?.();
await new Promise<void>((resolve, reject) => {
oversized.server.close((error) => (error ? reject(error) : resolve()));
});
});
await expect(getQaBusState(`http://127.0.0.1:${port}`)).rejects.toThrow(
"qa-channel.bus-state: JSON response exceeds 16777216 bytes",
);
});
});
extensions/qa-channel/src/bus-client.ts
+39 -21
View File
@@ -1,6 +1,8 @@
// Qa Channel plugin module implements bus client behavior.
import http from "node:http";
import https from "node:https";
import { readProviderJsonResponse } from "openclaw/plugin-sdk/provider-http";
import { readByteStreamWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime";
import type {
QaBusInboundMessageInput,
@@ -35,12 +37,32 @@ export type {
} from "./protocol.js";
type JsonResult<T> = Promise<T>;
const QA_BUS_JSON_RESPONSE_MAX_BYTES = 16 * 1024 * 1024;
function buildQaBusUrl(baseUrl: string, path: string): URL {
const normalizedBaseUrl = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/`;
return new URL(path.replace(/^\/+/, ""), normalizedBaseUrl);
}
async function readQaBusNodeJsonResponse<T>(
response: http.IncomingMessage,
label: string,
): Promise<T> {
const bytes = await readByteStreamWithLimit(response, {
maxBytes: QA_BUS_JSON_RESPONSE_MAX_BYTES,
onOverflow: ({ maxBytes }) => new Error(`${label}: JSON response exceeds ${maxBytes} bytes`),
});
const text = bytes.toString("utf8");
if (!text) {
return {} as T;
}
try {
return JSON.parse(text) as T;
} catch (cause) {
throw new Error(`${label}: malformed JSON response`, { cause });
}
}
async function postJson<T>(
baseUrl: string,
path: string,
@@ -70,27 +92,23 @@ async function postJson<T>(
},
},
(response) => {
const chunks: Buffer[] = [];
response.on("data", (chunk) => {
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
});
response.on("end", () => {
const text = Buffer.concat(chunks).toString("utf8");
let parsed: T | { error?: string };
try {
parsed = text ? (JSON.parse(text) as T | { error?: string }) : ({} as T);
} catch (error) {
const label = `qa-bus ${path}`;
void readQaBusNodeJsonResponse<T | { error?: string }>(response, label).then(
(parsed) => {
if ((response.statusCode ?? 500) < 200 || (response.statusCode ?? 500) >= 300) {
const error =
typeof parsed === "object" && parsed && "error" in parsed
? parsed.error
: undefined;
reject(new Error(error || `qa-bus request failed: ${response.statusCode ?? 500}`));
return;
}
resolve(parsed as T);
},
(error: unknown) => {
reject(toLintErrorObject(error, "Non-Error rejection"));
return;
}
if ((response.statusCode ?? 500) < 200 || (response.statusCode ?? 500) >= 300) {
const error =
typeof parsed === "object" && parsed && "error" in parsed ? parsed.error : undefined;
reject(new Error(error || `qa-bus request failed: ${response.statusCode ?? 500}`));
return;
}
resolve(parsed as T);
});
},
);
response.on("error", reject);
},
);
@@ -291,7 +309,7 @@ export async function getQaBusState(baseUrl: string): Promise<QaBusStateSnapshot
if (!response.ok) {
throw new Error(`qa-bus request failed: ${response.status}`);
}
return (await response.json()) as QaBusStateSnapshot;
return await readProviderJsonResponse<QaBusStateSnapshot>(response, "qa-channel.bus-state");
} finally {
await release();
}