diff --git a/docs/.generated/config-baseline.sha256 b/docs/.generated/config-baseline.sha256 index b3bf413401b0..bbbc47459e2d 100644 --- a/docs/.generated/config-baseline.sha256 +++ b/docs/.generated/config-baseline.sha256 @@ -1,4 +1,4 @@ -9fb9e35f29a8bfcaf6625e4d4500758a4521c558dd884a643b66af3f8500ecee config-baseline.json +4b5bfc6cdb729bcbd4a4452232e5594ac181128edab14201b18102050c35a1ef config-baseline.json 10a4d786887fafa09ac97fa4039863c96b5b313646f6591faa3857ec97a7857a config-baseline.core.json 29d7abfcdefe9fbdf5c9e2eac31fefba0d6c8225e876ae03c9fedb62f4394e53 config-baseline.channel.json -808f7fad87549a5267cbe568c178b9419e61dfbca9f415ed4f5f9db5da511160 config-baseline.plugin.json +71ca3291bdedaeb2ff28776df25102febb373f00b16073e1b003330ed656d7cb config-baseline.plugin.json diff --git a/docs/gateway/config-tools/custom-providers.md b/docs/gateway/config-tools/custom-providers.md index 2cbcd5a54b94..ace19917c4a4 100644 --- a/docs/gateway/config-tools/custom-providers.md +++ b/docs/gateway/config-tools/custom-providers.md @@ -89,7 +89,7 @@ Configuring a custom/local provider `baseUrl` is also the narrow network trust d - `request.auth`: auth strategy override. Modes: `"provider-default"` (use provider's built-in auth), `"authorization-bearer"` (with `token`), `"header"` (with `headerName`, `value`, optional `prefix`). - `request.proxy`: HTTP proxy override. Modes: `"env-proxy"` (use `HTTP_PROXY`/`HTTPS_PROXY` env vars), `"explicit-proxy"` (with `url`). Both modes accept an optional `tls` sub-object. - `request.tls`: TLS override for direct connections. Fields: `ca`, `cert`, `key`, `passphrase` (all accept SecretRef), `serverName`, `insecureSkipVerify`. - - `request.allowPrivateNetwork`: when `true`, allow model-provider HTTP requests to private, CGNAT, or similar ranges through the provider HTTP fetch guard. Custom/local provider base URLs already trust the exact configured origin, except metadata, link-local, and local-use NAT64 (`64:ff9b:1::/48`) origins, which remain blocked without explicit opt-in. Set this to `false` to opt out of exact-origin trust. WebSocket uses the same `request` for headers/TLS but not that fetch SSRF gate. Default `false`. + - `request.allowPrivateNetwork`: when `true`, allow guarded model-provider HTTP and WebSocket requests to private, CGNAT, or similar ranges through the shared private-network policy. Custom/local provider base URLs already trust the exact configured origin, except metadata, link-local, and local-use NAT64 (`64:ff9b:1::/48`) origins, which remain blocked without explicit opt-in. Set this to `false` to opt out of exact-origin trust. Default `false`. diff --git a/docs/plugins/plugin-inventory.md b/docs/plugins/plugin-inventory.md index 7815cf4a3368..690438b5f266 100644 --- a/docs/plugins/plugin-inventory.md +++ b/docs/plugins/plugin-inventory.md @@ -80,7 +80,7 @@ Each entry lists the package, distribution route, and description. - **[cua-computer](/plugins/reference/cua-computer)** (`@openclaw/cua-computer`) - included in OpenClaw. Experimental CUA Driver computer control for macOS, Windows, and Linux node hosts. -- **[deepgram](/plugins/reference/deepgram)** (`@openclaw/deepgram-provider`) - included in OpenClaw. Adds media understanding provider support. Adds realtime transcription provider support. +- **[deepgram](/plugins/reference/deepgram)** (`@openclaw/deepgram-provider`) - included in OpenClaw. Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition. - **[device-pair](/plugins/reference/device-pair)** (`openclaw`) - included in OpenClaw. Generate setup codes and approve device pairing requests. diff --git a/docs/plugins/reference/deepgram.md b/docs/plugins/reference/deepgram.md index fd03bd5491a6..f2d73fae3ba3 100644 --- a/docs/plugins/reference/deepgram.md +++ b/docs/plugins/reference/deepgram.md @@ -1,5 +1,5 @@ --- -summary: "Adds media understanding provider support. Adds realtime transcription provider support." +summary: "Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition." read_when: - You are installing, configuring, or auditing the deepgram plugin title: "Deepgram plugin reference" @@ -10,7 +10,7 @@ Run `pnpm plugins:inventory:gen` to rebuild it. Hand-written text survives only between the openclaw-plugin-reference:manual-start and openclaw-plugin-reference:manual-end comment markers. --> -Adds media understanding provider support. Adds realtime transcription provider support. +Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition. ## Distribution diff --git a/docs/plugins/sdk-provider-plugins/voice-and-audio.md b/docs/plugins/sdk-provider-plugins/voice-and-audio.md index b282309b914c..6d4230d44a7d 100644 --- a/docs/plugins/sdk-provider-plugins/voice-and-audio.md +++ b/docs/plugins/sdk-provider-plugins/voice-and-audio.md @@ -307,6 +307,37 @@ Register each capability inside `register(api)` alongside your existing model in its result. `transcribeAudio` remains available for providers using host-owned API-key resolution and rotation. + Bundled media providers can use `openProviderWebSocket(...)` from the + private-local `openclaw/plugin-sdk/provider-http` entrypoint. Resolve + request settings with `resolveProviderHttpRequestConfigWithOriginTrust(...)` + first, then pass its `baseUrl`, `headers`, `dispatcherPolicy`, + `allowPrivateNetwork`, and `trustConfiguredBaseUrlOrigin` alongside the + WebSocket `url`. + + Configured proxy routes retain resolved target-address checks. + Applicable ambient HTTP(S) proxies and OpenClaw-managed proxies retain + their existing DNS delegation; `NO_PROXY` bypasses and `ALL_PROXY` alone + do not disable target-address checks. + + Proxy connections use the shared Proxyline-backed Node agent. Prepared + proxy DNS lookups and proxy TLS settings go through the `proxyConnect` + option on `createNodeProxyAgent(...)`; target TLS remains separate. + Proxyline owns pending proxy sockets, including cleanup before CONNECT + completes, so providers do not need a separate proxy-agent dependency. + + - The promise resolves after network-policy and agent preparation, + while the returned socket is still connecting. Attach `error`, + `close`, and `open` handlers immediately; send frames after `open`. + - `timeoutMs` covers DNS preparation, proxy CONNECT, and the WebSocket + handshake as one connection deadline. The connection timer stops on + `open`; the provider owns the remaining transcription deadline. + - `signal` cancels preparation and remains active for the socket's + lifetime. Closing or terminating the socket also cancels a pending + proxy connection. Release the socket in the operation's cleanup path. + - `maxPayloadBytes` limits each incoming message and defaults to 16 MiB. + Compression is disabled. The provider owns audio buffering, frame + pacing, protocol parsing, and transcript-size limits. + ```typescript api.registerMediaUnderstandingProvider({ id: "acme-ai", diff --git a/docs/providers/deepgram.md b/docs/providers/deepgram.md index 99281712f3de..1ff49f34d32d 100644 --- a/docs/providers/deepgram.md +++ b/docs/providers/deepgram.md @@ -11,8 +11,9 @@ Deepgram is a speech-to-text API. OpenClaw uses it for inbound audio/voice-note transcription through `tools.media.audio` and for Voice Call streaming STT through `plugins.entries.voice-call.config.streaming`. -Batch transcription uploads the complete audio file to Deepgram and injects -the transcript into the reply pipeline (`{{Transcript}}` + `[Audio]` block). +Batch transcription uploads the complete audio file to Deepgram. Flux models +use Deepgram's one-shot WebSocket instead. Both paths inject the transcript +into the reply pipeline (`{{Transcript}}` + `[Audio]` block). Voice Call streaming forwards live G.711 u-law frames over Deepgram's WebSocket `listen` endpoint and emits partial/final transcripts as Deepgram returns them. @@ -102,6 +103,22 @@ Deepgram `/listen` request, so any Deepgram-supported param name works +### Flux models + +Use `flux-general-en` or `flux-general-multi` for Deepgram Flux. OpenClaw +converts the voice note to 16 kHz mono linear16 audio with `ffmpeg`, then sends +it to Deepgram's `/v2/listen` WebSocket endpoint. +Set `tools.media.models[].model` to either Flux model in the getting-started +configuration above. + +Flux supports `eager_eot_threshold`, `eot_threshold`, `eot_timeout_ms`, +`keyterm`, `language_hint`, `mip_opt_out`, `numerals`, `profanity_filter`, +`redact`, and `tag` in `providerOptions.deepgram`. OpenClaw ignores batch-only +options such as `detect_language`, `punctuate`, and `smart_format` on Flux. +Language hints apply only to `flux-general-multi`: OpenClaw maps the model entry's +`language` setting to `language_hint`, with an explicit provider option taking +precedence. Both settings are ignored for the English-only `flux-general-en` model. + ## Voice Call streaming STT The bundled `deepgram` plugin also registers a realtime transcription provider @@ -169,6 +186,10 @@ Twilio media frames can be forwarded directly. Output follows the same audio rules as other providers (size caps, timeouts, transcript injection). + + Install `ffmpeg` with the gateway host's package manager before selecting a + Flux model. + ## Related diff --git a/extensions/deepgram/audio-flux.test.ts b/extensions/deepgram/audio-flux.test.ts new file mode 100644 index 000000000000..7318710d4eab --- /dev/null +++ b/extensions/deepgram/audio-flux.test.ts @@ -0,0 +1,296 @@ +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; +import type { AudioTranscriptionRequest } from "openclaw/plugin-sdk/media-understanding"; +import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { RawData, WebSocket } from "ws"; +import { WebSocketServer } from "ws"; +import { isDeepgramFluxModel } from "./audio-flux.js"; +import { transcribeDeepgramAudio } from "./audio.js"; + +const runCommandBuffered = vi.hoisted(() => vi.fn()); +const prepareWebSocket = vi.hoisted(() => vi.fn<() => Promise>()); + +vi.mock("openclaw/plugin-sdk/media-runtime", async (importOriginal) => ({ + ...(await importOriginal()), + resolveFfmpegBin: () => "/usr/bin/ffmpeg", +})); +vi.mock("openclaw/plugin-sdk/process-runtime", () => ({ runCommandBuffered })); +vi.mock("openclaw/plugin-sdk/provider-http", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + openProviderWebSocket: async (params: Parameters[0]) => { + await prepareWebSocket(); + return await actual.openProviderWebSocket(params); + }, + }; +}); + +const cleanups: Array<() => Promise> = []; + +function parseClientMessage(data: RawData): Record | undefined { + if (typeof data !== "string" && !Buffer.isBuffer(data)) { + return undefined; + } + const parsed: unknown = JSON.parse(data.toString()); + return asOptionalRecord(parsed); +} + +async function createFluxServer(params: { + onCloseStream: (socket: WebSocket) => void; + onRequest?: (url: URL, headers: Record) => void; +}) { + const server = createServer(); + const websocketServer = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 }); + const audioFrames: Buffer[] = []; + server.on("upgrade", (request, socket, head) => { + params.onRequest?.(new URL(request.url ?? "/", "http://127.0.0.1"), request.headers); + websocketServer.handleUpgrade(request, socket, head, (client) => { + client.on("message", (data, isBinary) => { + if (isBinary) { + const bytes = Array.isArray(data) + ? Buffer.concat(data) + : Buffer.isBuffer(data) + ? data + : Buffer.from(data); + audioFrames.push(bytes); + return; + } + if (parseClientMessage(data)?.type === "CloseStream") { + params.onCloseStream(client); + } + }); + }); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const port = (server.address() as AddressInfo).port; + cleanups.push( + async () => + await new Promise((resolve, reject) => { + for (const client of websocketServer.clients) { + client.terminate(); + } + websocketServer.close(() => server.close((error) => (error ? reject(error) : resolve()))); + }), + ); + return { audioFrames, baseUrl: `http://127.0.0.1:${port}/v1` }; +} + +function fluxRequest( + baseUrl: string, + extra: Partial = {}, +): AudioTranscriptionRequest { + return { + buffer: Buffer.from("source audio"), + fileName: "note.ogg", + apiKey: "default-key", + baseUrl, + model: "flux-general-multi", + timeoutMs: 5000, + request: { allowPrivateNetwork: true }, + ...extra, + }; +} + +function mockDecodedPcm(pcm: Buffer): void { + runCommandBuffered.mockResolvedValueOnce({ + stdout: pcm, + stderr: Buffer.alloc(0), + code: 0, + signal: null, + killed: false, + termination: "exit", + }); +} + +describe("Deepgram Flux audio", () => { + afterEach(async () => { + runCommandBuffered.mockReset(); + prepareWebSocket.mockReset(); + vi.restoreAllMocks(); + vi.useRealTimers(); + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); + }); + + it("keeps decoding, connection preparation, and transcription within one deadline", async () => { + const decodeStarted = createDeferred(); + const releaseDecode = createDeferred(); + const preparationStarted = createDeferred(); + const releasePreparation = createDeferred(); + const flushed = createDeferred(); + const server = await createFluxServer({ onCloseStream: () => flushed.resolve() }); + runCommandBuffered.mockImplementationOnce(async () => { + decodeStarted.resolve(); + await releaseDecode.promise; + return { + stdout: Buffer.alloc(10, 1), + stderr: Buffer.alloc(0), + code: 0, + signal: null, + killed: false, + termination: "exit", + }; + }); + prepareWebSocket.mockImplementationOnce(async () => { + preparationStarted.resolve(); + await releasePreparation.promise; + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + let failure: unknown; + const transcription = transcribeDeepgramAudio( + fluxRequest(server.baseUrl, { timeoutMs: 1000 }), + ).catch((error: unknown) => { + failure = error; + }); + + await decodeStarted.promise; + await vi.advanceTimersByTimeAsync(200); + releaseDecode.resolve(); + await preparationStarted.promise; + await vi.advanceTimersByTimeAsync(300); + releasePreparation.resolve(); + await flushed.promise; + await vi.advanceTimersByTimeAsync(499); + expect(failure).toBeUndefined(); + await vi.advanceTimersByTimeAsync(1); + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ message: expect.stringContaining("timed out") }); + await transcription; + }); + + it("routes documented Flux models only", () => { + expect(isDeepgramFluxModel("flux-general-en")).toBe(true); + expect(isDeepgramFluxModel(" Flux-General-Multi ")).toBe(true); + expect(isDeepgramFluxModel("flux")).toBe(false); + expect(isDeepgramFluxModel("reflux-general-en")).toBe(false); + expect(isDeepgramFluxModel("nova-3")).toBe(false); + }); + + it.each([ + { model: "flux-general-multi", language: " en ", queryLanguage: undefined, expectedHint: "en" }, + { model: "flux-general-multi", language: "en", queryLanguage: "fr", expectedHint: "fr" }, + { model: "flux-general-en", language: " en ", queryLanguage: undefined, expectedHint: null }, + { model: "flux-general-en", language: undefined, queryLanguage: "en", expectedHint: null }, + ])( + "uses valid protocol fields for $model with language=$language and query=$queryLanguage", + async ({ model, language, queryLanguage, expectedHint }) => { + const pcm = Buffer.alloc(6000, 1); + mockDecodedPcm(pcm); + let requestUrl: URL | undefined; + let authorization: string | string[] | undefined; + const server = await createFluxServer({ + onRequest: (url, headers) => { + requestUrl = url; + authorization = headers.authorization; + }, + onCloseStream: (socket) => { + socket.send( + JSON.stringify({ type: "TurnInfo", event: "EndOfTurn", transcript: "life moves" }), + ); + socket.send( + JSON.stringify({ type: "TurnInfo", event: "EndOfTurn", transcript: "pretty fast" }), + ); + socket.close(); + }, + }); + + const result = await transcribeDeepgramAudio( + fluxRequest(server.baseUrl, { + model, + language, + query: { + ...(queryLanguage === undefined ? {} : { language_hint: queryLanguage }), + eot_threshold: 0.7, + numerals: true, + profanity_filter: true, + smart_format: true, + }, + request: { + allowPrivateNetwork: true, + auth: { + mode: "header", + headerName: "authorization", + value: "Token configured-key", + }, + }, + }), + ); + + expect(result).toEqual({ model, text: "life moves pretty fast" }); + expect(authorization).toBe("Token configured-key"); + expect(requestUrl?.pathname).toBe("/v2/listen"); + expect(requestUrl?.searchParams.get("encoding")).toBe("linear16"); + expect(requestUrl?.searchParams.get("sample_rate")).toBe("16000"); + expect(requestUrl?.searchParams.get("language_hint")).toBe(expectedHint); + expect(requestUrl?.searchParams.get("eot_threshold")).toBe("0.7"); + expect(requestUrl?.searchParams.get("numerals")).toBe("true"); + expect(requestUrl?.searchParams.get("profanity_filter")).toBe("true"); + expect(requestUrl?.searchParams.has("smart_format")).toBe(false); + expect(server.audioFrames.map((frame) => frame.byteLength)).toEqual([2560, 2560, 880]); + expect(Buffer.concat(server.audioFrames)).toEqual(pcm); + expect(runCommandBuffered).toHaveBeenCalledWith( + expect.arrayContaining([ + "/usr/bin/ffmpeg", + "-t", + "1200", + "-c:a", + "pcm_s16le", + "-ar", + "16000", + ]), + expect.objectContaining({ + maxOutputBytes: { stdout: 38_400_000, stderr: 65_536 }, + terminateOnOutputError: true, + }), + ); + }, + ); + + it.each(["null", "[]", "42"])("rejects valid non-object server JSON: %s", async (payload) => { + mockDecodedPcm(Buffer.alloc(10, 1)); + const server = await createFluxServer({ + onCloseStream: (socket) => socket.send(payload), + }); + await expect(transcribeDeepgramAudio(fluxRequest(server.baseUrl))).rejects.toThrow( + "malformed JSON response", + ); + }); + + it("rejects retained transcript growth above the provider limit", async () => { + mockDecodedPcm(Buffer.alloc(10, 1)); + const server = await createFluxServer({ + onCloseStream: (socket) => + socket.send( + JSON.stringify({ + type: "TurnInfo", + event: "EndOfTurn", + transcript: "x".repeat(256 * 1024 + 1), + }), + ), + }); + await expect(transcribeDeepgramAudio(fluxRequest(server.baseUrl))).rejects.toThrow( + "transcript exceeds size limit", + ); + }); + + it("does not open a private socket without the request-policy opt-in", async () => { + mockDecodedPcm(Buffer.alloc(10, 1)); + let opened = false; + const server = await createFluxServer({ + onRequest: () => { + opened = true; + }, + onCloseStream: () => undefined, + }); + await expect( + transcribeDeepgramAudio( + fluxRequest(server.baseUrl, { request: { allowPrivateNetwork: false } }), + ), + ).rejects.toThrow(/private|loopback|blocked/iu); + expect(opened).toBe(false); + }); +}); diff --git a/extensions/deepgram/audio-flux.ts b/extensions/deepgram/audio-flux.ts new file mode 100644 index 000000000000..d47a51973285 --- /dev/null +++ b/extensions/deepgram/audio-flux.ts @@ -0,0 +1,331 @@ +// Deepgram Flux voice-note transcription uses the provider's one-shot WebSocket protocol. +import { + MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS, + resolveFfmpegBin, +} from "openclaw/plugin-sdk/media-runtime"; +import type { + AudioTranscriptionRequest, + AudioTranscriptionResult, +} from "openclaw/plugin-sdk/media-understanding"; +import { runCommandBuffered } from "openclaw/plugin-sdk/process-runtime"; +import { + createProviderOperationDeadline, + createProviderOperationTimeoutResolver, + openProviderWebSocket, + requireTranscriptionText, +} from "openclaw/plugin-sdk/provider-http"; +import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; + +const DEEPGRAM_FLUX_SAMPLE_RATE = 16_000; +// Deepgram recommends 80 ms chunks. 16 kHz mono linear16 contains 32 bytes per millisecond. +const DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES = 2_560; +const DEEPGRAM_FLUX_MAX_MESSAGE_BYTES = 1024 * 1024; +const DEEPGRAM_FLUX_MAX_TRANSCRIPT_BYTES = 256 * 1024; +const DEEPGRAM_FLUX_MAX_PCM_BYTES = + DEEPGRAM_FLUX_SAMPLE_RATE * 2 * MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS; +const DEEPGRAM_FLUX_QUERY_KEYS = new Set([ + "eager_eot_threshold", + "eot_threshold", + "eot_timeout_ms", + "keyterm", + "language_hint", + "mip_opt_out", + "numerals", + "profanity_filter", + "redact", + "tag", +]); + +type DeepgramFluxRequestConfig = { + allowPrivateNetwork: boolean; + baseUrl: string; + dispatcherPolicy?: Parameters[0]["dispatcherPolicy"]; + headers: Headers; + trustConfiguredBaseUrlOrigin: boolean; +}; + +export function isDeepgramFluxModel(model?: string): boolean { + return model?.trim().toLowerCase().startsWith("flux-") ?? false; +} + +function buildDeepgramFluxUrl(params: { + baseUrl: string; + language?: string; + model: string; + query?: Record; +}): string { + let url: URL; + try { + url = new URL(params.baseUrl); + } catch { + throw new Error("Invalid Deepgram baseUrl: value is not a valid URL"); + } + if (url.protocol === "http:") { + url.protocol = "ws:"; + } else if (url.protocol === "https:") { + url.protocol = "wss:"; + } else if (url.protocol !== "ws:" && url.protocol !== "wss:") { + throw new Error( + `Invalid Deepgram baseUrl: unsupported scheme "${url.protocol}" (expected http, https, ws, or wss)`, + ); + } + const basePath = url.pathname.replace(/\/+$/u, ""); + url.pathname = `${basePath ? basePath.replace(/\/v1$/u, "/v2") : "/v2"}/listen`; + url.search = ""; + url.searchParams.set("model", params.model); + url.searchParams.set("encoding", "linear16"); + url.searchParams.set("sample_rate", String(DEEPGRAM_FLUX_SAMPLE_RATE)); + const query = { ...params.query }; + // Deepgram's v2 contract permits language hints only on its multilingual model. + query.language_hint = + params.model === "flux-general-multi" + ? (query.language_hint ?? (params.language?.trim() || undefined)) + : undefined; + for (const [key, value] of Object.entries(query)) { + if (value !== undefined && DEEPGRAM_FLUX_QUERY_KEYS.has(key)) { + url.searchParams.set(key, String(value)); + } + } + return url.toString(); +} + +function readFluxEvent(data: Buffer | ArrayBuffer | Buffer[]): Record { + const bytes = Array.isArray(data) + ? Buffer.concat(data) + : Buffer.isBuffer(data) + ? data + : Buffer.from(data); + let parsed: unknown; + try { + parsed = JSON.parse(bytes.toString("utf8")); + } catch { + throw new Error("Audio transcription failed: malformed JSON response"); + } + const event = asOptionalRecord(parsed); + if (!event) { + throw new Error("Audio transcription failed: malformed JSON response"); + } + return event; +} + +function readFluxErrorDetail(event: Record): string { + const nested = asOptionalRecord(event.error); + for (const value of [event.description, event.message, nested?.message]) { + if (typeof value === "string" && value.trim()) { + return value.trim(); + } + } + return "Deepgram Flux transcription error"; +} + +type DeepgramFluxSocket = Awaited>; + +function sendSocketFrame(socket: DeepgramFluxSocket, data: Buffer | string): Promise { + return new Promise((resolve, reject) => { + socket.send(data, (error) => (error ? reject(error) : resolve())); + }); +} + +async function decodeDeepgramFluxAudio(params: { + buffer: Buffer; + signal?: AbortSignal; + timeoutMs: number; +}): Promise { + const result = await runCommandBuffered( + [ + resolveFfmpegBin(), + "-hide_banner", + "-loglevel", + "error", + "-i", + "pipe:0", + "-t", + String(MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS), + "-vn", + "-sn", + "-dn", + "-c:a", + "pcm_s16le", + "-ar", + String(DEEPGRAM_FLUX_SAMPLE_RATE), + "-ac", + "1", + "-f", + "s16le", + "pipe:1", + ], + { + input: params.buffer, + maxOutputBytes: { stdout: DEEPGRAM_FLUX_MAX_PCM_BYTES, stderr: 64 * 1024 }, + signal: params.signal, + terminateOnOutputError: true, + timeoutMs: params.timeoutMs, + }, + ); + if (result.termination === "exit" && result.code === 0) { + return result.stdout; + } + if (result.termination === "output-limit") { + throw new Error("Audio transcription failed: decoded audio exceeds size limit"); + } + const detail = result.stderr.toString("utf8").trim(); + throw new Error( + `Audio transcription failed: ffmpeg ${result.termination}${detail ? `: ${detail}` : ""}`, + { cause: result.error }, + ); +} + +export async function transcribeDeepgramFluxAudio(params: { + request: AudioTranscriptionRequest; + requestConfig: DeepgramFluxRequestConfig; + model: string; +}): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.request.timeoutMs, + label: "Deepgram Flux transcription", + }); + const resolveTimeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: params.request.timeoutMs, + }); + const pcm = await decodeDeepgramFluxAudio({ + buffer: params.request.buffer, + signal: params.request.signal, + timeoutMs: resolveTimeoutMs(), + }); + if (pcm.byteLength === 0) { + throw new Error("Audio transcription failed: decoded audio is empty"); + } + + const url = buildDeepgramFluxUrl({ + baseUrl: params.requestConfig.baseUrl, + language: params.request.language, + model: params.model, + query: params.request.query, + }); + const timeoutMs = resolveTimeoutMs(); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: params.requestConfig.allowPrivateNetwork, + baseUrl: params.requestConfig.baseUrl, + dispatcherPolicy: params.requestConfig.dispatcherPolicy, + headers: params.requestConfig.headers, + maxPayloadBytes: DEEPGRAM_FLUX_MAX_MESSAGE_BYTES, + signal: params.request.signal, + timeoutMs, + trustConfiguredBaseUrlOrigin: params.requestConfig.trustConfiguredBaseUrlOrigin, + url, + }); + + const transcript = await new Promise((resolve, reject) => { + const finalizedTurns: string[] = []; + let finalizedTranscriptBytes = 0; + let lastPartial = ""; + let settled = false; + let closeStreamSent = false; + let timer: ReturnType | undefined; + + const settle = (outcome: { error?: Error; text?: string }) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + timer = undefined; + if (outcome.error) { + reject(outcome.error); + } else { + resolve(outcome.text ?? ""); + } + }; + socket.on("open", () => { + void (async () => { + try { + for (let offset = 0; offset < pcm.byteLength; offset += DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES) { + await sendSocketFrame( + socket, + pcm.subarray(offset, offset + DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES), + ); + } + await sendSocketFrame(socket, JSON.stringify({ type: "CloseStream" })); + closeStreamSent = true; + } catch (error) { + settle({ + error: new Error( + `Audio transcription failed: ${error instanceof Error ? error.message : String(error)}`, + ), + }); + } + })(); + }); + + socket.on("message", (data) => { + try { + const event = readFluxEvent(data); + if (event.type === "Error" || event.type === "error") { + settle({ + error: new Error(`Audio transcription failed: ${readFluxErrorDetail(event)}`), + }); + return; + } + if (event.type !== "TurnInfo" || typeof event.transcript !== "string") { + return; + } + if (event.event === "EndOfTurn") { + finalizedTranscriptBytes += Buffer.byteLength(event.transcript, "utf8"); + finalizedTurns.push(event.transcript); + lastPartial = ""; + } else { + lastPartial = event.transcript; + } + if ( + finalizedTranscriptBytes + Buffer.byteLength(lastPartial, "utf8") > + DEEPGRAM_FLUX_MAX_TRANSCRIPT_BYTES + ) { + settle({ error: new Error("Audio transcription failed: transcript exceeds size limit") }); + } + } catch (error) { + settle({ error: error instanceof Error ? error : new Error(String(error)) }); + } + }); + + socket.on("error", (error) => { + settle({ error: new Error(`Audio transcription failed: ${error.message}`) }); + }); + + socket.on("close", (code, reason) => { + if (!closeStreamSent) { + settle({ error: new Error("Audio transcription failed: Flux closed before flush") }); + return; + } + if (code !== 1000 && code !== 1005) { + const detail = reason.toString().trim(); + settle({ + error: new Error( + `Audio transcription failed: Flux closed abnormally (code ${code}${detail ? `: ${detail}` : ""})`, + ), + }); + return; + } + settle({ + text: [...finalizedTurns, lastPartial] + .map((part) => part.trim()) + .filter(Boolean) + .join(" "), + }); + }); + // Connection preparation consumed part of the original operation budget. + // Register handlers first so an exhausted budget still closes the socket safely. + timer = setTimeout( + () => settle({ error: new Error("Deepgram Flux transcription timed out") }), + resolveTimeoutMs(), + ); + }).finally(() => socket.terminate()); + + return { + text: requireTranscriptionText( + transcript || undefined, + "Audio transcription response missing transcript", + ), + model: params.model, + }; +} diff --git a/extensions/deepgram/audio.live.test.ts b/extensions/deepgram/audio.live.test.ts index be89bd36c70c..34a2aa9fe379 100644 --- a/extensions/deepgram/audio.live.test.ts +++ b/extensions/deepgram/audio.live.test.ts @@ -72,7 +72,9 @@ describeLive("deepgram live", () => { baseUrl: DEEPGRAM_BASE_URL, timeoutMs: 20000, }); - expect(result.text.trim().length).toBeGreaterThan(0); + expect(result.text.toLowerCase().replaceAll(/[^a-z0-9]/gu, "")).toContain( + "lifemovesprettyfast", + ); }, 30000); it("streams realtime STT through the registered transcription provider", async () => { diff --git a/extensions/deepgram/audio.ts b/extensions/deepgram/audio.ts index f6d5a94eeb75..2cd24318b956 100644 --- a/extensions/deepgram/audio.ts +++ b/extensions/deepgram/audio.ts @@ -45,25 +45,30 @@ export async function transcribeDeepgramAudio( assertOkOrThrowHttpError, postTranscriptionRequest, readProviderJsonObjectResponse, - resolveProviderHttpRequestConfig, + resolveProviderHttpRequestConfigWithOriginTrust, requireTranscriptionText, } = await import("openclaw/plugin-sdk/provider-http"); - const fetchFn = params.fetchFn ?? fetch; + const { isDeepgramFluxModel, transcribeDeepgramFluxAudio } = await import("./audio-flux.js"); const model = resolveModel(params.model); - const { baseUrl, allowPrivateNetwork, headers, dispatcherPolicy } = - resolveProviderHttpRequestConfig({ - baseUrl: params.baseUrl, - defaultBaseUrl: DEFAULT_DEEPGRAM_AUDIO_BASE_URL, - headers: params.headers, - request: params.request, - defaultHeaders: { - authorization: `Token ${params.apiKey}`, - "content-type": params.mime ?? "application/octet-stream", - }, - provider: "deepgram", - capability: "audio", - transport: "media-understanding", - }); + const flux = isDeepgramFluxModel(model); + const requestConfig = resolveProviderHttpRequestConfigWithOriginTrust({ + baseUrl: params.baseUrl, + defaultBaseUrl: DEFAULT_DEEPGRAM_AUDIO_BASE_URL, + headers: params.headers, + request: params.request, + defaultHeaders: { + authorization: `Token ${params.apiKey}`, + ...(flux ? {} : { "content-type": params.mime ?? "application/octet-stream" }), + }, + provider: "deepgram", + capability: "audio", + transport: "media-understanding", + }); + if (flux) { + return await transcribeDeepgramFluxAudio({ request: params, requestConfig, model }); + } + const fetchFn = params.fetchFn ?? fetch; + const { baseUrl, allowPrivateNetwork, headers, dispatcherPolicy } = requestConfig; const url = new URL(`${baseUrl}/listen`); url.searchParams.set("model", model); diff --git a/extensions/deepgram/openclaw.plugin.json b/extensions/deepgram/openclaw.plugin.json index 4a4d88bc880c..7a1d3b132d8c 100644 --- a/extensions/deepgram/openclaw.plugin.json +++ b/extensions/deepgram/openclaw.plugin.json @@ -1,6 +1,10 @@ { "id": "deepgram", - "categories": ["models", "voice", "media"], + "categories": [ + "models", + "voice", + "media" + ], "capabilityCatalogEntry": "./capability-catalog.ts", "activation": { "onStartup": false @@ -10,17 +14,25 @@ "providers": [ { "id": "deepgram", - "envVars": ["DEEPGRAM_API_KEY"] + "envVars": [ + "DEEPGRAM_API_KEY" + ] } ] }, "contracts": { - "mediaUnderstandingProviders": ["deepgram"], - "realtimeTranscriptionProviders": ["deepgram"] + "mediaUnderstandingProviders": [ + "deepgram" + ], + "realtimeTranscriptionProviders": [ + "deepgram" + ] }, "mediaUnderstandingProviderMetadata": { "deepgram": { - "capabilities": ["audio"], + "capabilities": [ + "audio" + ], "defaultModels": { "audio": "nova-3" }, @@ -33,5 +45,6 @@ "type": "object", "additionalProperties": false, "properties": {} - } + }, + "description": "Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition." } diff --git a/package.json b/package.json index df376c58825c..a854391efd8c 100644 --- a/package.json +++ b/package.json @@ -2159,7 +2159,7 @@ "@mozilla/readability": "0.6.0", "@openclaw/ai": "workspace:*", "@openclaw/fs-safe": "0.8.5", - "@openclaw/proxyline": "0.3.11", + "@openclaw/proxyline": "0.3.12", "@silvia-odwyer/photon-node": "0.3.4", "@trycua/cua-driver": "0.22.2", "acorn": "8.18.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 32b40252e7b8..89b3da742c3c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -205,8 +205,8 @@ importers: specifier: 0.8.5 version: 0.8.5 '@openclaw/proxyline': - specifier: 0.3.11 - version: 0.3.11(undici@8.10.2) + specifier: 0.3.12 + version: 0.3.12(undici@8.10.2) '@silvia-odwyer/photon-node': specifier: 0.3.4 version: 0.3.4 @@ -4211,8 +4211,8 @@ packages: node-pty: optional: true - '@openclaw/proxyline@0.3.11': - resolution: {integrity: sha512-XDlrrQE0fapfTpqCBK9WB+U4W+/0lsrRIXI2nZ74W/lWMF+X4T5t2dfj8Ms250Th4qjnNi6WuSX5OZJqyLoZ3g==} + '@openclaw/proxyline@0.3.12': + resolution: {integrity: sha512-KPUvtLgORF0NWxiuUPwHBuxYhUsFgxnOm8+XkDR0on236lnfMF44A5BnOvBkeDV8wrBaGHs/rQ8kt1tB4WNhmw==} engines: {node: '>=22.19.0'} peerDependencies: undici: '>=8.5.0 <9' @@ -11543,7 +11543,7 @@ snapshots: dependencies: ghostty-web: 0.4.0 - '@openclaw/proxyline@0.3.11(undici@8.10.2)': + '@openclaw/proxyline@0.3.12(undici@8.10.2)': dependencies: undici: 8.10.2 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 4f307cf72e12..9344a929e218 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -10,8 +10,8 @@ minimumReleaseAgeStrict: true # Trusted Codex runtimes are outside the dependency cooldown. minimumReleaseAgeExclude: - # Reviewed Proxyline Bun runtime fix; remove after 2026-09-13 20:51 UTC. - - "@openclaw/proxyline@0.3.11" + # Reviewed proxy DNS/TLS controls; remove after 2026-09-15 10:08 UTC. + - "@openclaw/proxyline@0.3.12" # GHSA-vp8m-p9jh-q5pm / GHSA-w293-vg96-wgc3 security fixes; remove after 2026-09-12. - "undici@8.10.2" - "undici@7.29.1" diff --git a/src/infra/net/fetch-guard.ts b/src/infra/net/fetch-guard.ts index 7027114f865b..cb174d156cd0 100644 --- a/src/infra/net/fetch-guard.ts +++ b/src/infra/net/fetch-guard.ts @@ -176,7 +176,7 @@ function resolveGuardedFetchMode(params: GuardedFetchOptions): GuardedFetchMode return GUARDED_FETCH_MODE.STRICT; } -function isManagedProxyActive(): boolean { +export function isManagedProxyActive(): boolean { return process.env["OPENCLAW_PROXY_ACTIVE"] === "1"; } diff --git a/src/infra/net/node-proxy-agent.ts b/src/infra/net/node-proxy-agent.ts index 85d2b297a987..6a1ecb489e81 100644 --- a/src/infra/net/node-proxy-agent.ts +++ b/src/infra/net/node-proxy-agent.ts @@ -15,6 +15,7 @@ type ProxylineCreateAmbientNodeProxyAgent = type ProxylineAgentOptions = NonNullable[0]>; type ProxylineEnvSnapshot = NonNullable; type ProxylineTlsOptions = ProxylineAgentOptions["proxyTls"]; +type ProxylineProxyConnectOptions = import("@openclaw/proxyline").ProxyConnectOptions; type NodeProxyAgentOptions = HttpAgentOptions & HttpsAgentOptions; type NodeProxyAgentWithOptions = HttpAgent & { keepAlive: boolean; @@ -36,12 +37,14 @@ export type CreateNodeProxyAgentOptions = targetUrl: string | URL; protocol?: NodeProxyProtocol; agentOptions?: NodeProxyAgentOptions; + proxyConnect?: ProxylineProxyConnectOptions; } | { mode: "explicit"; proxyUrl: string | URL; protocol?: NodeProxyProtocol; agentOptions?: NodeProxyAgentOptions; + proxyConnect?: ProxylineProxyConnectOptions; }; function proxyUrlWithDefaultScheme(proxyUrl: string, protocol: NodeProxyProtocol): URL { @@ -163,16 +166,19 @@ function createFixedNodeProxyAgent( protocol?: NodeProxyProtocol; proxyTls?: ProxylineTlsOptions; agentOptions?: NodeProxyAgentOptions; + proxyConnect?: ProxylineProxyConnectOptions; } = {}, ): HttpAgent { const parsedProxyUrl = proxyUrl instanceof URL ? proxyUrl : proxyUrlWithDefaultScheme(proxyUrl, options.protocol ?? "https"); + const proxyConnect = options.proxyConnect; const agent = loadCreateAmbientNodeProxyAgent()({ env: fixedProxyEnv(parsedProxyUrl), protocol: options.protocol ?? "https", ...(options.proxyTls !== undefined ? { proxyTls: options.proxyTls } : {}), + ...(proxyConnect !== undefined ? { resolveProxyConnectOptions: () => proxyConnect } : {}), }); if (agent === undefined) { throw new Error(`${UNSUPPORTED_PROXY_PROTOCOL_MESSAGE} Got ${parsedProxyUrl.protocol}`); @@ -194,11 +200,13 @@ export function createNodeProxyAgent(options: CreateNodeProxyAgentOptions): Http return createFixedNodeProxyAgent(options.proxyUrl, { protocol: options.protocol, agentOptions: options.agentOptions, + proxyConnect: options.proxyConnect, }); } return createEnvNodeProxyAgentForTarget(options.targetUrl, { protocol: options.protocol, agentOptions: options.agentOptions, + proxyConnect: options.proxyConnect, }); } @@ -207,6 +215,7 @@ function createEnvNodeProxyAgentForTarget( options: { protocol?: NodeProxyProtocol; agentOptions?: NodeProxyAgentOptions; + proxyConnect?: ProxylineProxyConnectOptions; } = {}, ): HttpAgent | undefined { const target = resolveEnvNodeProxyTarget(targetUrl); @@ -217,6 +226,7 @@ function createEnvNodeProxyAgentForTarget( protocol: options.protocol ?? target.protocol, proxyTls: resolveActiveManagedProxyTlsOptions({ proxyUrl: target.proxyUrl.href }), agentOptions: options.agentOptions, + proxyConnect: options.proxyConnect, }); } diff --git a/src/infra/net/provider-websocket.test.ts b/src/infra/net/provider-websocket.test.ts new file mode 100644 index 000000000000..1696fb5c16d3 --- /dev/null +++ b/src/infra/net/provider-websocket.test.ts @@ -0,0 +1,402 @@ +import { once } from "node:events"; +import { createServer as createHttpServer } from "node:http"; +import { createServer as createHttpsServer } from "node:https"; +import net, { type AddressInfo } from "node:net"; +import type { Duplex } from "node:stream"; +import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { WebSocketServer } from "ws"; +import { createDeferred, withTestTimeout } from "../../../test/helpers/promise.js"; +import { TEST_TLS_CERT_PEM, TEST_TLS_KEY_PEM } from "../../../test/helpers/tls-fixture.js"; +import { openProviderWebSocket } from "./provider-websocket.js"; +import * as ssrf from "./ssrf.js"; + +const cleanups: Array<() => Promise> = []; + +async function createStalledHandshakeServer(event: "upgrade" | "connect") { + const server = createHttpServer(); + const received = createDeferred(); + const connections = new Set(); + let connectionCount = 0; + server.on("connection", (socket) => { + connectionCount += 1; + connections.add(socket); + socket.once("close", () => connections.delete(socket)); + }); + server.on(event, (_request, socket) => { + // HTTP hands upgraded sockets to us with allowHalfOpen enabled. + // Complete our half only after the client ends its connection. + socket.once("end", () => socket.end()); + socket.resume(); + received.resolve(socket); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const port = (server.address() as AddressInfo).port; + cleanups.push(async () => { + for (const socket of connections) { + socket.destroy(); + } + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }); + return { + url: `http://127.0.0.1:${port}`, + received: received.promise, + connectionCount: () => connectionCount, + }; +} + +function configureProxyEnvironment(proxyUrl: string, noProxy = "") { + vi.stubEnv("OPENCLAW_PROXY_ACTIVE", undefined); + for (const name of [ + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "http_proxy", + "https_proxy", + "all_proxy", + ]) { + vi.stubEnv(name, proxyUrl); + } + vi.stubEnv("NO_PROXY", noProxy); + vi.stubEnv("no_proxy", noProxy); +} + +function resolveProviderToLoopback() { + const resolveHostname = ssrf.resolvePinnedHostnameWithPolicy; + vi.spyOn(ssrf, "resolvePinnedHostnameWithPolicy").mockImplementation((hostname, params) => + resolveHostname( + hostname, + hostname === "provider.example" + ? { ...params, lookupFn: async () => [{ address: "127.0.0.1", family: 4 }] } + : params, + ), + ); +} + +async function createLocalWebSocketServer(options: { tls?: boolean } = {}) { + const server = options.tls + ? createHttpsServer({ cert: TEST_TLS_CERT_PEM, key: TEST_TLS_KEY_PEM }) + : createHttpServer(); + const websocketServer = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 }); + const requestHeaders: Array> = []; + server.on("upgrade", (request, socket, head) => { + requestHeaders.push(request.headers); + websocketServer.handleUpgrade(request, socket, head, (client) => { + websocketServer.emit("connection", client, request); + }); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const port = (server.address() as AddressInfo).port; + cleanups.push( + async () => + await new Promise((resolve, reject) => { + for (const client of websocketServer.clients) { + client.terminate(); + } + websocketServer.close(() => server.close((error) => (error ? reject(error) : resolve()))); + }), + ); + return { + requestHeaders, + url: `${options.tls ? "wss" : "ws"}://127.0.0.1:${port}/listen`, + }; +} + +async function createConnectProxy(proxyHostname = "127.0.0.1") { + const server = createHttpServer(); + let connectCount = 0; + server.on("connect", (request, clientSocket, head) => { + connectCount += 1; + const [hostname, rawPort] = (request.url ?? "").split(":"); + const targetSocket = net.connect(Number(rawPort), hostname, () => { + clientSocket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); + if (head.byteLength > 0) { + targetSocket.write(head); + } + targetSocket.pipe(clientSocket); + clientSocket.pipe(targetSocket); + }); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const port = (server.address() as AddressInfo).port; + cleanups.push( + async () => + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }), + ); + return { connectCount: () => connectCount, url: `http://${proxyHostname}:${port}` }; +} + +describe("openProviderWebSocket", () => { + afterEach(async () => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + vi.useRealTimers(); + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); + }); + + it.each(["explicit-proxy", "env-proxy"] as const)( + "rejects private target DNS before connecting through configured %s", + async (mode) => { + const proxy = await createStalledHandshakeServer("connect"); + configureProxyEnvironment(proxy.url); + resolveProviderToLoopback(); + const opening = openProviderWebSocket({ + allowPrivateNetwork: false, + baseUrl: "wss://provider.example/audio", + dispatcherPolicy: + mode === "explicit-proxy" + ? { mode, proxyUrl: proxy.url, allowPrivateProxy: true } + : { mode }, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: "wss://provider.example/audio", + }).then((socket) => { + socket.on("error", () => {}); + socket.terminate(); + return socket; + }); + await expect(opening).rejects.toThrow(/private|loopback|blocked/iu); + expect(proxy.connectionCount()).toBe(0); + }, + ); + + it.each([ + { name: "an ambient proxy", managed: false, dispatcherPolicy: undefined }, + { + name: "a managed proxy with configured env-proxy policy", + managed: true, + dispatcherPolicy: { mode: "env-proxy" }, + }, + { + name: "a managed proxy with configured direct policy", + managed: true, + dispatcherPolicy: { mode: "direct" }, + }, + ] as const)("leaves target DNS to $name", async ({ managed, dispatcherPolicy }) => { + const proxy = await createStalledHandshakeServer("connect"); + configureProxyEnvironment(proxy.url); + if (managed) { + vi.stubEnv("OPENCLAW_PROXY_ACTIVE", "1"); + } + resolveProviderToLoopback(); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: false, + baseUrl: "wss://provider.example/audio", + dispatcherPolicy, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: "wss://provider.example/audio", + }); + socket.on("error", () => {}); + const peer = await withTestTimeout(proxy.received, 2000, "Proxy did not receive CONNECT"); + const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) }); + socket.terminate(); + await closed; + expect(proxy.connectionCount()).toBe(1); + }); + + it.each([ + { name: "ambient", managed: false, dispatcherPolicy: undefined }, + { name: "managed", managed: true, dispatcherPolicy: { mode: "env-proxy" } }, + ] as const)( + "checks target DNS when NO_PROXY bypasses a $name proxy", + async ({ managed, dispatcherPolicy }) => { + const proxy = await createStalledHandshakeServer("connect"); + configureProxyEnvironment(proxy.url, "provider.example"); + if (managed) { + vi.stubEnv("OPENCLAW_PROXY_ACTIVE", "1"); + } + resolveProviderToLoopback(); + await expect( + openProviderWebSocket({ + allowPrivateNetwork: false, + baseUrl: "wss://provider.example/audio", + dispatcherPolicy, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: "wss://provider.example/audio", + }), + ).rejects.toThrow(/private|loopback|blocked/iu); + expect(proxy.connectionCount()).toBe(0); + }, + ); + + it("checks target DNS when only ALL_PROXY is configured", async () => { + const proxy = await createStalledHandshakeServer("connect"); + configureProxyEnvironment(proxy.url); + for (const name of ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]) { + vi.stubEnv(name, undefined); + } + resolveProviderToLoopback(); + const opening = openProviderWebSocket({ + allowPrivateNetwork: false, + baseUrl: "wss://provider.example/audio", + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: "wss://provider.example/audio", + }).then((socket) => { + socket.on("error", () => {}); + socket.terminate(); + return socket; + }); + await expect(opening).rejects.toThrow(/private|loopback|blocked/iu); + expect(proxy.connectionCount()).toBe(0); + }); + + it("keeps DNS preparation and the handshake within one connection deadline", async () => { + const server = await createStalledHandshakeServer("upgrade"); + const lookupStarted = createDeferred(); + const releaseLookup = createDeferred(); + const resolveHostname = ssrf.resolvePinnedHostnameWithPolicy; + vi.spyOn(ssrf, "resolvePinnedHostnameWithPolicy").mockImplementationOnce(async (...args) => { + lookupStarted.resolve(); + await releaseLookup.promise; + return await resolveHostname(...args); + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const url = server.url.replace("http:", "ws:"); + const opening = openProviderWebSocket({ + allowPrivateNetwork: true, + baseUrl: url, + dispatcherPolicy: { mode: "direct" }, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url, + }); + await lookupStarted.promise; + await vi.advanceTimersByTimeAsync(600); + releaseLookup.resolve(); + const socket = await opening; + const errors: Error[] = []; + socket.on("error", (error) => errors.push(error)); + const peer = await server.received; + const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) }).then( + () => "closed", + (error: unknown) => error, + ); + await vi.advanceTimersByTimeAsync(399); + expect(socket.readyState).toBe(socket.CONNECTING); + await vi.advanceTimersByTimeAsync(1); + expect(errors).toHaveLength(1); + expect(await closed).toBe("closed"); + }); + + it.each(["deadline", "caller cancellation", "socket termination"] as const)( + "closes a stalled proxy CONNECT on %s", + async (stop) => { + const proxy = await createStalledHandshakeServer("connect"); + const controller = new AbortController(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: true, + baseUrl: "wss://127.0.0.1/audio", + dispatcherPolicy: { mode: "explicit-proxy", proxyUrl: proxy.url }, + signal: controller.signal, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: "wss://127.0.0.1/audio", + }); + socket.on("error", () => {}); + const peer = await proxy.received; + const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) }).then( + () => "closed", + (error: unknown) => error, + ); + if (stop === "deadline") { + await vi.advanceTimersByTimeAsync(1000); + } else if (stop === "caller cancellation") { + controller.abort(); + } else { + socket.terminate(); + } + expect(socket.readyState).not.toBe(socket.CONNECTING); + expect(await closed).toBe("closed"); + }, + ); + + it.each([ + { name: "private", allowPrivateNetwork: false }, + { name: "pre-aborted", allowPrivateNetwork: true, signal: AbortSignal.abort() }, + ])("does not open a $name socket", async (testCase) => { + const server = await createLocalWebSocketServer(); + await expect( + openProviderWebSocket({ + allowPrivateNetwork: testCase.allowPrivateNetwork, + baseUrl: server.url, + headers: { authorization: "Token test" }, + ...(testCase.signal ? { signal: testCase.signal } : {}), + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: server.url, + }), + ).rejects.toThrow(/private|loopback|blocked|aborted/iu); + expect(server.requestHeaders).toHaveLength(0); + }); + + it("opens an allowed socket with resolved request headers", async () => { + const server = await createLocalWebSocketServer(); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: true, + baseUrl: server.url, + headers: { authorization: "Token configured", "x-provider": "deepgram" }, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: server.url, + }); + await once(socket, "open"); + expect(server.requestHeaders[0]?.authorization).toBe("Token configured"); + expect(server.requestHeaders[0]?.["x-provider"]).toBe("deepgram"); + socket.close(); + await once(socket, "close"); + }); + + it("applies target TLS settings to the WebSocket handshake", async () => { + const server = await createLocalWebSocketServer({ tls: true }); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: true, + baseUrl: server.url, + dispatcherPolicy: { mode: "direct", connect: { rejectUnauthorized: false } }, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: server.url, + }); + await once(socket, "open"); + expect(server.requestHeaders).toHaveLength(1); + socket.close(); + await once(socket, "close"); + }); + + it("routes the WebSocket handshake through an explicit proxy", async () => { + const target = await createLocalWebSocketServer(); + const connectSpy = vi.spyOn(net, "connect"); + const proxy = await createConnectProxy("localhost"); + const socket = await openProviderWebSocket({ + allowPrivateNetwork: true, + baseUrl: target.url, + dispatcherPolicy: { mode: "explicit-proxy", proxyUrl: proxy.url }, + timeoutMs: 1000, + trustConfiguredBaseUrlOrigin: false, + url: target.url, + }); + await once(socket, "open"); + expect(proxy.connectCount()).toBe(1); + expect(target.requestHeaders).toHaveLength(1); + expect( + connectSpy.mock.calls.some( + ([options]) => typeof asOptionalRecord(options)?.lookup === "function", + ), + ).toBe(true); + socket.close(); + await once(socket, "close"); + }); +}); diff --git a/src/infra/net/provider-websocket.ts b/src/infra/net/provider-websocket.ts new file mode 100644 index 000000000000..bee1daa3e304 --- /dev/null +++ b/src/infra/net/provider-websocket.ts @@ -0,0 +1,211 @@ +// Provider WebSocket connector applies the same auth, proxy, TLS, and SSRF policy as provider HTTP. +import http from "node:http"; +import type { Agent as HttpAgent } from "node:http"; +import https from "node:https"; +import WebSocket from "ws"; +import { resolveProviderTransportSsrFPolicy } from "../../agents/provider-transport-fetch.js"; +import { buildTimeoutAbortSignal } from "../../utils/fetch-timeout.js"; +import { racePromiseWithAbortSignal } from "../abort-signal.js"; +import { isManagedProxyActive } from "./fetch-guard.js"; +import { createNodeProxyAgent, resolveEnvNodeProxyUrlForTarget } from "./node-proxy-agent.js"; +import { shouldUseEnvHttpProxyForUrl } from "./proxy-env.js"; +import { resolveActiveManagedProxyTlsOptions } from "./proxy/active-managed-proxy-tls.js"; +import { + assertHostnameAllowedWithPolicy, + resolvePinnedHostnameWithPolicy, + type PinnedDispatcherPolicy, + type SsrFPolicy, +} from "./ssrf.js"; + +const DEFAULT_PROVIDER_WEBSOCKET_MAX_PAYLOAD_BYTES = 16 * 1024 * 1024; + +type OpenProviderWebSocketParams = { + allowPrivateNetwork: boolean; + baseUrl: string; + dispatcherPolicy?: PinnedDispatcherPolicy; + headers?: HeadersInit; + maxPayloadBytes?: number; + signal?: AbortSignal; + timeoutMs: number; + trustConfiguredBaseUrlOrigin: boolean; + url: string; +}; + +function toHttpUrl(value: string): string { + const url = new URL(value); + if (url.protocol === "ws:") { + url.protocol = "http:"; + } else if (url.protocol === "wss:") { + url.protocol = "https:"; + } + return url.toString(); +} + +function targetTlsOptions(policy: PinnedDispatcherPolicy | undefined): Record { + return policy?.mode === "direct" || policy?.mode === "env-proxy" ? { ...policy.connect } : {}; +} + +function proxyPolicy( + policy: SsrFPolicy | undefined, + allowPrivateProxy: boolean, +): SsrFPolicy | undefined { + if (!policy && !allowPrivateProxy) { + return undefined; + } + return { + ...policy, + hostnameAllowlist: undefined, + ...(allowPrivateProxy ? { allowPrivateNetwork: true } : {}), + }; +} + +async function createProxyAgent(params: { + policy: SsrFPolicy | undefined; + proxyUrl: URL; + proxyTls?: Record; + allowPrivateProxy: boolean; + signal?: AbortSignal; +}): Promise { + const pinnedProxy = await resolvePinnedHostnameWithPolicy(params.proxyUrl.hostname, { + policy: proxyPolicy(params.policy, params.allowPrivateProxy), + signal: params.signal, + }); + return createNodeProxyAgent({ + mode: "explicit", + proxyUrl: params.proxyUrl, + proxyConnect: { ...params.proxyTls, lookup: pinnedProxy.lookup }, + }); +} + +async function createProviderWebSocketAgent(params: { + dispatcherPolicy?: PinnedDispatcherPolicy; + policy: SsrFPolicy | undefined; + url: URL; + signal?: AbortSignal; +}): Promise { + const { dispatcherPolicy, policy, url, signal } = params; + const canDelegateEnvDns = shouldUseEnvHttpProxyForUrl(toHttpUrl(url.href)); + const useManagedProxy = isManagedProxyActive() && canDelegateEnvDns; + const envProxyUrl = + useManagedProxy || dispatcherPolicy?.mode !== "direct" + ? resolveEnvNodeProxyUrlForTarget(url) + : undefined; + let proxyUrl: URL | undefined; + if (dispatcherPolicy?.mode === "explicit-proxy") { + try { + proxyUrl = new URL(dispatcherPolicy.proxyUrl); + } catch { + throw new Error("Invalid explicit proxy URL"); + } + if (proxyUrl.protocol !== "http:" && proxyUrl.protocol !== "https:") { + throw new Error("Explicit proxy URL must use http or https"); + } + } else if (dispatcherPolicy?.mode !== "direct") { + proxyUrl = envProxyUrl; + } + if (useManagedProxy) { + proxyUrl = envProxyUrl; + } + + if (!proxyUrl) { + const pinned = await resolvePinnedHostnameWithPolicy(url.hostname, { policy, signal }); + const options = { + keepAlive: false, + ...targetTlsOptions(dispatcherPolicy), + lookup: pinned.lookup, + }; + return url.protocol === "wss:" ? new https.Agent(options) : new http.Agent(options); + } + + // Match guarded HTTP: configured proxies stay strict, while applicable + // managed/ambient HTTP proxy routes own DNS. ALL_PROXY alone grants no trust. + if (!useManagedProxy && (dispatcherPolicy || !canDelegateEnvDns)) { + await resolvePinnedHostnameWithPolicy(url.hostname, { policy, signal }); + } else { + assertHostnameAllowedWithPolicy(url.hostname, policy); + } + return await createProxyAgent({ + policy, + proxyUrl, + proxyTls: + !useManagedProxy && + (dispatcherPolicy?.mode === "explicit-proxy" || dispatcherPolicy?.mode === "env-proxy") + ? dispatcherPolicy.proxyTls + : resolveActiveManagedProxyTlsOptions({ proxyUrl: proxyUrl.href }), + allowPrivateProxy: + useManagedProxy || + dispatcherPolicy?.mode !== "explicit-proxy" || + dispatcherPolicy.allowPrivateProxy === true, + signal, + }); +} + +/** Opens a provider WebSocket through the resolved request and network policy. */ +export async function openProviderWebSocket( + params: OpenProviderWebSocketParams, +): Promise { + let url: URL; + try { + url = new URL(params.url); + } catch { + throw new Error("Invalid provider WebSocket URL"); + } + if (url.protocol !== "ws:" && url.protocol !== "wss:") { + throw new Error("Provider WebSocket URL must use ws or wss"); + } + const policy = resolveProviderTransportSsrFPolicy({ + baseUrl: toHttpUrl(params.baseUrl), + url: toHttpUrl(url.toString()), + allowPrivateNetwork: params.allowPrivateNetwork, + trustConfiguredBaseUrlOrigin: params.trustConfiguredBaseUrlOrigin, + }); + // DNS preparation and the opening handshake share one deadline. Proxyline + // owns pending proxy sockets and closes them when the request or agent ends. + const { signal, cleanup } = buildTimeoutAbortSignal({ + signal: params.signal, + timeoutMs: Math.max(1, params.timeoutMs), + operation: "Provider WebSocket connection", + }); + let agent: HttpAgent; + try { + signal?.throwIfAborted(); + const pending = createProviderWebSocketAgent({ + dispatcherPolicy: params.dispatcherPolicy, + policy, + url, + signal, + }); + void pending.then( + (resolved) => signal?.aborted && resolved.destroy(), + () => undefined, + ); + agent = await racePromiseWithAbortSignal(pending, signal); + } catch (error) { + cleanup(); + throw error; + } + let socket: WebSocket; + try { + signal?.throwIfAborted(); + socket = new WebSocket(url, { + agent, + headers: Object.fromEntries(new Headers(params.headers).entries()), + maxPayload: params.maxPayloadBytes ?? DEFAULT_PROVIDER_WEBSOCKET_MAX_PAYLOAD_BYTES, + perMessageDeflate: false, + ...targetTlsOptions(params.dispatcherPolicy), + }); + } catch (error) { + cleanup(); + agent.destroy(); + throw error; + } + const onAbort = () => socket.terminate(); + signal?.addEventListener("abort", onAbort, { once: true }); + socket.once("open", cleanup); + socket.once("close", () => { + signal?.removeEventListener("abort", onAbort); + cleanup(); + agent.destroy(); + }); + return socket; +} diff --git a/src/plugin-sdk/provider-http.ts b/src/plugin-sdk/provider-http.ts index ee7d5d8c250f..656c4e426df5 100644 --- a/src/plugin-sdk/provider-http.ts +++ b/src/plugin-sdk/provider-http.ts @@ -42,11 +42,13 @@ export { postTranscriptionRequest, resolveProviderOperationTimeoutMs, resolveProviderHttpRequestConfig, + resolveProviderHttpRequestConfigWithOriginTrust, resolveAudioTranscriptionUploadFileName, requireTranscriptionText, sanitizeConfiguredModelProviderRequest, waitProviderOperationPollInterval, } from "../media-understanding/shared.js"; +export { openProviderWebSocket } from "../infra/net/provider-websocket.js"; export type { ProviderOperationDeadline, ProviderOperationTimeoutMs, diff --git a/src/plugin-sdk/test-helpers/provider-http-mocks.ts b/src/plugin-sdk/test-helpers/provider-http-mocks.ts index 74c2906d03c7..6808aa9e19e2 100644 --- a/src/plugin-sdk/test-helpers/provider-http-mocks.ts +++ b/src/plugin-sdk/test-helpers/provider-http-mocks.ts @@ -328,14 +328,28 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importActual) => ({ pollProviderOperationJson: providerHttpMocks.pollProviderOperationJsonMock, postJsonRequest: providerHttpMocks.postJsonRequestMock, postMultipartRequest: providerHttpMocks.postMultipartRequestMock, + postTranscriptionRequest: ( + await importActual() + ).postTranscriptionRequest, providerOperationRetryConfig: (_stage: string) => true, readProviderBinaryResponse: ( await importActual() ).readProviderBinaryResponse, readProviderJsonResponse: providerHttpMocks.readProviderJsonResponseMock, + readProviderJsonObjectResponse: ( + await importActual() + ).readProviderJsonObjectResponse, + requireTranscriptionText: ( + await importActual() + ).requireTranscriptionText, resolveProviderOperationTimeoutMs: ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) => defaultTimeoutMs, resolveProviderHttpRequestConfig: providerHttpMocks.resolveProviderHttpRequestConfigMock, + resolveProviderHttpRequestConfigWithOriginTrust: ( + await importActual() + ).resolveProviderHttpRequestConfigWithOriginTrust, + openProviderWebSocket: (await importActual()) + .openProviderWebSocket, resolveProviderRequestHeaders: providerHttpMocks.resolveProviderRequestHeadersMock, [providerHttpMockKeys.sanitizeConfiguredModelProviderRequest]: providerHttpMocks.sanitizeConfiguredModelProviderRequestMock,