From b80afbb74f8368e003ce9d0129be84ca3c1eedc8 Mon Sep 17 00:00:00 2001 From: agisota Date: Mon, 10 Aug 2026 09:24:51 +0300 Subject: [PATCH] fix(proxy): isolate TLS sessions by account (#9837) Co-authored-by: Antigravity Agent (via Agisota) --- .env.example | 3 + open-sse/services/browserBackedChat.ts | 1 + open-sse/utils/proxyFetch.ts | 350 +++++++-- open-sse/utils/tlsClient.ts | 699 +++++++++++++---- scripts/build/assembleStandalone.mjs | 22 +- src/sse/handlers/chatHelpers.ts | 23 +- tests/unit/build-next-isolated.test.ts | 58 +- tests/unit/chat-helpers.test.ts | 93 +++ tests/unit/tls-proxy-context.test.ts | 989 +++++++++++++++++++++++++ 9 files changed, 2025 insertions(+), 213 deletions(-) create mode 100644 tests/unit/tls-proxy-context.test.ts diff --git a/.env.example b/.env.example index 2fbd80a321..f664d06964 100644 --- a/.env.example +++ b/.env.example @@ -644,6 +644,9 @@ NEXT_PUBLIC_ENABLE_SOCKS5_PROXY=true # Reduces risk of JA3/JA4 fingerprint-based blocking by providers (e.g., Google). # Used by: open-sse/executors — replaces Node.js default TLS fingerprint. # ENABLE_TLS_FINGERPRINT=true +# New proxied TLS routing requires an explicit, comma-separated provider allowlist. +# Direct TLS keeps its legacy behavior when this is unset. +# TLS_FINGERPRINT_PROVIDERS=codex,openai # Allow the Claude Turnstile Playwright browser context to ignore HTTPS certificate errors. # Only enable for local debugging or trusted MITM/corporate proxy environments. diff --git a/open-sse/services/browserBackedChat.ts b/open-sse/services/browserBackedChat.ts index 433da826fb..ec7d6b2915 100644 --- a/open-sse/services/browserBackedChat.ts +++ b/open-sse/services/browserBackedChat.ts @@ -517,6 +517,7 @@ export async function httpBackedChat( headers, body, signal: signal ?? undefined, + sessionScope: req.poolKey, }); const fetchMs = Date.now() - fetchStart; diff --git a/open-sse/utils/proxyFetch.ts b/open-sse/utils/proxyFetch.ts index a5080c6fbb..634f30735b 100644 --- a/open-sse/utils/proxyFetch.ts +++ b/open-sse/utils/proxyFetch.ts @@ -13,7 +13,7 @@ import { proxyConfigToUrl, proxyUrlForLogs, } from "./proxyDispatcher.ts"; -import tlsClient from "./tlsClient.ts"; +import tlsClient, { type TlsFetchOptions } from "./tlsClient.ts"; import { isProxyReachable } from "@/lib/proxyHealth"; import { isControlPlaneProxyDirectFallbackEnabled, @@ -79,6 +79,32 @@ function isTlsFingerprintEnabled() { return process.env.ENABLE_TLS_FINGERPRINT === "true"; } +function tlsFingerprintProviderAllowed( + provider: string | null | undefined, + proxied: boolean +): boolean { + const configured = process.env.TLS_FINGERPRINT_PROVIDERS?.trim(); + // Preserve the legacy direct-only opt-in. The new proxied transport requires + // an explicit allowlist so enabling TLS cannot silently change proxy traffic. + if (!configured) return !proxied; + if (!provider) return false; + const normalizedProvider = provider.trim().toLowerCase(); + return configured + .split(",") + .some((candidate) => candidate.trim().toLowerCase() === normalizedProvider); +} + +type TlsClientLike = { + available: boolean; + fetch: (url: string, options?: TlsFetchOptions) => Promise; +}; +let activeTlsClient: TlsClientLike = tlsClient; + +/** Test seam for exercising wreq selection without replacing the module loader. */ +export function setTlsClientForTest(client: TlsClientLike | null): void { + activeTlsClient = client ?? tlsClient; +} + // #8376: transport-level connect-failure codes that mean "the configured upstream // proxy (or the target itself, for direct egress) is unreachable" — as opposed to an // ordinary upstream HTTP error. Read `.code` first (stable across undici/node @@ -122,9 +148,12 @@ function tagProxyUnreachable(err: T): T { return err; } -/** Per-request tracking of whether TLS fingerprint was used */ -type TlsFingerprintStore = { used: boolean }; -const tlsFingerprintContext = new AsyncLocalStorage(); +/** Per-request TLS identity and success telemetry. */ +type TlsFingerprintStore = { + used: boolean; + provider?: string | null; + sessionScope?: string; +}; /** * #5217 (Gap-secondary): a mutable sink that records the proxy actually applied @@ -227,20 +256,112 @@ function requestHasNonReplayableBody( return false; } +const TLS_ALLOWED_OPTION_KEYS: Record = { + body: true, + headers: true, + method: true, + redirect: true, + signal: true, +}; + +function isWreqBodySupported(body: unknown): boolean { + if (body == null || typeof body === "string") return true; + if (body instanceof ArrayBuffer || ArrayBuffer.isView(body)) return true; + if (body instanceof URLSearchParams) return true; + if (typeof Blob !== "undefined" && body instanceof Blob) return true; + if (typeof FormData !== "undefined" && body instanceof FormData) return true; + return false; +} + +function isTlsRequestEligible( + input: RequestInfo | URL, + options: FetchWithDispatcherOptions +): boolean { + if (typeof Request !== "undefined" && input instanceof Request) return false; + if (!isWreqBodySupported(options.body)) return false; + return Object.keys(options).every((key) => TLS_ALLOWED_OPTION_KEYS[key] === true); +} + +function isTlsFallbackReplaySafe( + input: RequestInfo | URL, + options: FetchWithDispatcherOptions +): boolean { + const method = ( + options.method ?? + (typeof Request !== "undefined" && input instanceof Request ? input.method : "GET") + ).toUpperCase(); + return ( + (method === "GET" || method === "HEAD" || method === "OPTIONS") && + !requestHasNonReplayableBody(input, options) + ); +} + +function getEffectiveSignal( + input: RequestInfo | URL, + options: FetchWithDispatcherOptions +): AbortSignal | null | undefined { + return ( + options.signal ?? + (typeof Request !== "undefined" && input instanceof Request ? input.signal : undefined) + ); +} + +function isWreqProxySupported(proxyUrl: string): boolean { + try { + const parsed = new URL(proxyUrl); + return ( + (parsed.protocol === "http:" || parsed.protocol === "https:") && + parsed.searchParams.get("family") === null + ); + } catch { + return false; + } +} + +function sanitizeTransportError( + error: unknown, + message: string, + fallbackCode: string +): Error & { code: string; errorCode?: string; statusCode?: number } { + const source = error && typeof error === "object" ? (error as Record) : {}; + const sanitized = new Error(message) as Error & { + code: string; + errorCode?: string; + statusCode?: number; + }; + sanitized.code = + typeof source.code === "string" && /^[A-Z0-9_:-]{1,64}$/.test(source.code) + ? source.code + : fallbackCode; + if ( + typeof source.errorCode === "string" && + /^[a-zA-Z0-9_:-]{1,64}$/.test(source.errorCode) + ) { + sanitized.errorCode = source.errorCode; + } + if (typeof source.statusCode === "number" && Number.isFinite(source.statusCode)) { + sanitized.statusCode = source.statusCode; + } + return sanitized; +} + /** Injectable dependencies for testability (Approach B DI). */ export type ProxyFetchDeps = { undiciFetch?: FetchWithDispatcher; nativeFetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise; + findWorkingProxy?: (hostname: string, targetUrl: string) => Promise; }; type PatchState = { originalFetch: typeof globalThis.fetch; proxyContext: AsyncLocalStorage; + tlsFingerprintContext?: AsyncLocalStorage; isPatched: boolean; }; const isCloud = typeof caches !== "undefined" && typeof caches === "object"; const PATCH_STATE_KEY = Symbol.for("omniroute.proxyFetch.state"); +const DIRECT_PROXY_CONTEXT = Symbol.for("omniroute.proxyFetch.direct-context"); function getPatchState(): PatchState { const scopedGlobal = globalThis as typeof globalThis & { @@ -251,6 +372,7 @@ function getPatchState(): PatchState { scopedGlobal[PATCH_STATE_KEY] = { originalFetch: globalThis.fetch, proxyContext: new AsyncLocalStorage(), + tlsFingerprintContext: new AsyncLocalStorage(), isPatched: false, }; } @@ -258,9 +380,11 @@ function getPatchState(): PatchState { } const patchState = getPatchState(); +patchState.tlsFingerprintContext ??= new AsyncLocalStorage(); const originalFetch = patchState.originalFetch; const originalFetchWithDispatcher = originalFetch as FetchWithDispatcher; const proxyContext = patchState.proxyContext; +const tlsFingerprintContext = patchState.tlsFingerprintContext; function noProxyMatch(targetUrl) { const noProxy = process.env.NO_PROXY || process.env.no_proxy; @@ -381,6 +505,9 @@ export function resolveProxyForRequest(targetUrl) { } const contextProxy = proxyContext.getStore(); + if (contextProxy === DIRECT_PROXY_CONTEXT) { + return { source: "direct", proxyUrl: null }; + } if (contextProxy) { // #9551: NO_PROXY must bypass context-proxy too if (target && noProxyMatch(targetUrl)) { @@ -398,16 +525,15 @@ export function resolveProxyForRequest(targetUrl) { } /** - * A caller-initiated abort/timeout is not a proxy transport failure — it must - * not be misreported as one. Prefer `signal.aborted` because - * `AbortController.abort(reason)` may surface a custom Error rather than a - * standard AbortError/TimeoutError name. - * Ported from decolua/9router#2589 (`isCallerAbort`). + * A caller-initiated abort is identified only by the caller's effective signal. + * Dependency-internal TimeoutError/AbortError values are transport failures and + * retain the normal safe-method fallback behavior. */ -function isCallerAbort(error: unknown, signal: AbortSignal | null | undefined): boolean { - if (signal?.aborted === true) return true; - const name = (error as { name?: unknown } | null)?.name; - return name === "AbortError" || name === "TimeoutError"; +function isCallerAbort( + _error: unknown, + signal: AbortSignal | null | undefined +): boolean { + return signal?.aborted === true; } function getTargetUrl(input) { @@ -425,9 +551,13 @@ export async function runWithProxyContext( throw new TypeError("runWithProxyContext requires a callback function"); } - // Inherit existing context if no specific proxyConfig is provided + // Inherit existing context if no specific proxyConfig is provided. A direct + // sentinel must remain direct without being mistaken for a proxy config. const currentContext = proxyContext.getStore(); - const effectiveProxyConfig = proxyConfig || currentContext || null; + const inheritsDirect = currentContext === DIRECT_PROXY_CONTEXT && !proxyConfig; + const effectiveProxyConfig = + proxyConfig || (inheritsDirect ? null : currentContext) || null; + const contextValue = inheritsDirect ? DIRECT_PROXY_CONTEXT : effectiveProxyConfig; const resolvedProxyUrl = effectiveProxyConfig ? proxyConfigToUrl(effectiveProxyConfig) : null; @@ -435,8 +565,9 @@ export async function runWithProxyContext( // This fallback changes egress IP, so upgrades must not silently turn it on. const directFallbackOnUnreachable = opts?.directFallbackOnUnreachable === true && isControlPlaneProxyDirectFallbackEnabled(); - // Run fn with the proxy context cleared so the request egresses directly. - const runDirect = () => proxyContext.run(null, fn); + // Keep an explicit direct sentinel so resolveProxyForRequest cannot re-read + // HTTPS_PROXY/HTTP_PROXY after the control-plane route decision. + const runDirect = () => proxyContext.run(DIRECT_PROXY_CONTEXT, fn); // T14: Proxy Fast-Fail (non-blocking, #9100) // Perform a short TCP reachability check BEFORE issuing upstream requests. @@ -502,7 +633,7 @@ export async function runWithProxyContext( } } - return proxyContext.run(effectiveProxyConfig, async () => { + return proxyContext.run(contextValue, async () => { if (resolvedProxyUrl && effectiveProxyConfig !== currentContext) { // #9158: this fires on EVERY proxied request (innermost context wins). // Gate it behind the same env flag as the relay routing log so request @@ -604,23 +735,47 @@ async function patchedFetch( const { source, proxyUrl } = resolved; if (!proxyUrl) { - // TLS fingerprint spoofing for direct connections (no proxy configured) - if (isTlsFingerprintEnabled() && tlsClient.available) { + // TLS fingerprint spoofing for an already-resolved direct route. Explicit + // proxy:null prevents wreq from re-reading a global environment proxy. + const tlsStore = tlsFingerprintContext.getStore(); + let tlsDirectFallback = false; + if ( + isTlsFingerprintEnabled() && + activeTlsClient.available && + tlsFingerprintProviderAllowed(tlsStore?.provider, false) && + isTlsRequestEligible(input, options) + ) { try { - const store = tlsFingerprintContext.getStore(); - if (store) store.used = true; - return await tlsClient.fetch(targetUrl, { - ...options, + const response = await activeTlsClient.fetch(targetUrl, { + method: options.method, headers: options.headers, - signal: options.signal ?? undefined, + body: options.body as TlsFetchOptions["body"], + redirect: options.redirect, + signal: getEffectiveSignal(input, options), + proxy: null, + sessionScope: tlsStore?.sessionScope, }); + if (tlsStore) tlsStore.used = true; + return response; } catch (error) { - const message = error instanceof Error ? error.message : String(error); - console.warn( - `[ProxyFetch] TLS fingerprint failed, falling back to native fetch: ${message}` - ); - const store = tlsFingerprintContext.getStore(); - if (store) store.used = false; + if (isCallerAbort(error, getEffectiveSignal(input, options))) throw error; + const sessionHadCookies = + !!error && + typeof error === "object" && + "sessionHadCookies" in error && + error.sessionHadCookies === true; + if (!isTlsFallbackReplaySafe(input, options) || sessionHadCookies) { + throw sanitizeTransportError( + error, + sessionHadCookies + ? "TLS fingerprint request failed; stateful session cannot be replayed" + : "TLS fingerprint request failed; request is not safe to replay", + "TLS_FINGERPRINT_FAILED" + ); + } + console.warn("[ProxyFetch] TLS fingerprint transport failed; using direct dispatcher"); + if (tlsStore) tlsStore.used = false; + tlsDirectFallback = true; } } // Direct connection (no proxy) — use undici with custom dispatcher for timeout control. @@ -695,7 +850,11 @@ async function patchedFetch( } // All attempts exhausted — try proxy fallback before native fetch - if (source === "direct" && isFeatureFlagEnabled("PROXY_AUTO_SELECT_ENABLED")) { + if ( + !tlsDirectFallback && + source === "direct" && + isFeatureFlagEnabled("PROXY_AUTO_SELECT_ENABLED") + ) { let targetHostname = ""; try { targetHostname = new URL(targetUrl).hostname; @@ -703,7 +862,8 @@ async function patchedFetch( // ignore } if (targetHostname) { - const { findWorkingProxy } = await import("./proxyFallback.ts"); + const findWorkingProxy = + deps.findWorkingProxy ?? (await import("./proxyFallback.ts")).findWorkingProxy; const fallbackProxyUrl = await findWorkingProxy(targetHostname, targetUrl); if (fallbackProxyUrl) { try { @@ -854,6 +1014,51 @@ async function patchedFetch( throw lastRelayError; } + // The proxied TLS overlay is deliberately narrow: approved provider, exact + // http(s) proxy, no relay/family pinning, and only options wreq can preserve. + const tlsStore = tlsFingerprintContext.getStore(); + if ( + isTlsFingerprintEnabled() && + typeof tlsStore?.sessionScope === "string" && + tlsStore.sessionScope.trim().length > 0 && + activeTlsClient.available && + tlsFingerprintProviderAllowed(tlsStore?.provider, true) && + isTlsRequestEligible(input, options) && + isWreqProxySupported(proxyUrl) + ) { + try { + const response = await activeTlsClient.fetch(targetUrl, { + method: options.method, + headers: options.headers, + body: options.body as TlsFetchOptions["body"], + redirect: options.redirect, + signal: getEffectiveSignal(input, options), + proxy: proxyUrl, + sessionScope: tlsStore?.sessionScope, + }); + if (tlsStore) tlsStore.used = true; + return response; + } catch (error) { + if (isCallerAbort(error, getEffectiveSignal(input, options))) throw error; + const sessionHadCookies = + !!error && + typeof error === "object" && + "sessionHadCookies" in error && + error.sessionHadCookies === true; + if (!isTlsFallbackReplaySafe(input, options) || sessionHadCookies) { + throw sanitizeTransportError( + error, + sessionHadCookies + ? "TLS fingerprint request failed; stateful session cannot be replayed" + : "TLS fingerprint request failed; request is not safe to replay", + "TLS_FINGERPRINT_FAILED" + ); + } + console.warn("[ProxyFetch] TLS fingerprint transport failed; using proxy dispatcher"); + if (tlsStore) tlsStore.used = false; + } + } + // #9100: proxy path — attempt 0 uses the pooled keep-alive dispatcher // (pipelining 4, ONE reused TCP connection per proxy host). A transient // socket error on a stale pooled socket is retried ONCE on a fresh @@ -872,6 +1077,7 @@ async function patchedFetch( attempt === 0 ? createProxyDispatcher(proxyUrl) : getProxyRetryDispatcher(proxyUrl), }); } catch (error) { + if (isCallerAbort(error, getEffectiveSignal(input, options))) throw error; const msg = error instanceof Error ? error.message : String(error); const errCode = (error as { code?: unknown })?.code; const isTransportFailure = @@ -889,13 +1095,16 @@ async function patchedFetch( await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS)); continue; } - // A caller abort/timeout must propagate unchanged and without a noisy - // "Proxy request failed" log — it's not a proxy transport failure. - if (!isCallerAbort(error, options?.signal)) { - const message = error instanceof Error ? error.message : String(error); - console.error(`[ProxyFetch] Proxy request failed (${source}, fail-closed): ${message}`); - } - throw error; + tagProxyUnreachable(error); + const sanitized = sanitizeTransportError( + error, + "Proxy request failed", + "PROXY_REQUEST_FAILED" + ); + console.error( + `[ProxyFetch] Proxy request failed (${source}, fail-closed; code=${sanitized.code})` + ); + throw sanitized; } } throw lastProxyError; @@ -919,19 +1128,64 @@ if (!isCloud && !patchState.isPatched) { patchState.isPatched = true; } +export type TlsTrackingIdentity = { + provider?: string | null; + sessionScope?: string; +}; + /** - * Run a function with TLS fingerprint tracking context. - * After fn completes, returns { result, tlsFingerprintUsed }. + * Run a function with account-scoped TLS fingerprint tracking. + * Both historical forms remain valid: runWithTlsTracking(fn) and + * runWithTlsTracking(provider, fn). */ -export async function runWithTlsTracking(fn) { - const store = { used: false }; - const result = await tlsFingerprintContext.run(store, fn); +export async function runWithTlsTracking( + fn: () => T +): Promise<{ result: Awaited; tlsFingerprintUsed: boolean }>; +export async function runWithTlsTracking( + provider: string | null | undefined, + fn: () => T +): Promise<{ result: Awaited; tlsFingerprintUsed: boolean }>; +export async function runWithTlsTracking( + identity: TlsTrackingIdentity, + fn: () => T +): Promise<{ result: Awaited; tlsFingerprintUsed: boolean }>; +export async function runWithTlsTracking( + providerOrIdentityOrFn: string | null | undefined | TlsTrackingIdentity | (() => T), + maybeFn?: () => T +): Promise<{ result: Awaited; tlsFingerprintUsed: boolean }> { + const legacyFn = + typeof providerOrIdentityOrFn === "function" ? providerOrIdentityOrFn : maybeFn; + if (typeof legacyFn !== "function") { + throw new TypeError("runWithTlsTracking requires a callback function"); + } + const identity: TlsTrackingIdentity = + providerOrIdentityOrFn && + typeof providerOrIdentityOrFn === "object" && + typeof providerOrIdentityOrFn !== "function" + ? providerOrIdentityOrFn + : { + provider: + typeof providerOrIdentityOrFn === "string" ? providerOrIdentityOrFn : undefined, + }; + const store: TlsFingerprintStore = { + used: false, + provider: identity.provider, + sessionScope: identity.sessionScope, + }; + const result = await tlsFingerprintContext.run(store, legacyFn); return { result, tlsFingerprintUsed: store.used }; } -/** Check if TLS fingerprint is enabled and available */ -export function isTlsFingerprintActive() { - return isTlsFingerprintEnabled() && tlsClient.available; +/** Check whether TLS fingerprint transport is enabled for this route identity. */ +export function isTlsFingerprintActive( + provider?: string | null, + proxied = false +): boolean { + return ( + isTlsFingerprintEnabled() && + activeTlsClient.available && + tlsFingerprintProviderAllowed(provider, proxied) + ); } /** diff --git a/open-sse/utils/tlsClient.ts b/open-sse/utils/tlsClient.ts index c2b41521d0..2411a89eb6 100644 --- a/open-sse/utils/tlsClient.ts +++ b/open-sse/utils/tlsClient.ts @@ -1,20 +1,40 @@ import { createRequire } from "module"; +import { createHash } from "node:crypto"; import { getTlsClientTimeoutConfig } from "@/shared/utils/runtimeTimeouts"; -const require = createRequire(import.meta.url); +const runtimeRequire = createRequire(import.meta.url); -type WreqSession = { - fetch: (url: string, options?: Record) => Promise; - close: () => Promise | void; +function loadRuntimeModule(moduleName: string): unknown { + // Keep the specifier dynamic. Turbopack rewrites a literal createRequire call + // to a hashed external name that is absent from the standalone Docker runtime. + return Reflect.apply(runtimeRequire, undefined, [moduleName]); +} + +export type WreqResponse = { + status: number; + statusText: string; + headers: Iterable<[string, string]>; + body: ReadableStream | null; + url?: string; + redirected?: boolean; }; -type CreateSessionFn = (options: Record) => Promise; +export type WreqSession = { + fetch: (url: string, options?: Record) => Promise; + close: () => Promise | void; + getCookies?: (url: string | URL) => Record; +}; + +export type CreateSessionFn = (options: Record) => Promise; let createSession: CreateSessionFn | null; try { - const loaded = require("wreq-js") as { createSession?: CreateSessionFn }; + const loaded = loadRuntimeModule("wreq-js") as { createSession?: CreateSessionFn }; createSession = typeof loaded.createSession === "function" ? loaded.createSession : null; } catch { + if (process.env.ENABLE_TLS_FINGERPRINT === "true") { + console.warn("[TlsClient] wreq-js unavailable; TLS fingerprint transport disabled"); + } createSession = null; } @@ -34,12 +54,26 @@ function getProxyFromEnv(): string | undefined { ); } -interface FetchOptions { +export type WreqBodyInit = + | string + | ArrayBuffer + | ArrayBufferView + | URLSearchParams + | Buffer + | Blob + | FormData + | null; + +export interface TlsFetchOptions { method?: string; headers?: HeadersInit; - body?: unknown; - redirect?: string; - signal?: AbortSignal; + body?: WreqBodyInit; + redirect?: RequestRedirect; + signal?: AbortSignal | null; + /** Exact resolved proxy. Undefined preserves legacy environment lookup; null means direct. */ + proxy?: string | null; + /** Stable account/connection identity used to isolate cookies and circuit state. */ + sessionScope?: string; } function normalizeHeaders(headers: HeadersInit | undefined): Record | undefined { @@ -62,182 +96,591 @@ function normalizeHeaders(headers: HeadersInit | undefined): Record= this.circuitOpenUntil; + if ( + "errorCode" in error && + typeof error.errorCode === "string" && + /^[a-zA-Z0-9_:-]{1,64}$/.test(error.errorCode) + ) { + sanitized.errorCode = error.errorCode; } + if ( + "statusCode" in error && + typeof error.statusCode === "number" && + Number.isFinite(error.statusCode) + ) { + sanitized.statusCode = error.statusCode; + } + return sanitized; +} - private recordFailure(): void { - this.failureCount++; - if (this.failureCount >= this.maxFailures) { - this.circuitOpenUntil = Date.now() + this.cooldownMs; - this.circuitTripped = true; - // Close the stale session so the next half-open retry creates a - // fresh one instead of reusing a broken connection. - if (this.session) { - Promise.resolve(this.session.close()).catch(() => {}); - this.session = null; - } - console.warn( - `[TlsClient] Circuit opened after ${this.failureCount} consecutive failures, cooling down for ${this.cooldownMs}ms` +function toNativeResponse( + response: WreqResponse, + onFinalize: () => void, + onBodyError: () => void, + signal?: AbortSignal | null +): Response { + let finalized = false; + let bodyFailureReported = false; + let consumerCancelled = false; + let consumerCancelReason: unknown; + const finalize = () => { + if (finalized) return; + finalized = true; + onFinalize(); + }; + const safeBodyError = (error: unknown): unknown => { + if (signal?.aborted) { + return signal.reason ?? new DOMException("The operation was aborted", "AbortError"); + } + if (consumerCancelled) { + return ( + consumerCancelReason ?? new DOMException("The response body was cancelled", "AbortError") ); - // Double cooldown for the next trip: 30s → 60s → 120s → ... → 10 min max - this.escalateCooldown(); } - } - - private recordSuccess(): void { - this.failureCount = 0; - if (this.circuitTripped) { - this.cooldownMultiplier = 1; - this.cooldownMs = this.baseCooldownMs; - console.log("[TlsClient] Circuit closed (success after cooldown)"); - this.circuitTripped = false; + if (!bodyFailureReported) { + bodyFailureReported = true; + onBodyError(); } + return sanitizeWreqError(error, "wreq-js response body failed"); + }; + if (response instanceof Response) { + finalize(); + return response; } - private escalateCooldown(): void { - this.cooldownMultiplier = Math.min(this.cooldownMultiplier * 2, 20); - this.cooldownMs = Math.min(this.baseCooldownMs * this.cooldownMultiplier, this.MAX_COOLDOWN_MS); + try { + const headers = new Headers(); + for (const [name, value] of response.headers) headers.append(name, value); + let body: ReadableStream | null = null; + if (response.body) { + const reader = response.body.getReader(); + body = new ReadableStream({ + async pull(controller) { + try { + const chunk = await reader.read(); + if (chunk.done) { + finalize(); + controller.close(); + } else { + controller.enqueue(chunk.value); + } + } catch (error) { + controller.error(safeBodyError(error)); + finalize(); + } + }, + async cancel(reason) { + consumerCancelled = true; + consumerCancelReason = reason; + try { + await reader.cancel(reason); + } catch (error) { + throw safeBodyError(error); + } finally { + finalize(); + } + }, + }); + } else { + finalize(); + } + const adapted = new Response(body, { + status: response.status, + statusText: response.statusText, + headers, + }); + if (response.url) { + Object.defineProperty(adapted, "url", { value: response.url, configurable: true }); + } + if (response.redirected !== undefined) { + Object.defineProperty(adapted, "redirected", { + value: response.redirected, + configurable: true, + }); + } + return adapted; + } catch (error) { + finalize(); + throw error; + } +} + +/** + * TLS Client — Chrome 124 TLS fingerprint spoofing via wreq-js. + * Sessions, cookie jars, and circuit state are isolated by account scope and exact proxy. + */ +export class TlsClient { + private readonly createSessionFn: CreateSessionFn | null; + private readonly sessions = new Map(); + private readonly pendingSessions = new Map>(); + private readonly pendingCloses = new Set>(); + private readonly sessionEpochs = new Map(); + private readonly sessionUseCounts = new Map(); + private readonly sessionLastUsed = new Map(); + private readonly pendingEvictions = new Set(); + private accessSequence = 0; + private readonly circuits = new Map< + string, + { + failureCount: number; + cooldownMs: number; + cooldownMultiplier: number; + circuitOpenUntil: number; + circuitTripped: boolean; + halfOpenInFlight: boolean; + sessionHadCookies: boolean; + } + >(); + private globalSessionEpoch = 0; + private readonly maxFailures = 3; + private readonly baseCooldownMs = 30_000; + private readonly maxCooldownMs = 600_000; + private readonly legacySessionScope = "legacy"; + private readonly _libraryAvailable: boolean; + private readonly maxSessions: number; + + constructor( + createSessionFn: CreateSessionFn | null = createSession, + maxSessions = 128 + ) { + this.createSessionFn = createSessionFn; + this._libraryAvailable = !!createSessionFn; + this.maxSessions = + Number.isInteger(maxSessions) && maxSessions > 0 ? maxSessions : 128; } - private checkCircuit(): boolean { - if (!this.circuitTripped) return true; + /** Library availability only. Per-session circuit state is enforced inside fetch(). */ + get available(): boolean { + return this._libraryAvailable; + } - if (Date.now() >= this.circuitOpenUntil) { - console.log("[TlsClient] Half-open: retrying after cooldown"); - // Don't call recordSuccess() here — that would reset failureCount. - // Instead, let the fetch() call succeed or fail naturally. - // If it succeeds, recordSuccess() in fetch() handles cleanup. - // If it fails, recordFailure() finds failureCount still >= maxFailures - // and re-opens with escalated cooldown. + private resolveProxy(proxy?: string | null): string | null { + return proxy === undefined ? (getProxyFromEnv() ?? null) : proxy; + } + + private getSessionKey(resolvedProxy: string | null, sessionScope?: string): string { + const scope = sessionScope?.trim() || this.legacySessionScope; + return createHash("sha256") + .update(scope) + .update("\0") + .update(resolvedProxy ?? "") + .digest("base64url"); + } + + private getDefaultSessionKey(): string { + return this.getSessionKey(this.resolveProxy(undefined), this.legacySessionScope); + } + + private getSessionEpoch(key: string): number { + return this.sessionEpochs.get(key) ?? 0; + } + + private hasSessionCookies(session: WreqSession | null, url: string): boolean { + if (!session) return false; + if (!session.getCookies) return true; + try { + return Object.keys(session.getCookies(url)).length > 0; + } catch { + // If cookie state cannot be inspected, fail closed and forbid replay. return true; } - - return false; } - async getSession() { - if (!this.checkCircuit()) return null; - if (!this.available) return null; - if (this.session) return this.session; - const createSessionFn = createSession; - if (!createSessionFn) return null; + private closeSession(session: WreqSession): Promise { + let closing: Promise; + closing = Promise.resolve() + .then(() => session.close()) + .catch(() => {}) + .finally(() => { + this.pendingCloses.delete(closing); + }); + this.pendingCloses.add(closing); + return closing; + } + + private findOldestIdleSession(protectedKey?: string): string | undefined { + let candidate: string | undefined; + let candidateSequence = Number.POSITIVE_INFINITY; + for (const key of this.sessions.keys()) { + if (key === protectedKey || (this.sessionUseCounts.get(key) ?? 0) > 0) continue; + const sequence = this.sessionLastUsed.get(key) ?? 0; + if (sequence < candidateSequence) { + candidate = key; + candidateSequence = sequence; + } + } + return candidate; + } + + private reserveSessionCapacity(protectedKey: string): void { + if ( + this.pendingSessions.size >= this.maxSessions || + this.pendingCloses.size >= this.maxSessions + ) { + const error = new Error("wreq-js session capacity exhausted") as Error & { + code?: string; + }; + error.code = "TLS_SESSION_CAPACITY"; + throw error; + } + while (this.sessions.size >= this.maxSessions) { + const candidate = this.findOldestIdleSession(protectedKey); + if (!candidate) { + const error = new Error("wreq-js session capacity exhausted") as Error & { + code?: string; + }; + error.code = "TLS_SESSION_CAPACITY"; + throw error; + } + void this.invalidateSession(candidate); + } + } + + private retainSession(key: string): void { + this.pendingEvictions.delete(key); + this.sessionUseCounts.set(key, (this.sessionUseCounts.get(key) ?? 0) + 1); + this.sessionLastUsed.set(key, ++this.accessSequence); + } + + private releaseSession(key: string): void { + const remaining = (this.sessionUseCounts.get(key) ?? 1) - 1; + if (remaining > 0) { + this.sessionUseCounts.set(key, remaining); + return; + } + this.sessionUseCounts.delete(key); + if (this.pendingEvictions.delete(key)) { + void this.invalidateSession(key); + return; + } + this.evictSessionsIfNeeded(); + } + + private evictSessionsIfNeeded(protectedKey?: string): void { + while (this.sessions.size > this.maxSessions) { + const candidate = this.findOldestIdleSession(protectedKey); + if (candidate) { + void this.invalidateSession(candidate); + continue; + } + + let activeCandidate: string | undefined; + let candidateSequence = Number.POSITIVE_INFINITY; + for (const key of this.sessions.keys()) { + if (key === protectedKey || this.pendingEvictions.has(key)) continue; + const sequence = this.sessionLastUsed.get(key) ?? 0; + if (sequence < candidateSequence) { + activeCandidate = key; + candidateSequence = sequence; + } + } + if (activeCandidate) this.pendingEvictions.add(activeCandidate); + return; + } + } + + private invalidateSession(key: string): Promise { + const pending = this.pendingSessions.get(key); + const invalidatedEpoch = this.getSessionEpoch(key) + 1; + this.sessionEpochs.set(key, invalidatedEpoch); + this.pendingSessions.delete(key); + this.sessionUseCounts.delete(key); + this.sessionLastUsed.delete(key); + this.pendingEvictions.delete(key); + const session = this.sessions.get(key); + this.sessions.delete(key); + if (pending) { + void pending + .finally(() => { + if ( + this.getSessionEpoch(key) === invalidatedEpoch && + !this.pendingSessions.has(key) && + !this.sessions.has(key) + ) { + this.sessionEpochs.delete(key); + } + }) + .catch(() => {}); + } else { + this.sessionEpochs.delete(key); + } + return session ? this.closeSession(session) : Promise.resolve(); + } + + private async closeSessions(): Promise { + const pending = [...this.pendingSessions.values()]; + this.globalSessionEpoch++; + this.pendingSessions.clear(); + this.sessionEpochs.clear(); + const sessions = [...this.sessions.values()]; + this.sessions.clear(); + this.sessionUseCounts.clear(); + this.sessionLastUsed.clear(); + this.pendingEvictions.clear(); + this.circuits.clear(); + const closes = sessions.map((session) => this.closeSession(session)); + await Promise.allSettled([...closes, ...pending]); + await Promise.allSettled([...this.pendingCloses]); + } + + private checkCircuit(key = this.getDefaultSessionKey()): boolean { + const state = this.circuits.get(key); + if (!state || !state.circuitTripped) return true; + if (Date.now() < state.circuitOpenUntil) return false; + if (state.halfOpenInFlight) return false; + state.halfOpenInFlight = true; + console.log("[TlsClient] Half-open: retrying after cooldown"); + return true; + } + + private recordFailure( + key = this.getDefaultSessionKey(), + sessionHadCookies = false + ): void { + const state = this.circuits.get(key) ?? { + failureCount: 0, + cooldownMs: this.baseCooldownMs, + cooldownMultiplier: 1, + circuitOpenUntil: 0, + circuitTripped: false, + halfOpenInFlight: false, + sessionHadCookies: false, + }; + state.sessionHadCookies ||= sessionHadCookies; + state.failureCount++; + state.halfOpenInFlight = false; + if (state.failureCount >= this.maxFailures) { + state.circuitOpenUntil = Date.now() + state.cooldownMs; + state.circuitTripped = true; + if ((this.sessionUseCounts.get(key) ?? 0) > 0) { + this.pendingEvictions.add(key); + } else { + void this.invalidateSession(key); + } + console.warn( + `[TlsClient] Circuit opened after ${state.failureCount} consecutive failures, cooling down for ${state.cooldownMs}ms` + ); + state.cooldownMultiplier = Math.min(state.cooldownMultiplier * 2, 20); + state.cooldownMs = Math.min( + this.baseCooldownMs * state.cooldownMultiplier, + this.maxCooldownMs + ); + } + this.circuits.delete(key); + this.circuits.set(key, state); + const maxCircuitEntries = this.maxSessions * 2; + while (this.circuits.size > maxCircuitEntries) { + const oldestKey = this.circuits.keys().next().value; + if (typeof oldestKey !== "string") break; + this.circuits.delete(oldestKey); + } + } + + private recordSuccess(key = this.getDefaultSessionKey()): void { + const state = this.circuits.get(key); + if (state?.circuitTripped) { + console.log("[TlsClient] Circuit closed (success after cooldown)"); + } + this.circuits.delete(key); + } + + private releaseHalfOpen(key: string): void { + const state = this.circuits.get(key); + if (state) state.halfOpenInFlight = false; + } + + private async getSession( + resolvedProxy: string | null, + key: string + ): Promise { + const cached = this.sessions.get(key); + if (cached) { + this.pendingEvictions.delete(key); + this.sessionLastUsed.set(key, ++this.accessSequence); + return cached; + } + const pending = this.pendingSessions.get(key); + if (pending) return pending; + if (!this.createSessionFn) return null; + this.reserveSessionCapacity(key); - const proxy = getProxyFromEnv(); const sessionOpts: Record = { browser: "chrome_124", os: "macos", }; - if (proxy) { - sessionOpts.proxy = proxy; - console.log(`[TlsClient] Using proxy: ${proxy}`); - } + if (resolvedProxy) sessionOpts.proxy = resolvedProxy; + const globalEpoch = this.globalSessionEpoch; + const sessionEpoch = this.getSessionEpoch(key); - this.session = await createSessionFn(sessionOpts); - console.log("[TlsClient] Session created (Chrome 124 TLS fingerprint)"); - return this.session; + const creating = Reflect.apply(this.createSessionFn, undefined, [sessionOpts]) + .then(async (session) => { + if ( + globalEpoch !== this.globalSessionEpoch || + sessionEpoch !== this.getSessionEpoch(key) + ) { + await this.closeSession(session); + throw new Error("wreq-js session invalidated"); + } + if (this.sessions.size >= this.maxSessions) { + const candidate = this.findOldestIdleSession(key); + if (!candidate) { + await this.closeSession(session); + const error = new Error("wreq-js session capacity exhausted") as Error & { + code?: string; + }; + error.code = "TLS_SESSION_CAPACITY"; + throw error; + } + void this.invalidateSession(candidate); + } + this.sessions.set(key, session); + this.sessionLastUsed.set(key, ++this.accessSequence); + this.evictSessionsIfNeeded(key); + console.log("[TlsClient] Session created (Chrome 124 TLS fingerprint)"); + return session; + }) + .finally(() => { + if (this.pendingSessions.get(key) === creating) { + this.pendingSessions.delete(key); + this.sessionEpochs.delete(key); + } + }); + this.pendingSessions.set(key, creating); + return creating; } - /** - * Fetch with Chrome 124 TLS fingerprint. - * wreq-js Response is already fetch-compatible (headers, text(), json(), clone(), body). - */ - async fetch(url: string, options: FetchOptions = {}) { - if (!this.checkCircuit()) { - throw new Error("wreq-js circuit open — skipping TLS request"); + /** Fetch with Chrome 124 TLS fingerprint and an account-scoped persistent cookie jar. */ + async fetch(url: string, options: TlsFetchOptions = {}): Promise { + const resolvedProxy = this.resolveProxy(options.proxy); + const key = this.getSessionKey(resolvedProxy, options.sessionScope); + if (!this.checkCircuit(key)) { + const state = this.circuits.get(key); + const error = new Error("wreq-js circuit open — skipping TLS request") as Error & { + code?: string; + }; + error.code = "TLS_CIRCUIT_OPEN"; + if (state?.sessionHadCookies) { + Object.defineProperty(error, "sessionHadCookies", { + value: true, + configurable: true, + }); + } + throw error; } + let session: WreqSession | null = null; + let sessionUseRetained = false; + const releaseSession = () => { + if (!sessionUseRetained) return; + sessionUseRetained = false; + this.releaseSession(key); + }; try { - const session = await this.getSession(); + session = await this.getSession(resolvedProxy, key); if (!session) throw new Error("wreq-js not available"); + this.retainSession(key); + sessionUseRetained = true; const { timeoutMs } = getTlsClientTimeoutConfig(process.env, (message) => { console.warn(`[TlsClient] ${message}`); }); - const method = (options.method || "GET").toUpperCase(); - const wreqOptions: Record = { - method, + method: (options.method || "GET").toUpperCase(), headers: normalizeHeaders(options.headers), body: options.body, - redirect: options.redirect === "manual" ? "manual" : "follow", + redirect: options.redirect ?? "follow", timeout: timeoutMs, }; + if (options.signal) wreqOptions.signal = options.signal; - if (options.signal) { - wreqOptions.signal = options.signal; - } - - const response = await session.fetch(url, wreqOptions); - this.recordSuccess(); + const response = toNativeResponse( + await session.fetch(url, wreqOptions), + releaseSession, + () => this.recordFailure(key, this.hasSessionCookies(session, url)), + options.signal + ); + this.recordSuccess(key); return response; } catch (err) { - const isAbort = - err instanceof Error && (err.name === "AbortError" || err.message.includes("aborted")); - if (!isAbort) { - this.recordFailure(); + const isCallerAbort = options.signal?.aborted === true; + const sessionHadCookies = + !isCallerAbort && this.hasSessionCookies(session, url); + releaseSession(); + if (isCallerAbort) { + this.releaseHalfOpen(key); + } else { + this.recordFailure(key, sessionHadCookies); } - throw err; + if (isCallerAbort) throw err; + const transportError = sanitizeWreqError(err, "wreq-js transport failed"); + if (sessionHadCookies) { + Object.defineProperty(transportError, "sessionHadCookies", { + value: true, + configurable: true, + }); + } + throw transportError; } } - async exit() { - if (this.session) { - await this.session.close(); - this.session = null; + async exit(): Promise { + await this.closeSessions(); + } + + resetCircuit(proxy?: string | null, sessionScope?: string): void { + if (arguments.length === 0) { + this.circuits.clear(); + return; } + const resolvedProxy = this.resolveProxy(proxy); + this.circuits.delete(this.getSessionKey(resolvedProxy, sessionScope)); } - resetCircuit(): void { - this.failureCount = 0; - this.circuitTripped = false; - this.circuitOpenUntil = 0; - } - - getCircuitState(): { + getCircuitState( + proxy?: string | null, + sessionScope?: string + ): { available: boolean; circuitTripped: boolean; failureCount: number; circuitOpenUntil: number; coolDownRemainingMs: number; } { + const resolvedProxy = this.resolveProxy(proxy); + const key = this.getSessionKey(resolvedProxy, sessionScope); + const state = this.circuits.get(key); + const circuitOpenUntil = state?.circuitOpenUntil ?? 0; + const circuitTripped = state?.circuitTripped ?? false; return { - available: this.available, - circuitTripped: this.circuitTripped, - failureCount: this.failureCount, - circuitOpenUntil: this.circuitOpenUntil, + available: + this._libraryAvailable && + (!circuitTripped || Date.now() >= circuitOpenUntil), + circuitTripped, + failureCount: state?.failureCount ?? 0, + circuitOpenUntil, coolDownRemainingMs: - this.circuitOpenUntil > 0 ? Math.max(0, this.circuitOpenUntil - Date.now()) : 0, + circuitOpenUntil > 0 ? Math.max(0, circuitOpenUntil - Date.now()) : 0, }; } } -const tlsClient = new TlsClient(); +const TLS_CLIENT_KEY = Symbol.for("omniroute.tlsClient.instance"); +const scopedGlobal = globalThis as typeof globalThis & { + [TLS_CLIENT_KEY]?: TlsClient; +}; +const tlsClient = scopedGlobal[TLS_CLIENT_KEY] ?? new TlsClient(); +scopedGlobal[TLS_CLIENT_KEY] = tlsClient; export default tlsClient; diff --git a/scripts/build/assembleStandalone.mjs b/scripts/build/assembleStandalone.mjs index 5a528f09b6..1daaf8684c 100644 --- a/scripts/build/assembleStandalone.mjs +++ b/scripts/build/assembleStandalone.mjs @@ -10,7 +10,7 @@ * .next/standalone -> outDir (cp) Y Y Y SHARED * .next/static -> outDir/.next/static (cp) Y Y Y SHARED * public/ -> outDir/public/ (cp) Y Y Y SHARED - * wreq-js/rust -> outDir/node_modules/wreq-js/rust Y - - SHARED (native asset) + * wreq-js -> outDir/node_modules/wreq-js Y Y Y SHARED (extra module) * better-sqlite3/build -> outDir/node_modules/better-sqlite3/ Y - - SHARED (native asset) * @swc/helpers -> outDir/node_modules/@swc/helpers Y Y Y SHARED (extra module) * pino-abstract-transport -> outDir/node_modules/... Y - - SHARED (extra module) @@ -76,11 +76,6 @@ async function exists(targetPath) { * for either path/platform. @type {{label:string, src:string[], dest:string[]}[]} */ export const NATIVE_ASSET_ENTRIES = [ - { - label: "wreq-js native runtime", - src: ["node_modules", "wreq-js", "rust"], - dest: ["node_modules", "wreq-js", "rust"], - }, { label: "better-sqlite3 native binary", src: ["node_modules", "better-sqlite3", "build"], @@ -117,6 +112,15 @@ export const NATIVE_ASSET_ENTRIES = [ /** @type {{label:string, src:string[], dest:string[]}[]} */ const EXTRA_MODULE_ENTRIES = [ + { + // tlsClient.ts intentionally resolves wreq-js through a runtime-dynamic + // require so Turbopack cannot rewrite the package name to a hashed external. + // That also makes the package invisible to static tracing, so copy the whole + // module—not only rust/—into every standalone artifact. + label: "wreq-js TLS runtime", + src: ["node_modules", "wreq-js"], + dest: ["node_modules", "wreq-js"], + }, { label: "@swc/helpers", src: ["node_modules", "@swc", "helpers"], @@ -278,7 +282,7 @@ const EXTRA_MODULE_ENTRIES = [ ]; /** - * Copy native standalone assets (wreq-js rust/, better-sqlite3 build/). + * Copy native standalone assets (better-sqlite3 build/prebuilds and TPROXY). * * The destination is derived as //standalone/node_modules/... * for backward compatibility with existing callers and tests. @@ -512,8 +516,8 @@ function copyStaticAndPublic({ distDir, relDistDir, projectRoot, resolvedOutDir } /** - * Copy native assets (wreq-js, better-sqlite3) and extra runtime modules/sidecars - * (pino, migrations, MITM server, helper scripts, sqlite-vec platform packages, …) + * Copy native assets (better-sqlite3 and TPROXY) and extra runtime modules/sidecars + * (wreq-js, pino, migrations, MITM server, helper scripts, sqlite-vec platform packages, …) * into the assembled bundle. Missing sources are skipped silently. * * @param {string} projectRoot diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index c94f055241..835c0d58e6 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -529,6 +529,15 @@ export async function executeChatWithBreaker({ ) ); + const tlsTrackingIdentity = { + provider, + sessionScope: credentials.connectionId, + }; + // Track whenever direct TLS is possible. proxyFetch decides against wreq only + // after resolving NO_PROXY/local bypasses, so predicting from proxyInfo here + // would drop the account scope when a configured proxy resolves to direct. + const tlsFingerprintActive = isTlsFingerprintActive(provider); + if (isShadowTraffic) { if (!bypassCircuitBreaker && breaker && !breaker.canExecute()) { const retryAfterMs = breaker.getRetryAfterMs(); @@ -542,8 +551,8 @@ export async function executeChatWithBreaker({ }; } - if (!proxyInfo?.proxy && isTlsFingerprintActive()) { - const tracked = await runWithTlsTracking(chatFn); + if (tlsFingerprintActive) { + const tracked = await runWithTlsTracking(tlsTrackingIdentity, chatFn); return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; } @@ -552,8 +561,8 @@ export async function executeChatWithBreaker({ } if (bypassCircuitBreaker) { - if (!proxyInfo?.proxy && isTlsFingerprintActive()) { - const tracked = await runWithTlsTracking(chatFn); + if (tlsFingerprintActive) { + const tracked = await runWithTlsTracking(tlsTrackingIdentity, chatFn); return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; } @@ -561,8 +570,10 @@ export async function executeChatWithBreaker({ return { result, tlsFingerprintUsed: false }; } - if (!proxyInfo?.proxy && isTlsFingerprintActive()) { - const tracked = await breaker.execute(async () => runWithTlsTracking(chatFn)); + if (tlsFingerprintActive) { + const tracked = await breaker.execute(async () => + runWithTlsTracking(tlsTrackingIdentity, chatFn) + ); return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; } diff --git a/tests/unit/build-next-isolated.test.ts b/tests/unit/build-next-isolated.test.ts index 401edf6bc6..13b2f8598c 100644 --- a/tests/unit/build-next-isolated.test.ts +++ b/tests/unit/build-next-isolated.test.ts @@ -4,14 +4,15 @@ import fs from "node:fs/promises"; import fsSync from "node:fs"; import os from "node:os"; import path from "node:path"; - -const { +import { getTransientBuildPaths, movePath, pruneStandaloneArtifacts, resolveNextBuildEnv, + syncStandaloneExtraModules, syncStandaloneNativeAssets, -} = await import("../../scripts/build/build-next-isolated.mjs"); +} from "../../scripts/build/build-next-isolated.mjs"; + async function withTempDir(fn) { const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "omniroute-build-next-isolated-")); @@ -177,36 +178,49 @@ test("pruneStandaloneArtifacts removes traced _tasks from standalone output", as }); }); -test("syncStandaloneNativeAssets copies wreq-js native runtime into standalone output", async () => { +test("syncStandaloneExtraModules copies the complete wreq-js runtime", async () => { await withTempDir(async (tempDir) => { - const sourceNativeFile = path.join( - tempDir, - "node_modules", - "wreq-js", - "rust", - "wreq-js.linux-x64-gnu.node" - ); - const destinationNativeFile = path.join( + const sourcePackage = path.join(tempDir, "node_modules", "wreq-js"); + const destinationPackage = path.join( tempDir, ".build", "next", "standalone", "node_modules", - "wreq-js", - "rust", - "wreq-js.linux-x64-gnu.node" + "wreq-js" ); const logs: string[] = []; - - await fs.mkdir(path.dirname(sourceNativeFile), { recursive: true }); - await fs.writeFile(sourceNativeFile, "native module bytes"); - - const changed = await syncStandaloneNativeAssets(tempDir, fs, { + const logger: Console = Object.assign(Object.create(console), { log: (message: unknown) => logs.push(String(message)), }); + await fs.mkdir(path.join(sourcePackage, "dist"), { recursive: true }); + await fs.mkdir(path.join(sourcePackage, "rust"), { recursive: true }); + await fs.writeFile(path.join(sourcePackage, "package.json"), '{"name":"wreq-js"}'); + await fs.writeFile(path.join(sourcePackage, "dist", "wreq-js.cjs"), "exports.fetch = () => {}"); + await fs.writeFile( + path.join(sourcePackage, "rust", "wreq-js.linux-x64-gnu.node"), + "native module bytes" + ); + + const changed = await syncStandaloneExtraModules(tempDir, fs, logger); + assert.equal(changed, true); - assert.equal(await fs.readFile(destinationNativeFile, "utf8"), "native module bytes"); - assert.match((logs[0] ?? "").replaceAll("\\", "/"), /wreq-js\/rust/); + assert.equal( + await fs.readFile(path.join(destinationPackage, "package.json"), "utf8"), + '{"name":"wreq-js"}' + ); + assert.equal( + await fs.readFile(path.join(destinationPackage, "dist", "wreq-js.cjs"), "utf8"), + "exports.fetch = () => {}" + ); + assert.equal( + await fs.readFile( + path.join(destinationPackage, "rust", "wreq-js.linux-x64-gnu.node"), + "utf8" + ), + "native module bytes" + ); + assert.match(logs[0] ?? "", /wreq-js TLS runtime/); }); }); diff --git a/tests/unit/chat-helpers.test.ts b/tests/unit/chat-helpers.test.ts index e27afdb402..b0c21865ea 100644 --- a/tests/unit/chat-helpers.test.ts +++ b/tests/unit/chat-helpers.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import net from "node:net"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-chat-helpers-")); process.env.DATA_DIR = TEST_DATA_DIR; @@ -21,6 +22,8 @@ const { } = await import("../../src/sse/handlers/chatHelpers.ts"); const { getCircuitBreaker, resetAllCircuitBreakers, STATE } = await import("../../src/shared/utils/circuitBreaker.ts"); +// DATA_DIR must be fixed before these modules load; keep this test seam dynamic. +const { setTlsClientForTest } = await import("../../open-sse/utils/proxyFetch.ts"); async function resetStorage() { resetAllCircuitBreakers(); @@ -421,6 +424,96 @@ test("executeChatWithBreaker converts proxy fast-fail errors", async () => { } }); +test("executeChatWithBreaker preserves account TLS scope when a proxy bypasses to direct", async () => { + const server = net.createServer((socket) => socket.end()); + const listening = Promise.withResolvers(); + server.listen(0, "127.0.0.1", listening.resolve); + await listening.promise; + const address = server.address(); + assert.ok(address && typeof address !== "string"); + const prior = { + enable: process.env.ENABLE_TLS_FINGERPRINT, + providers: process.env.TLS_FINGERPRINT_PROVIDERS, + noProxy: process.env.NO_PROXY, + }; + process.env.ENABLE_TLS_FINGERPRINT = "true"; + delete process.env.TLS_FINGERPRINT_PROVIDERS; + process.env.NO_PROXY = "api.openai.com"; + let observedProxy: string | null | undefined; + let observedScope: string | undefined; + setTlsClientForTest({ + available: true, + fetch: async (_url, options) => { + observedProxy = options?.proxy; + observedScope = options?.sessionScope; + return new Response( + JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + created: 0, + model: "gpt-4o-mini", + choices: [ + { + index: 0, + message: { role: "assistant", content: "ok" }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }), + { headers: { "content-type": "application/json" } }, + ); + }, + }); + + try { + const credentials = { + connectionId: "conn_tls_scope", + apiKey: "sk-openai-helper", + providerSpecificData: {}, + }; + const result = await executeChatWithBreaker({ + bypassCircuitBreaker: false, + breaker: getCircuitBreaker("openai"), + body: { model: "openai/gpt-4o-mini", messages: [] }, + provider: "openai", + model: "gpt-4o-mini", + refreshedCredentials: credentials, + proxyInfo: { + proxy: `http://127.0.0.1:${address.port}`, + level: "connection", + levelId: credentials.connectionId, + }, + log: console, + clientRawRequest: null, + credentials, + apiKeyInfo: null, + userAgent: "", + comboName: null, + comboStrategy: null, + isCombo: false, + extendedContext: false, + comboStepId: null, + comboExecutionKey: null, + }); + + assert.equal(result.tlsFingerprintUsed, true); + assert.equal(observedProxy, null); + assert.equal(observedScope, credentials.connectionId); + } finally { + setTlsClientForTest(null); + if (prior.enable === undefined) delete process.env.ENABLE_TLS_FINGERPRINT; + else process.env.ENABLE_TLS_FINGERPRINT = prior.enable; + if (prior.providers === undefined) delete process.env.TLS_FINGERPRINT_PROVIDERS; + else process.env.TLS_FINGERPRINT_PROVIDERS = prior.providers; + if (prior.noProxy === undefined) delete process.env.NO_PROXY; + else process.env.NO_PROXY = prior.noProxy; + const closed = Promise.withResolvers(); + server.close(() => closed.resolve()); + await closed.promise; + } +}); + test("safeLogEvents tolerates success and timeout payloads", () => { const credentials = { connectionId: "conn_log_12345678" }; diff --git a/tests/unit/tls-proxy-context.test.ts b/tests/unit/tls-proxy-context.test.ts new file mode 100644 index 0000000000..57b4a3e08d --- /dev/null +++ b/tests/unit/tls-proxy-context.test.ts @@ -0,0 +1,989 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + proxyFetch, + resolveProxyForRequest, + runWithProxyContext, + runWithTlsTracking, + setTlsClientForTest, +} from "../../open-sse/utils/proxyFetch.ts"; +import tlsClient, { + TlsClient, + type TlsFetchOptions, + type WreqSession, +} from "../../open-sse/utils/tlsClient.ts"; +import { httpBackedChat } from "../../open-sse/services/browserBackedChat.ts"; + +type EnvState = Record; + +const ENV_KEYS = [ + "ENABLE_TLS_FINGERPRINT", + "TLS_FINGERPRINT_PROVIDERS", + "HTTPS_PROXY", + "https_proxy", + "HTTP_PROXY", + "http_proxy", + "ALL_PROXY", + "all_proxy", + "NO_PROXY", + "no_proxy", + "OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK", + "PROXY_AUTO_SELECT_ENABLED", +] as const; + +async function withEnv(env: EnvState, fn: () => Promise | void): Promise { + const prior = Object.fromEntries(ENV_KEYS.map((key) => [key, process.env[key]])); + for (const key of ENV_KEYS) { + if (env[key] === undefined) delete process.env[key]; + else process.env[key] = env[key]; + } + try { + await fn(); + } finally { + for (const key of ENV_KEYS) { + if (prior[key] === undefined) delete process.env[key]; + else process.env[key] = prior[key]; + } + setTlsClientForTest(null); + } +} + +function fakeTlsClient( + fetch: (url: string, options?: TlsFetchOptions) => Promise, +) { + return { available: true, fetch }; +} + +test("explicit direct proxy resolution keeps a session and never rereads the environment", async () => { + await withEnv({ HTTPS_PROXY: "http://placeholder.proxy:8080" }, async () => { + const created: Array> = []; + const client = new TlsClient(async (options) => { + created.push(options); + return { + close: async () => {}, + fetch: async () => new Response("ok"), + }; + }); + + await client.fetch("https://upstream.example", { proxy: null }); + await client.fetch("https://upstream.example", { proxy: null }); + await client.fetch("https://upstream.example"); + + assert.equal(created.length, 2); + assert.equal(created[0]?.proxy, undefined); + assert.equal(created[1]?.proxy, "http://placeholder.proxy:8080"); + }); +}); + +test("same proxy is isolated by stable account session scope", async () => { + const created: Array> = []; + const client = new TlsClient(async (options) => { + created.push(options); + return { + close: async () => {}, + fetch: async () => new Response("ok"), + }; + }); + + const proxy = "http://shared.proxy:8080"; + await client.fetch("https://upstream.example", { proxy, sessionScope: "account-a" }); + await client.fetch("https://upstream.example", { proxy, sessionScope: "account-a" }); + await client.fetch("https://upstream.example", { proxy, sessionScope: "account-b" }); + + assert.equal(created.length, 2); +}); + +test("circuit failures are isolated to the exact session scope and proxy", async () => { + const client = new TlsClient(async (options) => ({ + close: async () => {}, + fetch: async () => { + if (options.proxy === "http://bad.proxy:8080") throw new Error("bad proxy"); + return new Response("good"); + }, + })); + + for (let attempt = 0; attempt < 3; attempt++) { + await assert.rejects( + client.fetch("https://upstream.example", { + proxy: "http://bad.proxy:8080", + sessionScope: "bad-account", + }), + ); + } + + const response = await client.fetch("https://upstream.example", { + proxy: "http://good.proxy:8080", + sessionScope: "good-account", + }); + assert.equal(await response.text(), "good"); +}); + +test("redirect error semantics are forwarded to wreq unchanged", async () => { + let redirect: unknown; + const client = new TlsClient(async () => ({ + close: async () => {}, + fetch: async (_url, options) => { + redirect = options?.redirect; + return new Response("ok"); + }, + })); + + await client.fetch("https://upstream.example", { redirect: "error" }); + assert.equal(redirect, "error"); +}); + +test("Request input bypasses wreq without losing method headers or body", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + let tlsCalls = 0; + let received: Request | null = null; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + return new Response("tls"); + }), + ); + + const input = new Request("https://upstream.example/v1", { + method: "POST", + headers: { "x-test": "present" }, + body: "payload", + }); + const tracked = await runWithTlsTracking("codex", () => + proxyFetch(input, {}, { + undiciFetch: async (forwarded) => { + received = forwarded as Request; + return new Response("dispatcher"); + }, + }), + ); + + assert.equal(tlsCalls, 0); + assert.equal(received, input); + assert.equal(received?.method, "POST"); + assert.equal(received?.headers.get("x-test"), "present"); + assert.equal(await received?.text(), "payload"); + assert.equal(tracked.tlsFingerprintUsed, false); + }, + ); +}); + +test("non-idempotent TLS failures are never replayed", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + throw new Error("post-send transport failure"); + }), + ); + + await assert.rejects( + runWithTlsTracking("codex", () => + proxyFetch( + "https://upstream.example/v1", + { method: "POST", body: "{}" }, + { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("unexpected"); + }, + }, + ), + ), + (error: Error & { code?: string }) => + error.code === "TLS_FINGERPRINT_FAILED" && + error.message === "TLS fingerprint request failed; request is not safe to replay", + ); + assert.equal(dispatcherCalls, 0); + }, + ); +}); + +test("safe GET TLS failure falls back through the same configured proxy", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + HTTPS_PROXY: "http://placeholder.proxy:8080", + }, + async () => { + setTlsClientForTest( + fakeTlsClient(async () => { + throw new Error("transport failed"); + }), + ); + let dispatcher: unknown; + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async (_input, init) => { + dispatcher = init?.dispatcher; + return new Response("fallback"); + }, + }), + ); + + assert.ok(dispatcher); + assert.equal(await tracked.result.text(), "fallback"); + assert.equal(tracked.tlsFingerprintUsed, false); + }, + ); +}); + +test("internal TimeoutError is not classified as a caller abort", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + const timeout = new Error("internal timeout"); + timeout.name = "TimeoutError"; + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + throw timeout; + }), + ); + + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("fallback"); + }, + }), + ); + + assert.equal(dispatcherCalls, 1); + assert.equal(await tracked.result.text(), "fallback"); + }, + ); +}); + +test("control-plane direct fallback bypasses an environment proxy", async () => { + await withEnv( + { + HTTPS_PROXY: "http://placeholder.proxy:8080", + OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK: "true", + }, + async () => { + const result = await runWithProxyContext( + { type: "http", host: "127.0.0.1", port: "9" }, + () => resolveProxyForRequest("https://upstream.example/v1"), + { directFallbackOnUnreachable: true }, + ); + + assert.deepEqual(result, { source: "direct", proxyUrl: null }); + }, + ); +}); + +test("new proxied TLS transport requires an explicit provider allowlist", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: undefined, + HTTPS_PROXY: "http://placeholder.proxy:8080", + }, + async () => { + let tlsCalls = 0; + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + return new Response("tls"); + }), + ); + + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("dispatcher"); + }, + }), + ); + + assert.equal(tlsCalls, 0); + assert.equal(dispatcherCalls, 1); + assert.equal(await tracked.result.text(), "dispatcher"); + }, + ); +}); + +test("caller abort propagates unchanged and never falls back", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + const controller = new AbortController(); + const abortError = new Error("caller stopped"); + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + controller.abort(abortError); + throw abortError; + }), + ); + + await assert.rejects( + runWithTlsTracking("codex", () => + proxyFetch( + "https://upstream.example/v1", + { signal: controller.signal }, + { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("unexpected"); + }, + }, + ), + ), + (error) => error === abortError, + ); + assert.equal(dispatcherCalls, 0); + }, + ); +}); + +test("stateful TLS session failures never fall back even for GET", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + const error = new Error("transport failed"); + Object.defineProperty(error, "sessionHadCookies", { value: true }); + throw error; + }), + ); + + await assert.rejects( + runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("unexpected"); + }, + }), + ), + (error: Error & { code?: string }) => + error.code === "TLS_FINGERPRINT_FAILED" && + error.message === "TLS fingerprint request failed; stateful session cannot be replayed", + ); + assert.equal(dispatcherCalls, 0); + }, + ); +}); + +test("TLS transport failures never expose proxy credentials", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + HTTPS_PROXY: "http://user:password@placeholder.proxy:8080", + }, + async () => { + const warnings: string[] = []; + const originalWarn = console.warn; + console.warn = (...args: unknown[]) => warnings.push(args.map(String).join(" ")); + try { + setTlsClientForTest( + fakeTlsClient(async () => { + throw new Error( + "connect failed via http://user:password@placeholder.proxy:8080", + ); + }), + ); + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => new Response("fallback"), + }), + ); + assert.equal(await tracked.result.text(), "fallback"); + assert.equal(warnings.some((line) => line.includes("user:password")), false); + assert.equal(warnings.some((line) => line.includes("placeholder.proxy")), false); + } finally { + console.warn = originalWarn; + } + }, + ); +}); + +test("family-pinned proxies retain dispatcher enforcement instead of using wreq", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + HTTPS_PROXY: "http://placeholder.proxy:8080?family=ipv4", + }, + async () => { + let tlsCalls = 0; + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + return new Response("tls"); + }), + ); + + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("dispatcher"); + }, + }), + ); + assert.equal(tlsCalls, 0); + assert.equal(dispatcherCalls, 1); + assert.equal(await tracked.result.text(), "dispatcher"); + }, + ); +}); + +test("relay contexts never route through wreq", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + }, + async () => { + let tlsCalls = 0; + let relayCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + return new Response("tls"); + }), + ); + + const tracked = await runWithTlsTracking("codex", () => + runWithProxyContext( + { type: "vercel", host: "relay.example", relayAuth: "test-auth" }, + () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + relayCalls++; + return new Response("relay"); + }, + }), + ), + ); + assert.equal(tlsCalls, 0); + assert.equal(relayCalls, 1); + assert.equal(await tracked.result.text(), "relay"); + }, + ); +}); + +test("wreq responses are adapted to the native Response API", async () => { + const sourceBody = new Response("adapted").body; + assert.ok(sourceBody); + const client = new TlsClient(async () => ({ + close: async () => {}, + fetch: async () => ({ + status: 200, + statusText: "OK", + headers: new Map([["x-source", "wreq"]]), + body: sourceBody, + url: "https://upstream.example/v1", + redirected: true, + }), + })); + + const response = await client.fetch("https://upstream.example/v1", { proxy: null }); + assert.equal(response instanceof Response, true); + assert.equal(response.headers.get("x-source"), "wreq"); + assert.equal(response.url, "https://upstream.example/v1"); + assert.equal(response.redirected, true); + assert.equal(await response.text(), "adapted"); +}); + +test("exit waits for pending session creation and closes the late session", async () => { + type TestSession = { + close: () => Promise; + fetch: () => Promise; + }; + const sessionGate = Promise.withResolvers(); + const creationStarted = Promise.withResolvers(); + const closeStarted = Promise.withResolvers(); + const closeGate = Promise.withResolvers(); + let closed = 0; + const client = new TlsClient(() => { + creationStarted.resolve(); + return sessionGate.promise; + }); + const request = client.fetch("https://upstream.example/v1", { proxy: null }); + await creationStarted.promise; + + let exitSettled = false; + const exiting = client.exit().then(() => { + exitSettled = true; + }); + sessionGate.resolve({ + close: async () => { + closed++; + closeStarted.resolve(); + await closeGate.promise; + }, + fetch: async () => new Response("unexpected"), + }); + await closeStarted.promise; + assert.equal(exitSettled, false); + + closeGate.resolve(); + await assert.rejects(request, /wreq-js transport failed/); + await exiting; + assert.equal(closed, 1); +}); + +test("bounded session cache closes the least-recently-used idle session", async () => { + const closed: string[] = []; + const client = new TlsClient(async (options) => { + const proxy = String(options.proxy); + return { + close: async () => { + closed.push(proxy); + }, + fetch: async () => new Response("ok"), + }; + }, 2); + + await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-1:8080" }); + await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-2:8080" }); + await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-3:8080" }); + + assert.deepEqual(closed, ["http://proxy-1:8080"]); + await client.exit(); +}); + +test("direct browser-backed TLS calls isolate sessions by pool key", async () => { + const originalFetch = tlsClient.fetch.bind(tlsClient); + let observedScope: string | undefined; + tlsClient.fetch = async (_url, options) => { + observedScope = options?.sessionScope; + return new Response("ok", { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + try { + const result = await httpBackedChat({ + poolKey: "claude-web:account-123", + chatUrl: "https://claude.ai/api/chat", + chatPageUrl: "https://claude.ai/new", + userMessage: "hello", + chatUrlMatchDomain: "claude.ai", + inputSelector: "#prompt", + }); + assert.equal(result.status, 200); + assert.equal(observedScope, "claude-web:account-123"); + } finally { + tlsClient.fetch = originalFetch; + } +}); + +test("allowlisted proxied TLS receives the exact proxy and account scope", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + HTTPS_PROXY: "http://placeholder.proxy:8080", + }, + async () => { + let observedOptions: TlsFetchOptions | undefined; + setTlsClientForTest( + fakeTlsClient(async (_url, options) => { + observedOptions = options; + return new Response("tls"); + }), + ); + const tracked = await runWithTlsTracking( + { provider: "codex", sessionScope: "connection-123" }, + () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + throw new Error("dispatcher must not run"); + }, + }), + ); + + assert.equal(observedOptions?.proxy, "http://placeholder.proxy:8080"); + assert.equal(observedOptions?.sessionScope, "connection-123"); + assert.equal(tracked.tlsFingerprintUsed, true); + assert.equal(await tracked.result.text(), "tls"); + }, + ); +}); + +test("proxy dispatcher failures sanitize logs and propagated errors", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "false", + HTTPS_PROXY: "http://user:password@placeholder.proxy:8080", + }, + async () => { + const errors: string[] = []; + const originalError = console.error; + console.error = (...args: unknown[]) => errors.push(args.map(String).join(" ")); + try { + let caught: unknown; + try { + await proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + const error = new Error( + "connect failed via http://user:password@placeholder.proxy:8080", + ) as Error & { code?: string }; + error.code = "ECONNREFUSED"; + throw error; + }, + }); + } catch (error) { + caught = error; + } + assert.ok(caught instanceof Error); + assert.equal(caught.message, "Proxy request failed"); + assert.equal("code" in caught ? caught.code : undefined, "PROXY_UNREACHABLE"); + assert.equal(errors.some((line) => line.includes("user:password")), false); + assert.equal(errors.some((line) => line.includes("placeholder.proxy")), false); + } finally { + console.error = originalError; + } + }, + ); +}); + +test("half-open circuit admits only one probe for an isolated session key", async () => { + const originalNow = Date.now; + const probeStarted = Promise.withResolvers(); + const probeGate = Promise.withResolvers(); + let probeMode = false; + let fetchCalls = 0; + const client = new TlsClient(async () => ({ + close: async () => {}, + fetch: async () => { + fetchCalls++; + if (!probeMode) throw new Error("upstream unavailable"); + probeStarted.resolve(); + await probeGate.promise; + return new Response("recovered"); + }, + })); + + try { + for (let attempt = 0; attempt < 3; attempt++) { + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }), + /wreq-js transport failed/, + ); + } + probeMode = true; + const afterCooldown = originalNow() + 31_000; + Date.now = () => afterCooldown; + + const probe = client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }); + await probeStarted.promise; + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }), + (error: unknown) => + error instanceof Error && + "code" in error && + error.code === "TLS_CIRCUIT_OPEN", + ); + assert.equal(fetchCalls, 4); + probeGate.resolve(); + assert.equal(await (await probe).text(), "recovered"); + } finally { + Date.now = originalNow; + probeGate.resolve(); + await client.exit(); + } +}); + +test("circuit invalidation snapshots cookies before closing the failed session", async () => { + let closed = false; + const client = new TlsClient(async () => ({ + close: () => { + closed = true; + }, + getCookies: () => (closed ? {} : { session: "account-a" }), + fetch: async () => { + throw new Error("upstream unavailable"); + }, + })); + + try { + for (let attempt = 0; attempt < 3; attempt++) { + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }), + (error: unknown) => + error instanceof Error && + "sessionHadCookies" in error && + error.sessionHadCookies === true, + ); + } + await Promise.resolve(); + assert.equal(closed, true); + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }), + (error: unknown) => + error instanceof Error && + "code" in error && + error.code === "TLS_CIRCUIT_OPEN" && + "sessionHadCookies" in error && + error.sessionHadCookies === true, + ); + } finally { + await client.exit(); + } +}); + +test("pending session creation is bounded per TLS client", async () => { + const sessionGates = [ + Promise.withResolvers(), + Promise.withResolvers(), + ]; + let creates = 0; + const client = new TlsClient(() => { + const gate = sessionGates[creates++]; + if (!gate) throw new Error("unexpected session creation"); + return gate.promise; + }, 2); + const session: WreqSession = { + close: async () => {}, + fetch: async () => new Response("ok"), + }; + + const first = client.fetch("https://upstream.example/v1", { + proxy: "http://proxy-1:8080", + sessionScope: "connection-1", + }); + const second = client.fetch("https://upstream.example/v1", { + proxy: "http://proxy-2:8080", + sessionScope: "connection-2", + }); + try { + assert.equal(creates, 2); + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: "http://proxy-3:8080", + sessionScope: "connection-3", + }), + (error: unknown) => + error instanceof Error && + "code" in error && + error.code === "TLS_SESSION_CAPACITY", + ); + assert.equal(creates, 2); + + for (const gate of sessionGates) gate.resolve(session); + assert.equal(await (await first).text(), "ok"); + assert.equal(await (await second).text(), "ok"); + } finally { + for (const gate of sessionGates) gate.resolve(session); + await Promise.allSettled([first, second]); + await client.exit(); + } +}); + +test("direct TLS fallback never auto-selects a different proxy route", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: undefined, + PROXY_AUTO_SELECT_ENABLED: "true", + }, + async () => { + let tlsCalls = 0; + let dispatcherCalls = 0; + let autoSelectCalls = 0; + let nativeCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + throw new Error("wreq transport failed"); + }), + ); + + const response = await proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + const error = new Error("fetch failed: ECONNREFUSED") as Error & { code?: string }; + error.code = "ECONNREFUSED"; + throw error; + }, + findWorkingProxy: async () => { + autoSelectCalls++; + return "http://unexpected.proxy:8080"; + }, + nativeFetch: async () => { + nativeCalls++; + return new Response("native"); + }, + }); + + assert.equal(tlsCalls, 1); + assert.equal(dispatcherCalls, 2); + assert.equal(autoSelectCalls, 0); + assert.equal(nativeCalls, 1); + assert.equal(await response.text(), "native"); + }, + ); +}); + +test("proxied TLS compatibility overload requires an explicit session scope", async () => { + await withEnv( + { + ENABLE_TLS_FINGERPRINT: "true", + TLS_FINGERPRINT_PROVIDERS: "codex", + HTTPS_PROXY: "http://placeholder.proxy:8080", + }, + async () => { + let tlsCalls = 0; + let dispatcherCalls = 0; + setTlsClientForTest( + fakeTlsClient(async () => { + tlsCalls++; + return new Response("unexpected"); + }), + ); + + const tracked = await runWithTlsTracking("codex", () => + proxyFetch("https://upstream.example/v1", {}, { + undiciFetch: async () => { + dispatcherCalls++; + return new Response("dispatcher"); + }, + }), + ); + + assert.equal(tlsCalls, 0); + assert.equal(dispatcherCalls, 1); + assert.equal(tracked.tlsFingerprintUsed, false); + assert.equal(await tracked.result.text(), "dispatcher"); + }, + ); +}); + +test("circuit trip defers session close until active response streams release", async () => { + let first = true; + let closed = 0; + const client = new TlsClient(async () => ({ + close: async () => { + closed++; + }, + fetch: async () => { + if (first) { + first = false; + return { + status: 200, + statusText: "OK", + headers: [], + body: new ReadableStream({}), + }; + } + throw new Error("upstream unavailable"); + }, + })); + + try { + const activeResponse = await client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }); + for (let attempt = 0; attempt < 3; attempt++) { + await assert.rejects( + client.fetch("https://upstream.example/v1", { + proxy: null, + sessionScope: "connection-123", + }), + /wreq-js transport failed/, + ); + } + assert.equal(closed, 0); + + await activeResponse.body?.cancel(); + await Promise.resolve(); + assert.equal(closed, 1); + } finally { + await client.exit(); + } +}); + +test("streaming wreq body failures are sanitized and counted by the circuit", async () => { + const secret = "http://user:password@proxy.example:8080"; + const client = new TlsClient(async () => ({ + close: async () => {}, + fetch: async () => ({ + status: 200, + statusText: "OK", + headers: [["content-type", "text/plain"]], + body: new ReadableStream({ + pull(controller) { + const error = new Error(`body failed through ${secret}`) as Error & { + code?: string; + }; + error.code = "UND_ERR_SOCKET"; + controller.error(error); + }, + }), + }), + })); + + try { + const response = await client.fetch("https://upstream.example/v1", { + proxy: "http://user:password@proxy.example:8080", + sessionScope: "connection-123", + }); + await assert.rejects( + response.text(), + (error: unknown) => + error instanceof Error && + error.message === "wreq-js response body failed" && + "code" in error && + error.code === "UND_ERR_SOCKET" && + !String(error).includes("user:password"), + ); + assert.equal( + client.getCircuitState( + "http://user:password@proxy.example:8080", + "connection-123", + ).failureCount, + 1, + ); + } finally { + await client.exit(); + } +});