furyhawk/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-21 10:16:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(signal): add timeouts to container REST requests (#104122)

Browse Source 
* fix(signal): add timeouts to container REST requests

* fix(signal): use idle timeouts for container response bodies

* fix(signal): remove redundant timeout release

Signed-off-by: sallyom <somalley@redhat.com>

---------

Signed-off-by: sallyom <somalley@redhat.com>
Co-authored-by: sallyom <somalley@redhat.com>
This commit is contained in:
Alix-007
2026-07-12 16:59:14 -04:00
committed by GitHub
co-authored by sallyom
parent 238c7973ad
commit 7c72fef247
2 changed files with 200 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/signal/src/client-container.test.ts
+149 -3
View File
@@ -18,9 +18,8 @@ import {
// spyOn approach works with vitest forks pool for cross-directory imports
const mockFetch = vi.fn();
// Build a Response-like `body` stream from a string so the production code exercises the
// bounded body readers (readProviderTextResponse / readResponseTextLimited) instead of an
// unbounded res.text(). Kept local to this file to avoid touching shared HTTP test mocks.
// Build Response-like `body` streams so production code exercises bounded readers instead
// of unbounded res.text()/arrayBuffer(). Kept local to avoid touching shared HTTP mocks.
function bodyStream(text: string): { body: ReadableStream<Uint8Array> } {
const bytes = new TextEncoder().encode(text);
return {
@@ -34,6 +33,31 @@ function bodyStream(text: string): { body: ReadableStream<Uint8Array> } {
}),
};
}
function stalledBodyStream(): { body: ReadableStream<Uint8Array> } {
return {
body: new ReadableStream<Uint8Array>(),
};
}
function delayedBodyStream(
chunks: Array<{ delayMs: number; text: string }>,
closeDelayMs = 1,
): { body: ReadableStream<Uint8Array> } {
const encoder = new TextEncoder();
return {
body: new ReadableStream<Uint8Array>({
start(controller) {
let elapsedMs = 0;
for (const chunk of chunks) {
elapsedMs += chunk.delayMs;
setTimeout(() => controller.enqueue(encoder.encode(chunk.text)), elapsedMs);
}
setTimeout(() => controller.close(), elapsedMs + closeDelayMs);
},
}),
};
}
const wsMockState = vi.hoisted(() => ({
behavior: "close" as "close" | "open" | "error" | "unexpected-response",
urls: [] as string[],
@@ -331,6 +355,36 @@ describe("containerRestRequest", () => {
).rejects.toThrow(`Signal REST 500: ${"x".repeat(16 * 1024)}`);
});
it("times out stalled REST error bodies before reporting the HTTP failure", async () => {
vi.useFakeTimers();
try {
let observedSignal: AbortSignal | undefined;
mockFetch.mockImplementation(async (_url, init: RequestInit) => {
observedSignal = init.signal ?? undefined;
return new Response(stalledBodyStream().body, {
status: 500,
statusText: "Internal Server Error",
});
});
const request = containerRestRequest("/v2/send", {
baseUrl: "http://localhost:8080",
timeoutMs: 25,
});
await vi.advanceTimersByTimeAsync(0);
const requestRejection = expect(request).rejects.toThrow(
"Signal REST 500: Internal Server Error",
);
await vi.advanceTimersByTimeAsync(25);
await requestRejection;
expect(observedSignal?.aborted).toBe(false);
} finally {
vi.useRealTimers();
}
});
it("handles empty response body", async () => {
mockFetch.mockResolvedValue({
ok: true,
@@ -380,6 +434,68 @@ describe("containerRestRequest", () => {
timeoutSpy.mockRestore();
}
});
it("times out stalled REST response bodies without aborting completed fetches", async () => {
vi.useFakeTimers();
try {
let observedSignal: AbortSignal | undefined;
mockFetch.mockImplementation(async (_url, init: RequestInit) => {
observedSignal = init.signal ?? undefined;
return new Response(stalledBodyStream().body, {
status: 200,
headers: { "content-type": "application/json" },
});
});
const request = containerRestRequest("/v1/about", {
baseUrl: "http://localhost:8080",
timeoutMs: 25,
});
await vi.advanceTimersByTimeAsync(0);
expect(mockFetch).toHaveBeenCalledOnce();
expect(observedSignal?.aborted).toBe(false);
const requestRejection = expect(request).rejects.toThrow(
"Signal REST response body stalled after 25ms",
);
await vi.advanceTimersByTimeAsync(25);
await requestRejection;
expect(observedSignal?.aborted).toBe(false);
} finally {
vi.useRealTimers();
}
});
it("allows slow REST response bodies while chunks keep arriving before the idle timeout", async () => {
vi.useFakeTimers();
try {
mockFetch.mockResolvedValue(
new Response(
delayedBodyStream([
{ delayMs: 10, text: "{" },
{ delayMs: 20, text: '"ok"' },
{ delayMs: 20, text: ":true" },
{ delayMs: 20, text: "}" },
]).body,
{
status: 200,
headers: { "content-type": "application/json" },
},
),
);
const request = containerRestRequest<{ ok: boolean }>("/v1/about", {
baseUrl: "http://localhost:8080",
timeoutMs: 25,
});
await vi.advanceTimersByTimeAsync(75);
await expect(request).resolves.toEqual({ ok: true });
} finally {
vi.useRealTimers();
}
});
});
describe("containerSendMessage", () => {
@@ -963,6 +1079,36 @@ describe("containerFetchAttachment", () => {
}),
).rejects.toThrow("Signal REST attachment exceeded size limit");
});
it("times out stalled attachment bodies without aborting completed fetches", async () => {
vi.useFakeTimers();
try {
let observedSignal: AbortSignal | undefined;
mockFetch.mockImplementation(async (_url, init: RequestInit) => {
observedSignal = init.signal ?? undefined;
return new Response(stalledBodyStream().body, {
status: 200,
headers: new Headers(),
});
});
const request = containerFetchAttachment("attachment-123", {
baseUrl: "http://localhost:8080",
timeoutMs: 25,
});
await vi.advanceTimersByTimeAsync(0);
const requestRejection = expect(request).rejects.toThrow(
"Signal REST attachment response body stalled after 25ms",
);
await vi.advanceTimersByTimeAsync(25);
await requestRejection;
expect(observedSignal?.aborted).toBe(false);
} finally {
vi.useRealTimers();
}
});
});
describe("normalizeBaseUrl edge cases", () => {
extensions/signal/src/client-container.ts
+51 -10
View File
@@ -14,10 +14,9 @@ import {
resolveTimerTimeoutMs,
} from "openclaw/plugin-sdk/number-runtime";
import {
readProviderTextResponse,
readResponseTextLimited,
} from "openclaw/plugin-sdk/provider-http";
import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
readResponseTextPrefix,
readResponseWithLimit,
} from "openclaw/plugin-sdk/response-limit-runtime";
import { readRegularFile } from "openclaw/plugin-sdk/security-runtime";
import WebSocket from "ws";
@@ -55,6 +54,8 @@ type ContainerWebSocketMessage = {
const DEFAULT_TIMEOUT_MS = 10_000;
const DEFAULT_ATTACHMENT_RESPONSE_MAX_BYTES = 1_048_576;
const SIGNAL_REST_ERROR_RESPONSE_MAX_BYTES = 16 * 1024;
const SIGNAL_REST_SUCCESS_RESPONSE_MAX_BYTES = 16 * 1024 * 1024;
// Receive envelopes contain JSON metadata; attachment bytes are fetched separately.
// Keep the ws pre-buffer limit narrow so a container cannot force 100 MiB frames.
const SIGNAL_CONTAINER_WS_MAX_PAYLOAD_BYTES = 1024 * 1024;
@@ -113,12 +114,44 @@ function readContentLength(res: Response): number | undefined {
return parseMediaContentLength(res.headers?.get("content-length") ?? null) ?? undefined;
}
async function readCappedResponseBuffer(res: Response, maxResponseBytes: number): Promise<Buffer> {
function signalRestIdleTimeoutError({ chunkTimeoutMs }: { chunkTimeoutMs: number }): Error {
return new Error(`Signal REST response body stalled after ${chunkTimeoutMs}ms`);
}
function signalAttachmentIdleTimeoutError({ chunkTimeoutMs }: { chunkTimeoutMs: number }): Error {
return new Error(`Signal REST attachment response body stalled after ${chunkTimeoutMs}ms`);
}
async function readSignalRestText(res: Response, bodyIdleTimeoutMs: number): Promise<string> {
const bytes = await readResponseWithLimit(res, SIGNAL_REST_SUCCESS_RESPONSE_MAX_BYTES, {
chunkTimeoutMs: bodyIdleTimeoutMs,
onIdleTimeout: signalRestIdleTimeoutError,
onOverflow: ({ maxBytes }) => new Error(`Signal REST: text response exceeds ${maxBytes} bytes`),
});
return new TextDecoder().decode(bytes);
}
async function readSignalRestErrorText(res: Response, bodyIdleTimeoutMs: number): Promise<string> {
return (
await readResponseTextPrefix(res, SIGNAL_REST_ERROR_RESPONSE_MAX_BYTES, {
chunkTimeoutMs: bodyIdleTimeoutMs,
onIdleTimeout: signalRestIdleTimeoutError,
})
).text;
}
async function readCappedResponseBuffer(
res: Response,
maxResponseBytes: number,
bodyIdleTimeoutMs: number,
): Promise<Buffer> {
const contentLength = readContentLength(res);
if (contentLength !== undefined && contentLength > maxResponseBytes) {
throw new Error("Signal REST attachment exceeded size limit");
}
return await readResponseWithLimit(res, maxResponseBytes, {
chunkTimeoutMs: bodyIdleTimeoutMs,
onIdleTimeout: signalAttachmentIdleTimeoutError,
onOverflow: () => new Error("Signal REST attachment exceeded size limit"),
});
}
@@ -246,7 +279,9 @@ export async function containerRestRequest<T = unknown>(
init.body = JSON.stringify(body);
}
const res = await fetchWithTimeout(url, init, opts.timeoutMs ?? DEFAULT_TIMEOUT_MS);
const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS;
const bodyIdleTimeoutMs = resolveTimerTimeoutMs(timeoutMs, DEFAULT_TIMEOUT_MS);
const res = await fetchWithTimeout(url, init, timeoutMs);
if (res.status === 204) {
return undefined as T;
@@ -255,14 +290,14 @@ export async function containerRestRequest<T = unknown>(
if (!res.ok) {
// Bound the error body: signal-cli-rest-api is an untrusted external container,
// and a hostile/buggy response must not let an error path buffer an unbounded body.
const errorText = await readResponseTextLimited(res).catch(() => "");
const errorText = await readSignalRestErrorText(res, bodyIdleTimeoutMs).catch(() => "");
throw new Error(`Signal REST ${res.status}: ${errorText || res.statusText}`);
}
// Bound the success body under the shared 16 MiB provider cap before JSON.parse so a
// malicious/runaway container response cannot OOM the runtime (send/typing/version all
// funnel through here). Reuse the same bounded reader family as the attachment path.
const text = await readProviderTextResponse(res, "Signal REST");
const text = await readSignalRestText(res, bodyIdleTimeoutMs);
if (!text) {
return undefined as T;
}
@@ -286,13 +321,19 @@ export async function containerFetchAttachment(
let res: Response | undefined;
try {
res = await fetchWithTimeout(url, { method: "GET" }, opts.timeoutMs ?? DEFAULT_TIMEOUT_MS);
const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS;
const bodyIdleTimeoutMs = resolveTimerTimeoutMs(timeoutMs, DEFAULT_TIMEOUT_MS);
res = await fetchWithTimeout(url, { method: "GET" }, timeoutMs);
if (!res.ok) {
return null;
}
return await readCappedResponseBuffer(res, normalizeMaxResponseBytes(opts.maxResponseBytes));
return await readCappedResponseBuffer(
res,
normalizeMaxResponseBytes(opts.maxResponseBytes),
bodyIdleTimeoutMs,
);
} finally {
await releaseUnreadResponseBody(res);
}