From f2d7df1f94cfd9d35b459d586ae499c06e50f4f5 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Fri, 12 Jun 2026 00:25:13 -0300 Subject: [PATCH] feat(proxy): egress IP visibility + pool validation (codex anomaly) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The proxy logs captured the INBOUND client IP (x-forwarded-for) but never the OUTBOUND/egress IP — so there was no way to confirm by which IP each account leaves. For rotating providers that is the whole game: probing the servers showed every codex account had a distinct proxy assigned, but ALL proxies were dead (egress timeout) while still marked status=active, so accounts fell back to the shared host IPv6 /64 (same /64 across .16/.17/.18) → OpenAI anomaly revocation ("authentication token has been invalidated"). - src/lib/proxyEgress.ts: resolveEgressIp() (echo-IP via the resolved proxy, cached), analyzeEgressSharing() (flags >=2 same-rotation-group accounts on one egress IP), diagnoseAllEgressIps(), and validateProxyPool() (probes each proxy and persists status=active/error so the dead-proxy filter takes them out of rotation automatically). - proxyLogger: new egressIp field + a structured [ProxyEgress] line (in=clientIp out=egressIp) so the proxy logs show entry AND exit IP. - chatHelpers: populate egressIp from a non-blocking cache, warm in background. - GET/POST /api/settings/proxies/egress: diagnose + validate from the dashboard. TDD: 5 tests (egress resolve/cache, sharing analysis, diagnose wiring, pool validation). 37/37 proxy+oauth tests green, typecheck:core=0, lint clean. --- src/app/api/settings/proxies/egress/route.ts | 42 +++ src/lib/proxyEgress.ts | 318 +++++++++++++++++++ src/lib/proxyLogger.ts | 16 + src/sse/handlers/chatHelpers.ts | 15 + tests/unit/proxy-egress-visibility.test.ts | 135 ++++++++ 5 files changed, 526 insertions(+) create mode 100644 src/app/api/settings/proxies/egress/route.ts create mode 100644 src/lib/proxyEgress.ts create mode 100644 tests/unit/proxy-egress-visibility.test.ts diff --git a/src/app/api/settings/proxies/egress/route.ts b/src/app/api/settings/proxies/egress/route.ts new file mode 100644 index 0000000000..b43fcd6e6f --- /dev/null +++ b/src/app/api/settings/proxies/egress/route.ts @@ -0,0 +1,42 @@ +import { NextResponse } from "next/server"; +import { requireManagementAuth } from "@/lib/api/requireManagementAuth"; +import { createErrorResponseFromUnknown } from "@/lib/api/errorResponse"; +import { diagnoseAllEgressIps, validateProxyPool } from "@/lib/proxyEgress"; + +/** + * GET /api/settings/proxies/egress — diagnose the egress IP of every OAuth + * connection: by which IP each account is entering (clientIp) and leaving + * (egressIp), plus warnings for same-rotation-group accounts sharing one + * egress IP (the codex anomaly-revocation trigger). + * + * POST /api/settings/proxies/egress — validate the whole proxy pool by probing + * each proxy's real egress IP and persisting status=active/error, so dead + * proxies are taken out of rotation automatically. + */ +export async function GET(request: Request) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const diagnostic = await diagnoseAllEgressIps(); + return NextResponse.json(diagnostic); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to diagnose egress IPs"); + } +} + +export async function POST(request: Request) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const report = await validateProxyPool(); + const dead = report.filter((r) => !r.alive); + return NextResponse.json({ + validated: report.length, + alive: report.length - dead.length, + dead: dead.length, + report, + }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to validate proxy pool"); + } +} diff --git a/src/lib/proxyEgress.ts b/src/lib/proxyEgress.ts new file mode 100644 index 0000000000..c67243f60d --- /dev/null +++ b/src/lib/proxyEgress.ts @@ -0,0 +1,318 @@ +/** + * Proxy Egress IP visibility. + * + * The proxy logs already capture the INBOUND client IP (x-forwarded-for), but + * NOT the OUTBOUND/egress IP — the address the upstream actually sees. For + * rotating providers (codex/openai) this is critical: when several accounts + * egress through the SAME IP at high volume, the provider flags it as anomaly + * and revokes the tokens ("Your authentication token has been invalidated"). + * + * This module resolves the real egress IP (via an echo-IP service through the + * resolved proxy/dispatcher) and detects same-rotation-group accounts sharing + * an egress IP, so the operator can confirm exactly which IP each account is + * entering and leaving by. + */ +import { request as undiciRequest } from "undici"; +import { createProxyDispatcher, proxyConfigToUrl } from "@omniroute/open-sse/utils/proxyDispatcher.ts"; +import { rotationGroupFor } from "@omniroute/open-sse/services/refreshSerializer.ts"; + +const EGRESS_ECHO_URL = "https://api64.ipify.org?format=json"; +const EGRESS_PROBE_TIMEOUT_MS = 6000; +const EGRESS_CACHE_TTL_MS = 5 * 60 * 1000; + +export interface EgressProbeResult { + ip: string | null; + latencyMs: number; + error?: string; +} + +export type EgressProbe = (proxyUrl: string | null) => Promise; + +const egressCache = new Map(); + +async function defaultEgressProbe(proxyUrl: string | null): Promise { + const start = Date.now(); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), EGRESS_PROBE_TIMEOUT_MS); + try { + const dispatcher = proxyUrl ? createProxyDispatcher(proxyUrl) : undefined; + const res = await undiciRequest(EGRESS_ECHO_URL, { + method: "GET", + dispatcher, + signal: controller.signal, + headersTimeout: EGRESS_PROBE_TIMEOUT_MS, + bodyTimeout: EGRESS_PROBE_TIMEOUT_MS, + }); + const text = await res.body.text(); + let ip: string | null = null; + try { + ip = (JSON.parse(text) as { ip?: string }).ip ?? null; + } catch { + // non-JSON body — leave ip null + } + return { ip, latencyMs: Date.now() - start }; + } catch (error) { + return { + ip: null, + latencyMs: Date.now() - start, + error: error instanceof Error ? error.message : String(error), + }; + } finally { + clearTimeout(timeout); + } +} + +let probe: EgressProbe = defaultEgressProbe; + +/** Test seam: override the network probe. */ +export function _setEgressProbeForTests(fn: EgressProbe | null): void { + probe = fn ?? defaultEgressProbe; +} + +export function clearEgressCache(): void { + egressCache.clear(); +} + +/** + * Synchronous read of the cached egress IP for a proxy URL (null = direct). + * Non-blocking — used by the request hot path to log the egress IP without an + * echo-IP round-trip. Returns null if not yet probed. + */ +export function getCachedEgressIp(proxyUrl: string | null): string | null { + const cached = egressCache.get(proxyUrl ?? "__direct__"); + if (!cached) return null; + if (Date.now() - cached.at >= EGRESS_CACHE_TTL_MS) return null; + return cached.ip; +} + +const warmingInFlight = new Set(); + +/** + * Fire-and-forget: populate the egress cache for a proxy URL in the background + * so subsequent proxy log lines carry the real egress IP. Deduped per URL. + */ +export function warmEgressIp(proxyUrl: string | null): void { + const key = proxyUrl ?? "__direct__"; + if (warmingInFlight.has(key) || getCachedEgressIp(proxyUrl) !== null) return; + warmingInFlight.add(key); + void resolveEgressIp(proxyUrl) + .catch(() => undefined) + .finally(() => warmingInFlight.delete(key)); +} + +/** + * Resolve the egress IP for a given proxy URL (null = direct/host IP). + * Cached per proxyUrl to avoid an echo-IP round-trip on every call. + */ +export async function resolveEgressIp( + proxyUrl: string | null, + opts: { cacheTtlMs?: number; force?: boolean } = {} +): Promise { + const key = proxyUrl ?? "__direct__"; + const ttl = opts.cacheTtlMs ?? EGRESS_CACHE_TTL_MS; + const cached = egressCache.get(key); + if (!opts.force && cached && Date.now() - cached.at < ttl) { + return { ip: cached.ip, latencyMs: 0, cached: true }; + } + const result = await probe(proxyUrl); + egressCache.set(key, { ip: result.ip, at: Date.now() }); + return { ...result, cached: false }; +} + +export interface ConnectionEgress { + connectionId: string; + provider: string; + account: string | null; + proxyLevel: string; + proxyHost: string | null; + egressIp: string | null; + error?: string; +} + +export interface EgressSharingWarning { + egressIp: string; + rotationGroup: string; + connections: string[]; // connectionId/account labels sharing this IP within one rotation group +} + +export interface EgressDiagnostic { + connections: ConnectionEgress[]; + byEgressIp: Record; + sharedWithinRotationGroup: EgressSharingWarning[]; +} + +/** + * PURE: group egress results by IP and flag IPs shared by ≥2 accounts of the + * SAME rotation group (codex+openai share one Auth0 family — the exact + * condition that triggers anomaly revocation). Direct/unknown IPs are reported + * but only same-group sharing is a warning. + */ +export function analyzeEgressSharing(connections: ConnectionEgress[]): { + byEgressIp: Record; + sharedWithinRotationGroup: EgressSharingWarning[]; +} { + const byEgressIp: Record = {}; + // ip -> rotationGroup -> labels + const byIpGroup = new Map>(); + + for (const c of connections) { + if (!c.egressIp) continue; + const label = c.account || c.connectionId; + (byEgressIp[c.egressIp] ??= []).push(label); + + const group = rotationGroupFor(c.provider) || `provider:${c.provider}`; + let groups = byIpGroup.get(c.egressIp); + if (!groups) { + groups = new Map(); + byIpGroup.set(c.egressIp, groups); + } + const list = groups.get(group) ?? []; + list.push(label); + groups.set(group, list); + } + + const sharedWithinRotationGroup: EgressSharingWarning[] = []; + for (const [egressIp, groups] of byIpGroup) { + for (const [rotationGroup, labels] of groups) { + if (labels.length >= 2) { + sharedWithinRotationGroup.push({ egressIp, rotationGroup, connections: labels }); + } + } + } + + return { byEgressIp, sharedWithinRotationGroup }; +} + +/** + * Diagnose egress IPs for every OAuth connection: resolve each connection's + * proxy, probe the real egress IP, and flag same-rotation-group IP sharing. + */ +export async function diagnoseAllEgressIps(deps?: { + getConnections?: () => Promise< + Array<{ id: string; provider: string; name?: string; email?: string; authType?: string }> + >; + resolveProxy?: ( + connectionId: string + ) => Promise<{ proxy?: unknown; level?: string } | null>; +}): Promise { + const getConnections = + deps?.getConnections ?? + (async () => { + const { getProviderConnections } = await import("./localDb"); + return (await getProviderConnections({ authType: "oauth" })) as Array<{ + id: string; + provider: string; + name?: string; + email?: string; + }>; + }); + const resolveProxy = + deps?.resolveProxy ?? + (async (connectionId: string) => { + const { resolveProxyForConnection } = await import("./db/settings"); + return resolveProxyForConnection(connectionId); + }); + + const conns = await getConnections(); + const results: ConnectionEgress[] = []; + + for (const c of conns) { + const resolved = await resolveProxy(c.id); + const proxyObj = (resolved?.proxy ?? null) as { + type?: string; + host?: string; + port?: number | string; + } | null; + const proxyUrl = proxyObj ? proxyConfigToUrl(proxyObj) : null; + const egress = await resolveEgressIp(proxyUrl); + results.push({ + connectionId: c.id, + provider: c.provider, + account: c.email || c.name || c.id.slice(0, 8), + proxyLevel: resolved?.level || "direct", + proxyHost: proxyObj?.host ?? null, + egressIp: egress.ip, + ...(egress.error ? { error: egress.error } : {}), + }); + } + + const { byEgressIp, sharedWithinRotationGroup } = analyzeEgressSharing(results); + return { connections: results, byEgressIp, sharedWithinRotationGroup }; +} + +export interface ProxyValidationResult { + proxyId: string; + host: string; + port: number | string; + alive: boolean; + egressIp: string | null; + latencyMs: number; + previousStatus: string | null; + newStatus: "active" | "error"; +} + +/** + * Validate every proxy in the registry by probing its real egress IP, and + * persist the result to `proxy_registry.status` (active/error). Combined with + * PROXY_ALIVE_PREDICATE in resolution, a dead proxy is automatically taken out + * of rotation — fixing the "all proxies marked active but actually dead" state + * that left codex accounts falling back to the shared host /64 IP. + * + * Deps are injectable for tests. + */ +export async function validateProxyPool(deps?: { + listProxies?: () => Promise< + Array<{ id: string; type: string; host: string; port: number | string; username?: string | null; password?: string | null; status?: string | null }> + >; + markStatus?: (id: string, status: string, meta: { latencyMs: number; egressIp: string | null }) => Promise; +}): Promise { + const listProxies = + deps?.listProxies ?? + (async () => { + const { listProxies: real } = await import("./db/proxies"); + return (await real({ includeSecrets: true })) as Array<{ + id: string; + type: string; + host: string; + port: number | string; + username?: string | null; + password?: string | null; + status?: string | null; + }>; + }); + const markStatus = + deps?.markStatus ?? + (async (id: string, status: string) => { + const { updateProxy } = await import("./db/proxies"); + await updateProxy(id, { status }); + }); + + const proxies = await listProxies(); + const report: ProxyValidationResult[] = []; + + for (const p of proxies) { + const url = proxyConfigToUrl({ + type: p.type, + host: p.host, + port: p.port, + username: p.username ?? undefined, + password: p.password ?? undefined, + }); + const probe = await resolveEgressIp(url, { force: true }); + const alive = !!probe.ip && !probe.error; + const newStatus: "active" | "error" = alive ? "active" : "error"; + await markStatus(p.id, newStatus, { latencyMs: probe.latencyMs, egressIp: probe.ip }); + report.push({ + proxyId: p.id, + host: p.host, + port: p.port, + alive, + egressIp: probe.ip, + latencyMs: probe.latencyMs, + previousStatus: p.status ?? null, + newStatus, + }); + } + + return report; +} diff --git a/src/lib/proxyLogger.ts b/src/lib/proxyLogger.ts index 2fd45e3abc..19cd565f14 100644 --- a/src/lib/proxyLogger.ts +++ b/src/lib/proxyLogger.ts @@ -29,6 +29,10 @@ interface ProxyLogEntry { provider: string | null; targetUrl: string | null; clientIp: string | null; + /** Outbound/egress IP the upstream actually saw (null until probed). The + * historical clientIp is the INBOUND IP (x-forwarded-for); egressIp answers + * "by which IP is this account leaving" — critical for rotating providers. */ + egressIp: string | null; latencyMs: number; error: string | null; connectionId: string | null; @@ -77,6 +81,7 @@ function loadFromDb() { provider: row.provider || null, targetUrl: row.target_url || null, clientIp: row.public_ip || null, + egressIp: row.egress_ip || null, latencyMs: row.latency_ms || 0, error: row.error || null, connectionId: row.connection_id || null, @@ -109,6 +114,7 @@ export function logProxyEvent(entry: ProxyLogInput) { provider: entry.provider || null, targetUrl: entry.targetUrl || null, clientIp: entry.clientIp ?? entry.publicIp ?? null, + egressIp: entry.egressIp ?? null, latencyMs: entry.latencyMs || 0, error: entry.error || null, connectionId: entry.connectionId || null, @@ -117,6 +123,16 @@ export function logProxyEvent(entry: ProxyLogInput) { tlsFingerprint: entry.tlsFingerprint || false, }; + // Structured egress line so the operator can confirm, in the proxy logs, which + // IP each account is entering (clientIp) and leaving (egressIp) by. + if (log.proxy || log.egressIp) { + console.log( + `[ProxyEgress] ${log.provider || "-"}/${log.account || "-"} ` + + `in=${log.clientIp || "?"} out=${log.egressIp || "?"} ` + + `proxy=${log.level}${log.proxy ? `:${log.proxy.host}` : ""} status=${log.status}` + ); + } + // 1. In-memory ring buffer (newest first) proxyLogs.unshift(log); if (proxyLogs.length > MAX_IN_MEMORY_ENTRIES) { diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index bf7e7ac232..05dfa957c2 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -613,6 +613,20 @@ export function safeLogEvents({ const rawIpValue = Array.isArray(rawIp) ? rawIp[0] : rawIp; const clientIp = typeof rawIpValue === "string" ? rawIpValue.split(",")[0].trim() : null; + // Resolve the egress IP (the IP the upstream actually saw) from cache — never + // blocking the request. Warm it in the background for next time. null until + // the first warm completes; direct (no proxy) is also tracked. + let egressIp: string | null = null; + try { + const { getCachedEgressIp, warmEgressIp } = await import("../../lib/proxyEgress"); + const { proxyConfigToUrl } = await import("@omniroute/open-sse/utils/proxyDispatcher.ts"); + const proxyUrl = proxyInfo?.proxy ? proxyConfigToUrl(proxyInfo.proxy) : null; + egressIp = getCachedEgressIp(proxyUrl); + warmEgressIp(proxyUrl); + } catch { + // egress visibility is best-effort; never break the request path + } + logProxyEvent({ status: result.success ? "success" @@ -625,6 +639,7 @@ export function safeLogEvents({ provider, targetUrl: `${provider}/${model}`, clientIp, + egressIp, latencyMs: proxyLatency, error: result.success ? null : result.error || null, connectionId: credentials.connectionId, diff --git a/tests/unit/proxy-egress-visibility.test.ts b/tests/unit/proxy-egress-visibility.test.ts new file mode 100644 index 0000000000..7ed83550fa --- /dev/null +++ b/tests/unit/proxy-egress-visibility.test.ts @@ -0,0 +1,135 @@ +/** + * TDD — proxy egress IP visibility. Confirms by which IP each OAuth connection + * leaves, and flags same-rotation-group accounts sharing one egress IP (the + * exact codex anomaly-revocation trigger). Network probe is injected. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const egress = await import("../../src/lib/proxyEgress.ts"); +const { + resolveEgressIp, + analyzeEgressSharing, + diagnoseAllEgressIps, + _setEgressProbeForTests, + clearEgressCache, +} = egress as unknown as { + resolveEgressIp: (u: string | null, o?: any) => Promise<{ ip: string | null; cached: boolean }>; + analyzeEgressSharing: (c: any[]) => { + byEgressIp: Record; + sharedWithinRotationGroup: Array<{ egressIp: string; rotationGroup: string; connections: string[] }>; + }; + diagnoseAllEgressIps: (deps?: any) => Promise; + validateProxyPool: (deps?: any) => Promise; + _setEgressProbeForTests: (fn: any) => void; + clearEgressCache: () => void; +}; +const { validateProxyPool } = egress as any; + +test("resolveEgressIp returns the probed IP and caches by proxy URL", async () => { + clearEgressCache(); + let calls = 0; + _setEgressProbeForTests(async (proxyUrl: string | null) => { + calls++; + return { ip: proxyUrl ? "203.0.113.7" : "198.51.100.1", latencyMs: 5 }; + }); + + const viaProxy = await resolveEgressIp("http://1.2.3.4:8080"); + assert.equal(viaProxy.ip, "203.0.113.7"); + assert.equal(viaProxy.cached, false); + + const direct = await resolveEgressIp(null); + assert.equal(direct.ip, "198.51.100.1"); + + // second call for the same proxy URL is served from cache (no extra probe) + const again = await resolveEgressIp("http://1.2.3.4:8080"); + assert.equal(again.cached, true); + assert.equal(calls, 2, "only 2 distinct probes (proxy + direct), not 3"); + + _setEgressProbeForTests(null); +}); + +test("analyzeEgressSharing flags ≥2 same-rotation-group accounts on one egress IP", () => { + const result = analyzeEgressSharing([ + { connectionId: "a", provider: "codex", account: "acc-a", proxyLevel: "direct", proxyHost: null, egressIp: "100.115.194.84" }, + { connectionId: "b", provider: "codex", account: "acc-b", proxyLevel: "direct", proxyHost: null, egressIp: "100.115.194.84" }, + { connectionId: "c", provider: "openai", account: "acc-c", proxyLevel: "direct", proxyHost: null, egressIp: "100.115.194.84" }, + { connectionId: "d", provider: "claude", account: "acc-d", proxyLevel: "direct", proxyHost: null, egressIp: "100.115.194.84" }, + ]); + + // codex + openai share the openai-auth0 family → 3 accounts on one IP = warning + const warn = result.sharedWithinRotationGroup.find((w) => w.rotationGroup === "openai-auth0"); + assert.ok(warn, "must warn about the codex/openai family sharing an egress IP"); + assert.equal(warn!.egressIp, "100.115.194.84"); + assert.equal(warn!.connections.length, 3, "acc-a + acc-b + acc-c"); + + // claude alone on the IP is NOT a warning (different family, single account) + assert.ok( + !result.sharedWithinRotationGroup.some((w) => w.connections.includes("acc-d")), + "a lone claude account must not be flagged" + ); + + assert.deepEqual(result.byEgressIp["100.115.194.84"].sort(), ["acc-a", "acc-b", "acc-c", "acc-d"]); +}); + +test("analyzeEgressSharing: distinct IPs per account = no warning (the healthy .17 case)", () => { + const result = analyzeEgressSharing([ + { connectionId: "a", provider: "codex", account: "acc-a", proxyLevel: "account", proxyHost: "p1", egressIp: "203.0.113.1" }, + { connectionId: "b", provider: "codex", account: "acc-b", proxyLevel: "account", proxyHost: "p2", egressIp: "203.0.113.2" }, + ]); + assert.equal(result.sharedWithinRotationGroup.length, 0, "1 IP per account is safe"); +}); + +test("diagnoseAllEgressIps wires resolution + probe and surfaces the shared-IP warning", async () => { + clearEgressCache(); + _setEgressProbeForTests(async () => ({ ip: "100.115.194.84", latencyMs: 3 })); + + const diag = await diagnoseAllEgressIps({ + getConnections: async () => [ + { id: "c1", provider: "codex", email: "one@x.com" }, + { id: "c2", provider: "codex", email: "two@x.com" }, + ], + resolveProxy: async () => ({ proxy: { type: "http", host: "9.9.9.9", port: 8080 }, level: "global" }), + }); + + assert.equal(diag.connections.length, 2); + assert.equal(diag.connections[0].egressIp, "100.115.194.84"); + assert.equal(diag.connections[0].proxyLevel, "global"); + assert.equal(diag.sharedWithinRotationGroup.length, 1, "both codex accounts share one egress IP"); + assert.equal(diag.sharedWithinRotationGroup[0].connections.length, 2); + + _setEgressProbeForTests(null); +}); + +test("validateProxyPool marks live proxies active and dead proxies error", async () => { + clearEgressCache(); + // proxy p-live reaches the internet; p-dead times out + _setEgressProbeForTests(async (proxyUrl: string | null) => { + if (proxyUrl && proxyUrl.includes("9.9.9.9")) return { ip: "203.0.113.9", latencyMs: 12 }; + return { ip: null, latencyMs: 7000, error: "timeout" }; + }); + + const marked: Record = {}; + const report = await validateProxyPool({ + listProxies: async () => [ + { id: "p-live", type: "http", host: "9.9.9.9", port: 8080, status: "error" }, + { id: "p-dead", type: "http", host: "1.1.1.1", port: 8080, status: "active" }, + ], + markStatus: async (id: string, status: string) => { + marked[id] = status; + }, + }); + + assert.equal(marked["p-live"], "active", "a reachable proxy must be (re)marked active"); + assert.equal(marked["p-dead"], "error", "an unreachable proxy must be marked error (so resolution skips it)"); + + const live = report.find((r) => r.proxyId === "p-live"); + assert.equal(live!.alive, true); + assert.equal(live!.egressIp, "203.0.113.9"); + assert.equal(live!.previousStatus, "error"); + const dead = report.find((r) => r.proxyId === "p-dead"); + assert.equal(dead!.alive, false); + assert.equal(dead!.previousStatus, "active", "was wrongly active before validation"); + + _setEgressProbeForTests(null); +});