From 08d25e37bfefeaa0b900630269c7b6f3b837fdeb Mon Sep 17 00:00:00 2001 From: Clive Rosfield <64878945+S0RYUASUKA@users.noreply.github.com> Date: Sat, 5 Sep 2026 00:02:29 +0800 Subject: [PATCH] fix(responses): honor proxy routing for upstream websocket --- .../src/content/docs/guides/providers.md | 6 + .../src/content/docs/reference/adapters.md | 5 + .../content/docs/reference/proxy-formats.md | 13 ++ src/config.ts | 11 +- src/lib/provider-outbound.ts | 47 +----- src/lib/proxy-env.ts | 67 +++++++++ src/server/responses/codex-ws-pool.ts | 8 +- src/server/responses/codex-ws-session.ts | 4 +- src/server/responses/ws-upstream.ts | 11 +- structure/04_transports-and-sidecars.md | 8 +- tests/responses/ws-upstream-reuse.test.ts | 29 +++- tests/responses/ws-upstream.test.ts | 82 +++++++++-- tests/server/proxy-env.test.ts | 135 +++++++++++++++++- 13 files changed, 350 insertions(+), 76 deletions(-) diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index 6a37cf8a70..255c0d8dc4 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -620,6 +620,12 @@ A provider is included when opencodex has a matching wire adapter, **not** based (AI Studio, Vertex, and Antigravity/Cloud Code Assist modes), `azure` / `azure-openai`, `kiro`, and `cursor`. A proprietary API without one of these implementations, such as native Amazon Bedrock, is not supported directly. + +Provider configuration selects the adapter; upstream transport selection is separate. Eligible +Responses traffic can use WSS with [explicit proxy routing](/reference/proxy-formats/#json-and-sse-output). +Invalid or unsupported WebSocket proxy settings fall back to HTTP/SSE, which uses Bun's HTTP +proxy rules rather than the WSS-specific `ALL_PROXY` fallback. + **GitHub Copilot** is an OAuth provider (`ocx login github-copilot`) that exchanges a GitHub device-flow login for a short-lived Copilot API token — not a pasted API key. **GitLab Duo** remains a key/subscription-token gateway on its OpenAI-compatible endpoint. **Cloudflare AI diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 1db98357d3..e2a24c67df 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -95,6 +95,11 @@ body and response, with narrow compatibility rewrites for routed gateways. `forward` uses configured static headers without relaying caller authorization; `key` uses the configured provider key. +Adapter selection does not select the upstream transport. Eligible requests can use the +[upstream WebSocket proxy route](/reference/proxy-formats/#json-and-sse-output); invalid or unsupported +WebSocket proxy settings fall back to HTTP/SSE. HTTP fetch-based Responses handling uses Bun's +HTTP proxy rules and does not inherit the WSS-specific `ALL_PROXY` fallback. + Noncanonical Responses gateways receive Codex's client-executed `tool_search` declaration as a collision-safe public function tool. Matching request history and JSON/SSE function calls are translated back to the private `tool_search` lifecycle for the client. Canonical OpenAI forward diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 77a67147ac..b4d7e5dead 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -113,6 +113,19 @@ the raw JSON frame and its SSE envelope at 4 MiB, and closes the upstream when i would overflow. That overflow emits a terminal downstream `response.failed` event followed by `[DONE]`. +The upstream WebSocket checks `NO_PROXY`/`no_proxy` first. Otherwise it uses the first non-empty +`HTTPS_PROXY`, `https_proxy`, `ALL_PROXY`, or `all_proxy` value; `HTTP_PROXY` alone does not proxy a +WSS connection. HTTP and HTTPS proxy URLs are passed to Bun. If the selected value is invalid or +uses an unsupported protocol, opencodex skips the WebSocket attempt and uses HTTP/SSE instead of +dialing the upstream directly. + +These rules belong to the upstream WebSocket transport, independently of the selected provider +adapter. HTTP fetch-based Responses requests, including SSE fallback, use Bun's HTTP proxy rules +and do not use `ALL_PROXY`. `config.proxy` fills missing `HTTP_PROXY`/`HTTPS_PROXY` values; the +resulting scheme-specific value also takes precedence over an existing `ALL_PROXY` for WebSocket. +For an HTTPS upstream that requires a proxy, set `HTTPS_PROXY` or `config.proxy`; `HTTP_PROXY` +alone leaves both WSS and its HTTPS fallback without a scheme-matched proxy. + Every terminal Responses usage object includes both detail objects, even when the provider did not report those details: diff --git a/src/config.ts b/src/config.ts index 5d67275dce..72da455382 100644 --- a/src/config.ts +++ b/src/config.ts @@ -3738,11 +3738,12 @@ function warnProxyConfigDiscardOnce(kind: "proxy" | "noProxy" | "noProxyElements } /** - * Mirror `config.proxy` into HTTP(S)_PROXY env vars so Bun's native fetch routes every outbound - * provider call through the proxy — no per-callsite changes (verified: Bun honors these plus - * NO_PROXY). User-set env vars always win; localhost/127.0.0.1 are appended to NO_PROXY so the - * CLI's own health checks and running-proxy API calls stay direct. Call once per process entry - * that makes outbound provider requests (server start, catalog sync). + * Mirror `config.proxy` into HTTP(S)_PROXY env vars. Bun fetch consumes them natively; transports + * such as the ChatGPT upstream WebSocket select the same environment explicitly. User-set HTTP(S)_PROXY + * variables win; config fills missing scheme proxies, which take precedence over ALL_PROXY for WS. + * localhost/127.0.0.1 are appended to NO_PROXY so the CLI's own health checks and + * running-proxy API calls stay direct. Call once per process entry that makes outbound provider + * requests (server start, catalog sync). */ export function applyProxyEnv(config: OcxConfig): void { applyProxyEnvWith(config); diff --git a/src/lib/provider-outbound.ts b/src/lib/provider-outbound.ts index 495fef0b8b..02bdbc2077 100644 --- a/src/lib/provider-outbound.ts +++ b/src/lib/provider-outbound.ts @@ -7,7 +7,7 @@ import { resolvePublicAddresses, } from "./destination-policy"; import { pinnedHttpGet, pinnedHttpPost } from "./pinned-http"; -import { effectiveProxyFor, outboundProxyConfigured } from "./proxy-env"; +import { effectiveProxyFor, noProxyMatches, normalizeProxyHostname, outboundProxyConfigured } from "./proxy-env"; import { publicProviderBaseUrl } from "./provider-url"; type ProviderGetInit = Omit; @@ -37,10 +37,6 @@ function pickPinnedAddress(addresses: Array<{ address: string; family: number }> return addresses.find(address => address.family === 4) ?? addresses[0]!; } -function configuredProxyFor(): boolean { - return outboundProxyConfigured(); -} - /** * Registry-owned fake-IP transparency exception (Clash/Surge/Mihomo TUN mode). * @@ -76,45 +72,6 @@ function transparentFakeIpException( return isCanonicalUrl(name, url); } -function normalizeProxyHostname(hostname: string): string { - const normalized = hostname.trim().toLowerCase().replace(/\.+$/, ""); - return normalized.startsWith("[") && normalized.endsWith("]") - ? normalized.slice(1, -1) - : normalized; -} - -function noProxyMatches(url: URL): boolean { - const raw = process.env.NO_PROXY ?? process.env.no_proxy ?? ""; - const hostname = normalizeProxyHostname(url.hostname); - const port = url.port || (url.protocol === "https:" ? "443" : "80"); - for (const rawEntry of raw.split(",")) { - let entry = rawEntry.trim().toLowerCase(); - if (!entry) continue; - if (entry === "*") return true; - entry = entry.replace(/^https?:\/\//, "").split("/", 1)[0]!; - - let entryHost = entry; - let entryPort = ""; - const bracketed = /^\[([^\]]+)](?::(\d+))?$/.exec(entry); - if (bracketed) { - entryHost = bracketed[1]!; - entryPort = bracketed[2] ?? ""; - } else if ((entry.match(/:/g)?.length ?? 0) === 1) { - const separator = entry.lastIndexOf(":"); - const possiblePort = entry.slice(separator + 1); - if (/^\d+$/.test(possiblePort)) { - entryHost = entry.slice(0, separator); - entryPort = possiblePort; - } - } - if (entryPort && entryPort !== port) continue; - entryHost = normalizeProxyHostname(entryHost.replace(/^\*?\./, "")); - if (!entryHost) continue; - if (hostname === entryHost || hostname.endsWith(`.${entryHost}`)) return true; - } - return false; -} - let proxyBoundaryWarned = false; let proxyDnsDegradationWarned = false; @@ -181,7 +138,7 @@ async function providerOutboundRequest( return provider.fetch(url, { ...init, method, redirect: "manual" }); } const parsed = postUrl ?? new URL(url); - const proxyConfigured = configuredProxyFor(); + const proxyConfigured = outboundProxyConfigured(); // Snapshot the scheme-matched proxy once, before the DNS await, so admission and transport // below reason about the same value. `null` here means "no proxy fetch would actually use", // even if some other proxy variable is set. diff --git a/src/lib/proxy-env.ts b/src/lib/proxy-env.ts index 46df592689..0ac9ed735c 100644 --- a/src/lib/proxy-env.ts +++ b/src/lib/proxy-env.ts @@ -3,6 +3,73 @@ export const PROXY_ENV_KEYS = [...OUTBOUND_PROXY_ENV_KEYS, "NO_PROXY"] as const; export type ProxyEnvKey = typeof PROXY_ENV_KEYS[number]; export type ProxyEnvMap = Record; +export type ProxyRoute = + | { kind: "direct" } + | { kind: "proxy"; proxy: string } + | { kind: "fallback" }; + +export function normalizeProxyHostname(hostname: string): string { + const normalized = hostname.trim().toLowerCase().replace(/\.+$/, ""); + return normalized.startsWith("[") && normalized.endsWith("]") + ? normalized.slice(1, -1) + : normalized; +} + +export function noProxyMatches( + url: URL, + env: ProxyEnvMap = process.env, +): boolean { + const raw = env.NO_PROXY ?? env.no_proxy ?? ""; + const hostname = normalizeProxyHostname(url.hostname); + const port = url.port || (url.protocol === "https:" || url.protocol === "wss:" ? "443" : "80"); + for (const rawEntry of raw.split(",")) { + let entry = rawEntry.trim().toLowerCase(); + if (!entry) continue; + if (entry === "*") return true; + entry = entry.replace(/^(?:https?|wss?):\/\//, "").split("/", 1)[0]!; + + let entryHost = entry; + let entryPort = ""; + const bracketed = /^\[([^\]]+)](?::(\d+))?$/.exec(entry); + if (bracketed) { + entryHost = bracketed[1]!; + entryPort = bracketed[2] ?? ""; + } else if ((entry.match(/:/g)?.length ?? 0) === 1) { + const separator = entry.lastIndexOf(":"); + const possiblePort = entry.slice(separator + 1); + if (/^\d+$/.test(possiblePort)) { + entryHost = entry.slice(0, separator); + entryPort = possiblePort; + } + } + if (entryPort && entryPort !== port) continue; + entryHost = normalizeProxyHostname(entryHost.replace(/^\*?\./, "")); + if (entryHost && (hostname === entryHost || hostname.endsWith(`.${entryHost}`))) return true; + } + return false; +} + +export function resolveProxyRoute( + url: URL, + env: ProxyEnvMap = process.env, +): ProxyRoute { + if (noProxyMatches(url, env)) return { kind: "direct" }; + const key = url.protocol === "https:" || url.protocol === "wss:" + ? "HTTPS_PROXY" + : "HTTP_PROXY"; + const proxy = [key, key.toLowerCase(), "ALL_PROXY", "all_proxy"] + .map(candidate => env[candidate]?.trim()) + .find(Boolean); + if (!proxy) return { kind: "direct" }; + try { + const protocol = new URL(proxy).protocol; + return protocol === "http:" || protocol === "https:" + ? { kind: "proxy", proxy } + : { kind: "fallback" }; + } catch { + return { kind: "fallback" }; + } +} export function proxyEnvPresent( key: ProxyEnvKey, diff --git a/src/server/responses/codex-ws-pool.ts b/src/server/responses/codex-ws-pool.ts index 378cf2d4a3..5d406bee4f 100644 --- a/src/server/responses/codex-ws-pool.ts +++ b/src/server/responses/codex-ws-pool.ts @@ -25,7 +25,7 @@ function digest(input: unknown): string { } /** Identity comes from the selected outgoing request, never a model label or caller hint. */ -export function codexWsReuseIdentity(url: string, headers: Record, frameText: string): CodexWsReuseIdentity | null { +export function codexWsReuseIdentity(url: string, headers: Record, frameText: string, proxy?: string): CodexWsReuseIdentity | null { if (url !== CODEX_RESPONSES_HTTP_URL) return null; let body: unknown; try { body = JSON.parse(frameText); } catch { return null; } @@ -52,7 +52,7 @@ export function codexWsReuseIdentity(url: string, headers: Record): CodexWsSession | null { + acquire(identity: CodexWsReuseIdentity, url: string, headers: Record, proxy?: string): CodexWsSession | null { this.sweep(); for (const entry of this.entries.values()) { if (entry.identity.scope !== identity.scope || entry.identity.key === identity.key) continue; @@ -94,7 +94,7 @@ export class CodexWsPool { this.remove(oldest); } const createdAt = this.now(); - const session = new CodexWsSession(url, headers, true, () => this.changed(entry)); + const session = new CodexWsSession(url, headers, true, () => this.changed(entry), proxy); const entry: Entry = { identity, session, createdAt, idleAt: createdAt, retired: false }; session.reserve(); this.entries.set(identity.key, entry); diff --git a/src/server/responses/codex-ws-session.ts b/src/server/responses/codex-ws-session.ts index bbf62f8137..32716a5297 100644 --- a/src/server/responses/codex-ws-session.ts +++ b/src/server/responses/codex-ws-session.ts @@ -10,8 +10,8 @@ export class CodexWsSession { private readonly completedIds = new Set(); constructor(url: string, headers: Record, readonly retainable = false, - private readonly changed: () => void = () => {}) { - this.socket = new WebSocket(url, { headers } as unknown as string[]); + private readonly changed: () => void = () => {}, proxy?: string) { + this.socket = new WebSocket(url, { headers, ...(proxy ? { proxy } : {}) } as unknown as string[]); this.socket.addEventListener("open", this.onOpen); this.socket.addEventListener("message", this.onIdleMessage); this.socket.addEventListener("close", this.onClose); diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index e9773d02a3..87b3767d2b 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -13,6 +13,7 @@ // (passthrough relay, adapter parsers, usage sniffing) is unchanged. import { compareBunVersions } from "../../lib/bun-stream-caps"; +import { resolveProxyRoute } from "../../lib/proxy-env"; import type { CodexWsQuotaObserver } from "./codex-ws-metadata"; import { CODEX_RESPONSES_HTTP_URL, CODEX_RESPONSES_WS_URL, prepareCodexHttpInit, prepareCodexWsRequest } from "./codex-ws-request"; import { codexWsExchange } from "./codex-ws-exchange"; @@ -150,6 +151,10 @@ export function codexWsUpstreamFetch( return sseFallback(url, init); } + const wsUrl = wsUpstreamUrlFor(url); + const proxyRoute = resolveProxyRoute(new URL(wsUrl)); + if (proxyRoute.kind === "fallback") return sseFallback(url, init); + const proxy = proxyRoute.kind === "proxy" ? proxyRoute.proxy : undefined; // A genuine caller `originator` is already in these headers via the forward // set. Never fabricate one here: pool/forward traffic must not impersonate // Codex CLI, per the metadata-integrity contract. (The backend's fast lane @@ -164,9 +169,9 @@ export function codexWsUpstreamFetch( } let session: CodexWsSession; try { - const identity = codexWsReuseIdentity(url, headers, frameText); - session = (identity ? codexWsPool.acquire(identity, wsUpstreamUrlFor(url), headers) : null) - ?? new CodexWsSession(wsUpstreamUrlFor(url), headers); + const identity = codexWsReuseIdentity(url, headers, frameText, proxy); + session = (identity ? codexWsPool.acquire(identity, wsUrl, headers, proxy) : null) + ?? new CodexWsSession(wsUrl, headers, false, undefined, proxy); if (!session.busy && !session.reserve()) { session.dispose(); return sseFallback(url, init); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 4ee22c114c..a45a98c87b 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -435,7 +435,7 @@ These are transport-fidelity guarantees, not a provider-billing guarantee. Eligible complete-input creates can retain a canonical upstream socket within one selected account, credential, thread and turn. Model/tier and immutable -handshake headers must also match. Turn-state and turn-metadata headers are +handshake headers and the selected outbound proxy must also match. Turn-state and turn-metadata headers are projected into their same-name per-frame metadata slots; explicit body values win. The pool retains at most 32 sockets, expires idle sockets after 30 seconds, and retires a socket after five minutes or 32 successful exchanges (after active work @@ -644,7 +644,11 @@ the upgrade with 426 so Codex falls back to HTTP cleanly. That setting controls the client-facing upgrade only. The transparent upstream ChatGPT WS optimization described above is selected independently and still -returns the same downstream SSE contract. +returns the same downstream SSE contract. Its WSS route checks NO_PROXY first, then selects the +first non-empty HTTPS_PROXY, https_proxy, ALL_PROXY, or all_proxy value. HTTP_PROXY alone does not +route WSS. Unsupported or malformed selected proxy values skip the WebSocket attempt and use the +existing SSE path immediately; they never fall through to a lower-priority proxy or direct WebSocket +egress. HTTP/SSE fallback retains Bun fetch's own proxy rules, which do not consult ALL_PROXY. The endpoint handles `response.create`, ignores `response.processed`, supports warmup `generate: false`, and feeds the same request pipeline as HTTP/SSE. diff --git a/tests/responses/ws-upstream-reuse.test.ts b/tests/responses/ws-upstream-reuse.test.ts index fd0a8fb5a1..b957fdb317 100644 --- a/tests/responses/ws-upstream-reuse.test.ts +++ b/tests/responses/ws-upstream-reuse.test.ts @@ -6,6 +6,8 @@ import { prepareCodexWsRequest } from "../../src/server/responses/codex-ws-reque const URL = "https://chatgpt.com/backend-api/codex/responses"; const realWebSocket = globalThis.WebSocket; +const proxyEnvKeys = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"]; +let savedProxyEnv: Record; let sequence = 0; class Socket extends EventTarget { @@ -13,7 +15,7 @@ class Socket extends EventTarget { static onSend: (socket: Socket, frame: Record) => void = (socket) => socket.complete(); readyState = 0; frames: Record[] = []; - constructor(readonly url: string) { + constructor(readonly url: string, readonly options?: { proxy?: string }) { super(); Socket.all.push(this); queueMicrotask(() => { if (this.readyState === 0) { this.readyState = 1; this.dispatchEvent(new Event("open")); } }); @@ -58,7 +60,11 @@ function bodyWith(fields: Record) { options.body = JSON.stringify({ ...JSON.parse(options.body as string), ...fields }); return options; } -beforeEach(() => { globalThis.WebSocket = Socket as unknown as typeof WebSocket; }); +beforeEach(() => { + globalThis.WebSocket = Socket as unknown as typeof WebSocket; + savedProxyEnv = Object.fromEntries(proxyEnvKeys.map(key => [key, process.env[key]])); + for (const key of proxyEnvKeys) delete process.env[key]; +}); afterEach(() => { runOptionalShutdownHooks(); @@ -67,6 +73,25 @@ afterEach(() => { Socket.onSend = socket => socket.complete(); sequence = 0; globalThis.WebSocket = realWebSocket; + for (const key of proxyEnvKeys) delete process.env[key]; + for (const key of proxyEnvKeys) { + if (savedProxyEnv[key] !== undefined) process.env[key] = savedProxyEnv[key]; + } +}); + +test("proxy changes and NO_PROXY retire the old route while unchanged routes reuse", async () => { + for (const proxy of ["http://proxy-a.example:8080", "http://proxy-b.example:8080"]) { + process.env.HTTPS_PROXY = proxy; + await drain(); + await drain(); + } + process.env.NO_PROXY = "chatgpt.com:443"; + await drain(); + await drain(); + expect(Socket.all.map(socket => socket.options?.proxy)) + .toEqual(["http://proxy-a.example:8080", "http://proxy-b.example:8080", undefined]); + expect(Socket.all.map(socket => socket.frames.length)).toEqual([2, 2, 2]); + expect(Socket.all.map(socket => socket.readyState)).toEqual([3, 3, 1]); }); test("same account/thread/turn reuses one socket without trimming either HTTP input", async () => { diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index fd09513070..cfb087a4bb 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, jest, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, jest, test } from "bun:test"; import { providerFetch } from "../../src/server/responses/fetch-helpers"; import { handleResponses } from "../../src/server/responses"; import { isEagerRelaySseResponse } from "../../src/server/relay"; @@ -162,18 +162,24 @@ describe("shouldUseCodexWsUpstream", () => { }); type Listener = (event: unknown) => void; +type FakeWebSocketOptions = { + headers?: Record; + proxy?: string; +}; /** Minimal scriptable stand-in for Bun's WebSocket. */ class FakeWebSocket { static instances: FakeWebSocket[] = []; static script: (ws: FakeWebSocket) => void = () => {}; url: string; + options?: FakeWebSocketOptions; sent: string[] = []; closed = false; listeners = new Map(); - constructor(url: string) { + constructor(url: string, options?: FakeWebSocketOptions) { this.url = url; + this.options = options; FakeWebSocket.instances.push(this); queueMicrotask(() => FakeWebSocket.script(this)); } @@ -205,12 +211,23 @@ class FakeWebSocket { const RealWebSocket = globalThis.WebSocket; const RealFetch = globalThis.fetch; +const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"] as const; +let savedProxyEnv: Record; + +beforeEach(() => { + savedProxyEnv = Object.fromEntries(PROXY_ENV_KEYS.map(key => [key, process.env[key]])); + for (const key of PROXY_ENV_KEYS) delete process.env[key]; +}); afterEach(() => { globalThis.WebSocket = RealWebSocket; globalThis.fetch = RealFetch; FakeWebSocket.instances = []; FakeWebSocket.script = () => {}; + for (const key of PROXY_ENV_KEYS) delete process.env[key]; + for (const key of PROXY_ENV_KEYS) { + if (savedProxyEnv[key] !== undefined) process.env[key] = savedProxyEnv[key]; + } }); function installFake(script: (ws: FakeWebSocket) => void) { @@ -525,6 +542,41 @@ describe("codexWsUpstreamFetch", () => { expect(text).not.toContain("must-not-leak"); }); + test("passes the selected proxy without changing handshake headers", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) }); + }); + + await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("fallback must not run"); + }) as unknown as typeof fetch); + + const options = FakeWebSocket.instances[0]!.options; + expect(options?.proxy).toBe("http://proxy.example:8080"); + expect(options?.headers?.authorization).toBe("Bearer test"); + expect(options?.headers?.["openai-beta"]).toContain("responses_websockets"); + expect(options?.headers?.["content-type"]).toBeUndefined(); + }); + + test.each([ + ["unsupported protocol", "socks5://proxy.example:1080"], + ["invalid URL", "not a proxy URL"], + ])("falls back once without dialing for an %s", async (_label, proxy) => { + process.env.HTTPS_PROXY = proxy; + const sentinel = new Response("sse-fallback"); + let fallbackCalls = 0; + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (async () => { + fallbackCalls += 1; + return sentinel; + }) as typeof fetch); + + expect(response).toBe(sentinel); + expect(fallbackCalls).toBe(1); + expect(FakeWebSocket.instances).toHaveLength(0); + }); + test("relays event frames as an SSE response and sends one response.create frame", async () => { installFake(ws => { ws.emit("open", {}); @@ -654,6 +706,7 @@ describe("codexWsUpstreamFetch", () => { }); test("falls back to the HTTP fetch when the upgrade is rejected before open", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; installFake(ws => ws.close()); const sentinel = new Response("sse-fallback", { status: 429 }); let fallbackCalls = 0; @@ -666,6 +719,7 @@ describe("codexWsUpstreamFetch", () => { expect(response).toBe(sentinel); expect(isCodexWsUpstreamResponse(response)).toBe(false); expect(fallbackCalls).toBe(1); + expect(FakeWebSocket.instances[0]!.options?.proxy).toBe("http://proxy.example:8080"); }); test("falls back to the HTTP fetch when the upgrade deadline elapses without open or close", async () => { @@ -800,15 +854,17 @@ describe("codexWsUpstreamFetch", () => { }); test("preserves caller headers on the handshake without fabricating an originator", async () => { - const seen: Record[] = []; + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + process.env.NO_PROXY = "chatgpt.com:443"; + const seen: FakeWebSocketOptions[] = []; FakeWebSocket.script = ws => { ws.emit("open", {}); ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) }); }; class HeaderCapturingWebSocket extends FakeWebSocket { - constructor(url: string, options?: { headers?: Record }) { - super(url); - seen.push(options?.headers ?? {}); + constructor(url: string, options?: FakeWebSocketOptions) { + super(url, options); + seen.push(options ?? {}); } } globalThis.WebSocket = HeaderCapturingWebSocket as unknown as typeof WebSocket; @@ -817,18 +873,19 @@ describe("codexWsUpstreamFetch", () => { await codexWsUpstreamFetch(CODEX_URL, streamingInit(), fallback); // Without a caller originator none is invented: pool/forward traffic must // not impersonate Codex CLI (metadata-integrity contract). - expect(seen[0].originator).toBeUndefined(); - expect(seen[0]["openai-beta"]).toContain("responses_websockets"); - expect(seen[0].authorization).toBe("Bearer test"); + expect(seen[0].proxy).toBeUndefined(); + expect(seen[0].headers?.originator).toBeUndefined(); + expect(seen[0].headers?.["openai-beta"]).toContain("responses_websockets"); + expect(seen[0].headers?.authorization).toBe("Bearer test"); // HTTP body-framing headers do not belong on a WS handshake. - expect(seen[0]["content-type"]).toBeUndefined(); + expect(seen[0].headers?.["content-type"]).toBeUndefined(); // A genuine caller originator is forwarded verbatim. await codexWsUpstreamFetch(CODEX_URL, { ...streamingInit(), headers: { ...streamingInit().headers as Record, originator: "codex_cli_rs" }, }, fallback); - expect(seen[1].originator).toBe("codex_cli_rs"); + expect(seen[1].headers?.originator).toBe("codex_cli_rs"); }); test("aborting before open rejects like an aborted fetch", async () => { @@ -1194,6 +1251,8 @@ describe("oversized Codex create frames", () => { }); test("dials the configured provider's own wss URL for an opt-in upstream", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + process.env.NO_PROXY = "sub2api.example.com:443"; installFake(ws => { ws.emit("open", {}); ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: { id: "r-ws" } }) }); @@ -1206,6 +1265,7 @@ describe("oversized Codex create frames", () => { ); expect(FakeWebSocket.instances).toHaveLength(1); expect(FakeWebSocket.instances[0]!.url).toBe("wss://sub2api.example.com/v1/responses"); + expect(FakeWebSocket.instances[0]!.options?.proxy).toBeUndefined(); expect(response.headers.get("content-type")).toContain("text/event-stream"); expect(await response.text()).toContain("response.completed"); }); diff --git a/tests/server/proxy-env.test.ts b/tests/server/proxy-env.test.ts index e43ad2d9be..c795c6cf2d 100644 --- a/tests/server/proxy-env.test.ts +++ b/tests/server/proxy-env.test.ts @@ -1,8 +1,10 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { createServer } from "node:http"; import { applyProxyEnv } from "../../src/config"; +import { resolveProxyRoute } from "../../src/lib/proxy-env"; import type { OcxConfig } from "../../src/types"; -const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy", "OCX_TEST_PROXY_REF", "OCX_TEST_NO_PROXY_REF"] as const; +const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy", "OCX_TEST_PROXY_REF", "OCX_TEST_NO_PROXY_REF"] as const; let saved: Record; beforeEach(() => { @@ -30,6 +32,128 @@ function configWithRawProxy(proxy: unknown, noProxy?: unknown): OcxConfig { return { proxy, noProxy, providers: {} } as unknown as OcxConfig; } +describe("resolveProxyRoute", () => { + test("wss uses HTTPS_PROXY and never HTTP_PROXY", () => { + const target = new URL("wss://chatgpt.com/backend-api/codex/responses"); + expect(resolveProxyRoute(target, { + HTTPS_PROXY: "http://secure-proxy.example:8443", + HTTP_PROXY: "http://plain-proxy.example:8080", + })).toEqual({ kind: "proxy", proxy: "http://secure-proxy.example:8443" }); + expect(resolveProxyRoute(target, { + HTTP_PROXY: "http://plain-proxy.example:8080", + })).toEqual({ kind: "direct" }); + }); + + test.each([ + ["exact host", "wss://chatgpt.com/path", "chatgpt.com", "direct"], + ["domain suffix", "wss://api.chatgpt.com/path", ".chatgpt.com", "direct"], + ["wildcard suffix", "wss://api.chatgpt.com/path", "*.chatgpt.com", "direct"], + ["wss default port", "wss://chatgpt.com/path", "chatgpt.com:443", "direct"], + ["ws default port", "ws://chatgpt.com/path", "chatgpt.com:80", "direct"], + ["port mismatch", "wss://chatgpt.com/path", "chatgpt.com:80", "proxy"], + ["bracketed IPv6", "wss://[2001:db8::1]/path", "[2001:db8::1]:443", "direct"], + ["URL-style entry", "wss://chatgpt.com/path", "https://chatgpt.com/ignored", "direct"], + ] as const)("honors NO_PROXY for %s", (_label, target, noProxy, expectedKind) => { + expect(resolveProxyRoute(new URL(target), { + HTTPS_PROXY: "http://secure-proxy.example:8443", + NO_PROXY: noProxy, + }).kind).toBe(expectedKind); + }); + + test("uses stable proxy precedence and fails closed on the first unusable proxy", () => { + const target = new URL("wss://chatgpt.com/backend-api/codex/responses"); + const route = (env: Record) => resolveProxyRoute(target, env); + expect([ + route({ HTTPS_PROXY: "http://upper-https:1", https_proxy: "http://lower-https:2", ALL_PROXY: "http://upper-all:3", all_proxy: "http://lower-all:4" }), + route({ HTTPS_PROXY: " ", https_proxy: "http://lower-https:2", ALL_PROXY: "http://upper-all:3" }), + route({ ALL_PROXY: "http://upper-all:3", all_proxy: "http://lower-all:4" }), + route({ all_proxy: "https://lower-all:4" }), + route({ HTTPS_PROXY: "socks5://unsupported:1080", ALL_PROXY: "http://must-not-win:3" }), + route({ HTTPS_PROXY: "not a proxy URL", ALL_PROXY: "http://must-not-win:3" }), + route({}), + ]).toEqual([ + { kind: "proxy", proxy: "http://upper-https:1" }, + { kind: "proxy", proxy: "http://lower-https:2" }, + { kind: "proxy", proxy: "http://upper-all:3" }, + { kind: "proxy", proxy: "https://lower-all:4" }, + { kind: "fallback" }, + { kind: "fallback" }, + { kind: "direct" }, + ]); + }); + + test("preserves uppercase NO_PROXY precedence when it is explicitly empty", () => { + expect(resolveProxyRoute(new URL("wss://chatgpt.com/path"), { + HTTPS_PROXY: "http://secure-proxy.example:8443", + NO_PROXY: "", + no_proxy: "chatgpt.com", + })).toEqual({ kind: "proxy", proxy: "http://secure-proxy.example:8443" }); + }); + + test("Bun WebSocket sends WSS through an HTTP CONNECT proxy", async () => { + let resolveConnect!: (target: string) => void; + const connected = new Promise(resolve => { resolveConnect = resolve; }); + const proxy = createServer(); + proxy.on("connect", (request, socket) => { + resolveConnect(request.url ?? ""); + socket.end("HTTP/1.1 502 Probe Complete\r\nContent-Length: 0\r\n\r\n"); + }); + await new Promise((resolve, reject) => { + proxy.once("error", reject); + proxy.listen(0, "127.0.0.1", resolve); + }); + const address = proxy.address(); + if (!address || typeof address === "string") throw new Error("proxy did not bind a TCP port"); + const socket = new WebSocket("wss://proxy-probe.invalid/backend-api/codex/responses", { + proxy: `http://127.0.0.1:${address.port}`, + } as unknown as string[]); + try { + expect(await Promise.race([ + connected, + new Promise((_, reject) => setTimeout(() => reject(new Error("CONNECT was not observed")), 5_000)), + ])).toBe("proxy-probe.invalid:443"); + } finally { + try { socket.close(); } catch { /* probe is already complete */ } + await new Promise(resolve => proxy.close(() => resolve())); + } + }, 10_000); + + test.skipIf(process.platform !== "win32")("Bun fetch honors NO_PROXY on Windows", async () => { + let providerRequests = 0; + let proxyRequests = 0; + const provider = createServer((_request, response) => { + providerRequests += 1; + response.end("direct"); + }); + const proxy = createServer((_request, response) => { + proxyRequests += 1; + response.end("proxied"); + }); + const listen = async (server: typeof provider): Promise => { + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("server did not bind a TCP port"); + return address.port; + }; + const [providerPort, proxyPort] = await Promise.all([listen(provider), listen(proxy)]); + process.env.HTTP_PROXY = `http://127.0.0.1:${proxyPort}`; + process.env.NO_PROXY = "127.0.0.1"; + try { + expect(await (await fetch(`http://127.0.0.1:${providerPort}/models`)).text()).toBe("direct"); + expect(providerRequests).toBe(1); + expect(proxyRequests).toBe(0); + } finally { + await Promise.all([ + new Promise(resolve => provider.close(() => resolve())), + new Promise(resolve => proxy.close(() => resolve())), + ]); + } + }); +}); + describe("applyProxyEnv with values the schema does not constrain", () => { test("warns once per discarded proxy setting without exposing its raw value", () => { const secret = "raw-proxy-credential-sentinel-2947"; @@ -122,6 +246,14 @@ describe("applyProxyEnv", () => { expect(process.env.HTTP_PROXY).toBe("http://proxy.corp:8080"); }); + test.each(["ALL_PROXY", "all_proxy"])("config fills a scheme proxy ahead of %s for WSS", key => { + process.env[key] = "http://fallback-proxy.example:8081"; + applyProxyEnv(configWithProxy("http://configured-proxy.example:8080")); + expect(process.env[key]).toBe("http://fallback-proxy.example:8081"); + expect(resolveProxyRoute(new URL("wss://chatgpt.com/backend-api/codex/responses"))) + .toEqual({ kind: "proxy", proxy: "http://configured-proxy.example:8080" }); + }); + test("appends loopback entries to an existing NO_PROXY without duplicating", () => { process.env.NO_PROXY = "internal.corp,localhost"; applyProxyEnv(configWithProxy("http://proxy.corp:8080")); @@ -217,4 +349,3 @@ describe("applyProxyEnv with proxy: \"auto\" (#1525)", () => { expect(process.env.HTTP_PROXY).toBeUndefined(); }); }); -