From a817aec1cf65c7293aaeea0cb60d1a3b3cfd9037 Mon Sep 17 00:00:00 2001 From: Alix-007 Date: Mon, 29 Jun 2026 08:08:36 +0800 Subject: [PATCH] fix(inworld): bound TTS audio, voices, and error response reads to prevent OOM (#95416) * fix(inworld): bound TTS audio, voices, and error response reads to prevent OOM * fix(inworld): enforce decoded TTS audio cap --------- Co-authored-by: Vincent Koc Co-authored-by: Vincent Koc <25068+vincentkoc@users.noreply.github.com> (cherry picked from commit 4c8470c0699cbbb1ce6a7423928a41a465d1c05e) --- extensions/inworld/tts.test.ts | 137 +++++++++++++++++++++++++++++++++ extensions/inworld/tts.ts | 92 ++++++++++++++++++++-- 2 files changed, 224 insertions(+), 5 deletions(-) diff --git a/extensions/inworld/tts.test.ts b/extensions/inworld/tts.test.ts index a9b0f50ac03d..d03cc3106bd7 100644 --- a/extensions/inworld/tts.test.ts +++ b/extensions/inworld/tts.test.ts @@ -343,3 +343,140 @@ describe("inworldTTS", () => { expect(release).toHaveBeenCalledTimes(1); }); }); + +describe("Inworld response read bounding", () => { + const MiB = 1024 * 1024; + + // A never-ending stream that enqueues one fixed-size chunk per pull. An + // unbounded reader (the previous `await response.text()` / `response.json()`) + // would buffer this forever and OOM; the bounded reader must stop at the cap + // and cancel the stream. + function infiniteByteStream(chunkBytes: number): { + stream: ReadableStream; + state: { enqueued: number; cancelled: boolean }; + } { + const state = { enqueued: 0, cancelled: false }; + const chunk = new Uint8Array(chunkBytes).fill(0x61); // "a" + const stream = new ReadableStream({ + pull(controller) { + state.enqueued += 1; + controller.enqueue(chunk); + }, + cancel() { + state.cancelled = true; + }, + }); + return { stream, state }; + } + + it("fail-closed: rejects and cancels an oversized TTS audio stream instead of buffering it (32 MiB cap)", async () => { + const { stream, state } = infiniteByteStream(8 * MiB); + queueGuardedResponse(new Response(stream, { status: 200 })); + + await expect(inworldTTS({ text: "test", apiKey: "test-key" })).rejects.toThrow( + /Inworld TTS audio stream too large: \d+ bytes \(limit: 33554432 bytes\)/, + ); + // Enforced after a bounded number of 8 MiB chunks, never the full unbounded + // stream, and the stream is cancelled so the socket/buffers are released. + expect(state.enqueued).toBeLessThanOrEqual(8); + expect(state.cancelled).toBe(true); + }); + + it("happy-path: a normal multi-line NDJSON audio payload still decodes unchanged", async () => { + const part1 = Buffer.from("hello-").toString("base64"); + const part2 = Buffer.from("world").toString("base64"); + const body = [ + JSON.stringify({ result: { audioContent: part1 } }), + JSON.stringify({ result: { audioContent: part2 } }), + ].join("\n"); + queueGuardedResponse(new Response(body, { status: 200 })); + + const audio = await inworldTTS({ text: "test", apiKey: "test-key" }); + expect(audio.toString("utf8")).toBe("hello-world"); + }); + + it("edge: an under-cap ~1 MiB audio payload is read intact, not truncated", async () => { + const payload = "x".repeat(MiB); + const encoded = Buffer.from(payload).toString("base64"); + const body = JSON.stringify({ result: { audioContent: encoded } }); + queueGuardedResponse(new Response(body, { status: 200 })); + + const audio = await inworldTTS({ text: "test", apiKey: "test-key" }); + expect(audio.length).toBe(payload.length); + expect(audio.toString("utf8")).toBe(payload); + }); + + it("fail-closed: rejects decoded audio that exceeds the shared audio cap", async () => { + const decodedPayload = Buffer.alloc(16 * MiB + 1, 0x61); + const body = JSON.stringify({ + result: { audioContent: decodedPayload.toString("base64") }, + }); + queueGuardedResponse(new Response(body, { status: 200 })); + + await expect(inworldTTS({ text: "test", apiKey: "test-key" })).rejects.toThrow( + /Inworld TTS decoded audio too large: 16777217 bytes \(limit: 16777216 bytes\)/, + ); + }); + + it("regression: a malformed NDJSON line under the cap still throws a bounded parse error", async () => { + queueGuardedResponse(new Response("this-is-not-json", { status: 200 })); + await expect(inworldTTS({ text: "test", apiKey: "test-key" })).rejects.toThrow( + /Inworld TTS stream parse error/, + ); + }); + + it("fail-closed: truncates an oversized HTTP error body to a bounded marker", async () => { + queueGuardedResponse(new Response("E".repeat(64 * 1024), { status: 500 })); + + let captured: unknown; + await inworldTTS({ text: "test", apiKey: "test-key" }).catch((error: unknown) => { + captured = error; + }); + + expect(captured).toBeInstanceOf(Error); + const message = (captured as Error).message; + expect(message.startsWith("Inworld TTS API error (500): ")).toBe(true); + // Never the full 64 KiB hostile body: it collapses to a fixed marker. + expect(message).toContain("(error body exceeded diagnostic limit; truncated)"); + expect(message.length).toBeLessThan(512); + }); + + it("edge: a small error body is preserved verbatim in the thrown message", async () => { + queueGuardedResponse(new Response("invalid api key", { status: 401 })); + await expect(inworldTTS({ text: "test", apiKey: "test-key" })).rejects.toThrow( + "Inworld TTS API error (401): invalid api key", + ); + }); + + it("fail-closed: rejects and cancels an oversized voices JSON stream (16 MiB cap)", async () => { + const { stream, state } = infiniteByteStream(8 * MiB); + queueGuardedResponse(new Response(stream, { status: 200 })); + + await expect(listInworldVoices({ apiKey: "test-key" })).rejects.toThrow( + /Inworld voices response too large: \d+ bytes \(limit: 16777216 bytes\)/, + ); + expect(state.enqueued).toBeLessThanOrEqual(4); + expect(state.cancelled).toBe(true); + }); + + it("happy-path: a normal voices JSON list still parses unchanged", async () => { + queueGuardedResponse( + new Response( + JSON.stringify({ + voices: [{ voiceId: "Sarah", displayName: "Sarah", langCode: "en-US", tags: ["female"] }], + }), + { status: 200 }, + ), + ); + + const voices = await listInworldVoices({ apiKey: "test-key" }); + expect(voices).toEqual([ + { id: "Sarah", name: "Sarah", description: undefined, locale: "en-US", gender: "female" }, + ]); + }); + + it("regression: malformed voices JSON under the cap still throws", async () => { + queueGuardedResponse(new Response("{not-json", { status: 200 })); + await expect(listInworldVoices({ apiKey: "test-key" })).rejects.toThrow(); + }); +}); diff --git a/extensions/inworld/tts.ts b/extensions/inworld/tts.ts index a0da553e44fd..141f90281ebb 100644 --- a/extensions/inworld/tts.ts +++ b/extensions/inworld/tts.ts @@ -1,4 +1,6 @@ // Inworld plugin module implements tts behavior. +import { MAX_AUDIO_BYTES } from "openclaw/plugin-sdk/media-runtime"; +import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import type { SpeechVoiceOption } from "openclaw/plugin-sdk/speech-core"; import { fetchWithSsrFGuard, type SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime"; @@ -6,6 +8,60 @@ const DEFAULT_INWORLD_BASE_URL = "https://api.inworld.ai"; export const DEFAULT_INWORLD_VOICE_ID = "Sarah"; export const DEFAULT_INWORLD_MODEL_ID = "inworld-tts-1.5-max"; +// The streaming TTS endpoint returns newline-delimited JSON whose audio is +// base64-encoded, so the wire body is ~4/3 larger than the decoded audio plus a +// JSON envelope. Cap the read at double the shared 16 MiB audio limit so a +// full-size legitimate clip still fits, while bounding memory against an +// unbounded or hijacked SSE stream that would otherwise be buffered whole by the +// previous `await response.text()`. +const INWORLD_TTS_BODY_MAX_BYTES = MAX_AUDIO_BYTES * 2; +// The voices listing is a small JSON catalog, so the shared 16 MiB audio limit +// is already generous headroom while still closing the unbounded +// `await response.json()` read. +const INWORLD_VOICES_BODY_MAX_BYTES = MAX_AUDIO_BYTES; +// Abort the read if the upstream stalls mid-body so a hung stream cannot pin the +// socket and buffers open indefinitely. +const INWORLD_BODY_READ_IDLE_TIMEOUT_MS = 30_000; +// Error responses only need a short diagnostic snippet, never the whole body. +const INWORLD_ERROR_BODY_MAX_BYTES = 8 * 1024; +const INWORLD_ERROR_BODY_MAX_CHARS = 400; +const INWORLD_ERROR_BODY_READ_IDLE_TIMEOUT_MS = 10_000; + +// Sentinel so the error-snippet reader can tell a cap overflow apart from an +// unrelated read failure without leaking the (possibly hostile) body. +class InworldErrorBodyOverflow extends Error {} + +/** + * Reads a bounded, whitespace-collapsed diagnostic snippet from a non-OK + * response body. A misbehaving or hostile endpoint can stream an arbitrarily + * large error body, so this never buffers it whole: it reuses the shared + * `readResponseWithLimit` reader (which cancels the underlying stream on + * overflow and enforces an idle timeout) with a small cap. On overflow it + * returns a fixed marker instead of echoing attacker-controlled bytes into the + * thrown error. Kept local to this extension so it depends only on the + * already-exported `response-limit-runtime` entry and adds no shared plugin-SDK + * surface. + */ +async function readInworldErrorBodySnippet(response: Response): Promise { + let buffer: Buffer; + try { + buffer = await readResponseWithLimit(response, INWORLD_ERROR_BODY_MAX_BYTES, { + chunkTimeoutMs: INWORLD_ERROR_BODY_READ_IDLE_TIMEOUT_MS, + onOverflow: () => new InworldErrorBodyOverflow(), + }); + } catch (error) { + return error instanceof InworldErrorBodyOverflow + ? "(error body exceeded diagnostic limit; truncated)" + : ""; + } + + const collapsed = buffer.toString("utf8").replace(/\s+/g, " ").trim(); + if (collapsed.length > INWORLD_ERROR_BODY_MAX_CHARS) { + return `${collapsed.slice(0, INWORLD_ERROR_BODY_MAX_CHARS)}…`; + } + return collapsed; +} + export const INWORLD_TTS_MODELS = [ "inworld-tts-1.5-max", "inworld-tts-1.5-mini", @@ -90,12 +146,21 @@ export async function inworldTTS(params: { try { if (!response.ok) { - const errorBody = await response.text().catch(() => ""); + const errorBody = await readInworldErrorBodySnippet(response); throw new Error(`Inworld TTS API error (${response.status}): ${errorBody}`); } - const body = await response.text(); + const body = ( + await readResponseWithLimit(response, INWORLD_TTS_BODY_MAX_BYTES, { + chunkTimeoutMs: INWORLD_BODY_READ_IDLE_TIMEOUT_MS, + onOverflow: ({ size, maxBytes }) => + new Error(`Inworld TTS audio stream too large: ${size} bytes (limit: ${maxBytes} bytes)`), + onIdleTimeout: ({ chunkTimeoutMs }) => + new Error(`Inworld TTS audio stream stalled: no data received for ${chunkTimeoutMs}ms`), + }) + ).toString("utf8"); const chunks: Buffer[] = []; + let decodedAudioBytes = 0; for (const line of body.split("\n")) { const trimmed = line.trim(); @@ -120,7 +185,15 @@ export async function inworldTTS(params: { } if (parsed.result?.audioContent) { - chunks.push(Buffer.from(parsed.result.audioContent, "base64")); + const chunk = Buffer.from(parsed.result.audioContent, "base64"); + const nextDecodedAudioBytes = decodedAudioBytes + chunk.length; + if (nextDecodedAudioBytes > MAX_AUDIO_BYTES) { + throw new Error( + `Inworld TTS decoded audio too large: ${nextDecodedAudioBytes} bytes (limit: ${MAX_AUDIO_BYTES} bytes)`, + ); + } + decodedAudioBytes = nextDecodedAudioBytes; + chunks.push(chunk); } } @@ -159,11 +232,20 @@ export async function listInworldVoices(params: { try { if (!response.ok) { - const errorBody = await response.text().catch(() => ""); + const errorBody = await readInworldErrorBodySnippet(response); throw new Error(`Inworld voices API error (${response.status}): ${errorBody}`); } - const json = (await response.json()) as { + const voicesBody = ( + await readResponseWithLimit(response, INWORLD_VOICES_BODY_MAX_BYTES, { + chunkTimeoutMs: INWORLD_BODY_READ_IDLE_TIMEOUT_MS, + onOverflow: ({ size, maxBytes }) => + new Error(`Inworld voices response too large: ${size} bytes (limit: ${maxBytes} bytes)`), + onIdleTimeout: ({ chunkTimeoutMs }) => + new Error(`Inworld voices response stalled: no data received for ${chunkTimeoutMs}ms`), + }) + ).toString("utf8"); + const json = JSON.parse(voicesBody) as { voices?: Array<{ voiceId?: string; displayName?: string;