diff --git a/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.test.ts b/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.test.ts index 01f24a8a8be2..8561398cd741 100644 --- a/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.test.ts +++ b/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.test.ts @@ -1,4 +1,5 @@ // Qa Lab tests cover credential lease plugin behavior. +import { createServer } from "node:http"; import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime"; import { afterEach, describe, expect, it, vi } from "vitest"; import { @@ -39,6 +40,74 @@ function fetchInit(fetchImpl: FetchMock, index = 0): RequestInit { return init; } +async function startStreamingFailureBroker(params: { + chunkBytes?: number; + intervalMs?: number; + totalBytes?: number; +}) { + const chunkBytes = params.chunkBytes ?? 64 * 1024; + const intervalMs = params.intervalMs ?? 1; + const totalBytes = params.totalBytes ?? 4 * 1024 * 1024; + let bytesWritten = 0; + let requestCount = 0; + let resolveClose: () => void = () => {}; + const closePromise = new Promise((resolve) => { + resolveClose = resolve; + }); + + const server = createServer((_req, res) => { + requestCount += 1; + res.writeHead(500, { "content-type": "text/plain" }); + const interval = setInterval(() => { + if (bytesWritten >= totalBytes || res.destroyed) { + clearInterval(interval); + if (!res.destroyed) { + res.end(); + } + return; + } + const nextBytes = Math.min(chunkBytes, totalBytes - bytesWritten); + bytesWritten += nextBytes; + res.write("x".repeat(nextBytes)); + }, intervalMs); + res.on("close", () => { + clearInterval(interval); + resolveClose(); + }); + }); + + await new Promise((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("expected streaming broker address"); + } + return { + closePromise, + getBytesWritten: () => bytesWritten, + getRequestCount: () => requestCount, + totalBytes, + url: `http://127.0.0.1:${address.port}`, + stop: async () => { + await new Promise((resolve, reject) => { + server.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); + }, + }; +} + describe("credential lease runtime", () => { afterEach(() => { vi.restoreAllMocks(); @@ -107,6 +176,60 @@ describe("credential lease runtime", () => { expect(headers.authorization).toBe("Bearer maintainer-secret"); }); + it("bounds oversized convex broker failure bodies before parsing", async () => { + const fetchImpl = vi.fn().mockResolvedValueOnce( + new Response("x".repeat(1_048_577), { + status: 500, + headers: { "content-type": "text/plain" }, + }), + ); + + await expect( + acquireQaCredentialLease({ + kind: "telegram", + source: "convex", + role: "maintainer", + env: { + OPENCLAW_QA_CONVEX_SITE_URL: "https://qa-cred.example.convex.site", + OPENCLAW_QA_CONVEX_SECRET_MAINTAINER: "maintainer-secret", + }, + fetchImpl, + resolveEnvPayload: () => ({ groupId: "-1", driverToken: "unused", sutToken: "unused" }), + parsePayload: (payload) => + payload as { groupId: string; driverToken: string; sutToken: string }, + }), + ).rejects.toThrow("Convex credential broker: text response exceeds 1048576 bytes"); + + expect(fetchImpl).toHaveBeenCalledTimes(1); + }); + + it("cancels a streaming convex broker failure body after the response cap", async () => { + const broker = await startStreamingFailureBroker({}); + try { + await expect( + acquireQaCredentialLease({ + kind: "telegram", + source: "convex", + role: "maintainer", + env: { + OPENCLAW_QA_CONVEX_SITE_URL: broker.url, + OPENCLAW_QA_CONVEX_SECRET_MAINTAINER: "maintainer-secret", + OPENCLAW_QA_ALLOW_INSECURE_HTTP: "1", + }, + resolveEnvPayload: () => ({ groupId: "-1", driverToken: "unused", sutToken: "unused" }), + parsePayload: (payload) => + payload as { groupId: string; driverToken: string; sutToken: string }, + }), + ).rejects.toThrow("Convex credential broker: text response exceeds 1048576 bytes"); + + await broker.closePromise; + expect(broker.getRequestCount()).toBe(1); + expect(broker.getBytesWritten()).toBeLessThan(broker.totalBytes); + } finally { + await broker.stop(); + } + }); + it("hydrates chunked convex credential payloads after acquire", async () => { const serialized = JSON.stringify({ groupId: "-100123", diff --git a/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.ts b/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.ts index 90d70ac20d91..0c3a0b018b99 100644 --- a/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.ts +++ b/extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; +import { readProviderTextResponse } from "openclaw/plugin-sdk/provider-http"; import { z } from "zod"; import { isQaCredentialTruthyOptIn, @@ -19,6 +20,7 @@ const DEFAULT_HTTP_TIMEOUT_MS = 15_000; const DEFAULT_LEASE_TTL_MS = 20 * 60 * 1_000; const DEFAULT_CHUNKED_PAYLOAD_MAX_BYTES = 64 * 1024 * 1024; const DEFAULT_CHUNKED_PAYLOAD_MAX_CHUNKS = 4096; +const CONVEX_BROKER_RESPONSE_MAX_BYTES = 1 * 1024 * 1024; const CHUNKED_PAYLOAD_MAX_BYTES_ENV = "OPENCLAW_QA_CREDENTIAL_PAYLOAD_MAX_BYTES"; const CHUNKED_PAYLOAD_MAX_CHUNKS_ENV = "OPENCLAW_QA_CREDENTIAL_PAYLOAD_MAX_CHUNKS"; const RETRY_BACKOFF_MS = [500, 1_000, 2_000, 4_000, 5_000] as const; @@ -282,6 +284,7 @@ async function postConvexBroker(params: { authToken: string; body: Record; fetchImpl: typeof fetch; + maxBytes: number; timeoutMs: number; url: string; }): Promise { @@ -296,7 +299,11 @@ async function postConvexBroker(params: { signal: AbortSignal.timeout(timeoutMs), }); - const text = await response.text(); + // Keep ordinary broker responses small, while allowing chunk payloads to use + // the larger declared payload ceiling. + const text = await readProviderTextResponse(response, "Convex credential broker", { + maxBytes: params.maxBytes, + }); const payload: unknown = (() => { if (!text.trim()) { return undefined; @@ -341,6 +348,7 @@ async function resolveConvexCredentialPayload(params: { for (let index = 0; index < marker.chunkCount; index += 1) { const payload = await postConvexBroker({ fetchImpl: params.fetchImpl, + maxBytes: params.config.payloadMaxBytes, timeoutMs: params.config.httpTimeoutMs, authToken: params.config.authToken, url: params.config.payloadChunkUrl, @@ -437,6 +445,7 @@ export async function acquireQaCredentialLease( try { const payload = await postConvexBroker({ fetchImpl, + maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES, timeoutMs: config.httpTimeoutMs, authToken: config.authToken, url: config.acquireUrl, @@ -452,6 +461,7 @@ export async function acquireQaCredentialLease( const releaseLease = async () => { const releasePayload = await postConvexBroker({ fetchImpl, + maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES, timeoutMs: config.httpTimeoutMs, authToken: config.authToken, url: config.releaseUrl, @@ -503,6 +513,7 @@ export async function acquireQaCredentialLease( async heartbeat() { const heartbeatPayload = await postConvexBroker({ fetchImpl, + maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES, timeoutMs: config.httpTimeoutMs, authToken: config.authToken, url: config.heartbeatUrl,