From 987b6448f723ca004e70ca4ecd1ef2ab95c3dbb6 Mon Sep 17 00:00:00 2001 From: Isiah Wheeler Date: Sat, 18 Jul 2026 20:19:50 -0400 Subject: [PATCH] feat(resilience): guard OmniRoute peer routing loops (#7555) * feat(resilience): guard OmniRoute peer routing loops * refactor(resilience): fold peer-loop log+response into rejectPeerRequest helper (file-size budget on chat.ts) Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: Isiah Wheeler <2122839+isiahw1@users.noreply.github.com> Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --- .env.example | 8 ++ docs/guides/USER_GUIDE.md | 30 +++++ docs/reference/ENVIRONMENT.md | 3 + open-sse/executors/base.ts | 4 + src/shared/resilience/peerRouting.ts | 133 +++++++++++++++++++++ src/sse/handlers/chat.ts | 4 + tests/unit/peer-routing.test.ts | 170 +++++++++++++++++++++++++++ 7 files changed, 352 insertions(+) create mode 100644 src/shared/resilience/peerRouting.ts create mode 100644 tests/unit/peer-routing.test.ts diff --git a/.env.example b/.env.example index fdb38b67f7..f45d46af25 100644 --- a/.env.example +++ b/.env.example @@ -88,6 +88,14 @@ PORT=20128 # Used by: src/sse/utils/backpressure.ts — disabled when unset/0. # OMNI_MAX_CONCURRENT_CONNECTIONS=0 +# Optional OmniRoute-to-OmniRoute peer chaining guard. Give every instance a +# unique ID and allowlist only the other OmniRoute base URLs it may call. +# Requests to allowlisted peers carry X-OmniRoute-Peer-Trace; repeated instances +# and exhausted hop budgets are rejected with HTTP 508 before provider routing. +# OMNIROUTE_INSTANCE_ID=gateway-a +# OMNIROUTE_PEER_URLS=http://gateway-b:20128/v1 +# OMNIROUTE_PEER_MAX_HOPS=4 + # Port for the real-time WebSocket live monitoring server. # Used by: src/server/ws/liveServer.ts, src/app/api/v1/ws/route.ts # Default: 20132 diff --git a/docs/guides/USER_GUIDE.md b/docs/guides/USER_GUIDE.md index b35a8094a5..80505b7c76 100644 --- a/docs/guides/USER_GUIDE.md +++ b/docs/guides/USER_GUIDE.md @@ -641,6 +641,36 @@ Notes: - OpenRouter and OpenAI/Anthropic-compatible providers are managed from **Available Models** only. Manual add, import, and auto-sync all land in the same available-model list, so there is no separate Custom Models section for those providers. - The **Custom Models** section is intended for providers that do not expose managed available-model imports. +### Chaining OmniRoute Peers + +Another OmniRoute gateway can be added as a **Custom OpenAI-compatible** provider. Use the +peer's `/v1` base URL and a dedicated, least-privilege API key issued by that peer. + +For reciprocal or multi-hop chains, enable the opt-in loop guard on every gateway: + +```bash +# gateway-a +OMNIROUTE_INSTANCE_ID=gateway-a +OMNIROUTE_PEER_URLS=http://gateway-b:20128/v1 +OMNIROUTE_PEER_MAX_HOPS=4 +``` + +```bash +# gateway-b +OMNIROUTE_INSTANCE_ID=gateway-b +OMNIROUTE_PEER_URLS=http://gateway-a:20128/v1 +OMNIROUTE_PEER_MAX_HOPS=4 +``` + +Only requests sent to an explicitly allowlisted peer URL receive the +`X-OmniRoute-Peer-Trace` header. A gateway rejects a repeated instance ID or exhausted hop +budget with HTTP `508 Loop Detected`; ordinary upstream providers receive no peer metadata. + +Peer chaining is not database replication or host failover. Each gateway keeps independent +SQLite state, caches, rate counters, and sessions. Use a health-checked reverse proxy or client +failover for active/passive or active/active availability, and never mount one SQLite database +into multiple running OmniRoute instances. + ### Dedicated Provider Routes Route requests directly to a specific provider with model validation: diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 6527b4b303..d978ac4a80 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -123,6 +123,9 @@ OmniRoute uses **SQLite** (via `better-sqlite3`) for all persistence. These vari | `API_HOST` | `0.0.0.0` | `src/lib/runtime/ports.ts` | Bind address for the API port. | | `DASHBOARD_PORT` | _(unset)_ | `src/lib/runtime/ports.ts` | When set, serves the Dashboard UI on this separate port. | | `OMNI_MAX_CONCURRENT_CONNECTIONS` | `0` _(disabled)_ | `src/sse/utils/backpressure.ts` | Caps concurrent in-flight chat connections; requests over the cap get `503` with `Retry-After`. Positive integer enables the guard; unset/`0` disables it. | +| `OMNIROUTE_INSTANCE_ID` | _(unset)_ | `src/shared/resilience/peerRouting.ts` | Stable, unique ID for this gateway when chaining OmniRoute instances. Enables inbound peer-loop checks. Allowed characters: letters, digits, `.`, `_`, `:`, and `-`; maximum 64 characters. | +| `OMNIROUTE_PEER_URLS` | _(unset)_ | `src/shared/resilience/peerRouting.ts`, `open-sse/executors/base.ts` | Comma-separated OmniRoute base URLs that may receive `X-OmniRoute-Peer-Trace`. Only explicitly allowlisted upstream URLs receive peer metadata; all other providers are untouched. | +| `OMNIROUTE_PEER_MAX_HOPS` | `4` | `src/shared/resilience/peerRouting.ts` | Maximum number of previously visited OmniRoute instances accepted on a chained request (`1`-`32`). Repeated instances or an exhausted budget return HTTP `508 Loop Detected`. | | `PROD_DASHBOARD_PORT` | `20130` | `docker-compose.prod.yml` | Host-side published port for the Dashboard in Docker production mode. | | `PROD_API_PORT` | `20131` | `docker-compose.prod.yml` | Host-side published port for the API in Docker production mode. | | `OMNIROUTE_PORT` | _(unset)_ | `src/lib/runtime/ports.ts` | Takes precedence over `PORT` when running inside Electron or other wrappers. | diff --git a/open-sse/executors/base.ts b/open-sse/executors/base.ts index 6b4b5d841c..04d40336ad 100644 --- a/open-sse/executors/base.ts +++ b/open-sse/executors/base.ts @@ -87,6 +87,7 @@ import { applyConfiguredUserAgent, stripStainlessHeadersForOpenAICompat, } from "./base/headers.ts"; +import { applyPeerTraceHeader } from "@/shared/resilience/peerRouting"; // Header helpers extracted to a pure leaf; re-exported for external importers // (executors + tests) that import them from "./base.ts". export { @@ -1184,6 +1185,9 @@ export class BaseExecutor { } mergeUpstreamExtraHeaders(finalHeaders, upstreamExtraHeaders); + // Enforce peer tracing after all configurable headers have been merged so + // operator/provider metadata cannot accidentally erase the loop guard. + applyPeerTraceHeader(finalHeaders, clientHeaders, url); const serializedBody = prl.parseBody(bodyString); // #4307 — Preserve the non-enumerable tool-name cloak/remap reverse map // (`_toolNameMap`, set on the live `transformedBody` by diff --git a/src/shared/resilience/peerRouting.ts b/src/shared/resilience/peerRouting.ts new file mode 100644 index 0000000000..c6af678ba4 --- /dev/null +++ b/src/shared/resilience/peerRouting.ts @@ -0,0 +1,133 @@ +const DEFAULT_MAX_PEER_HOPS = 4; +const MAX_TRACE_HEADER_LENGTH = 2048; +const INSTANCE_ID_PATTERN = /^[A-Za-z0-9._:-]{1,64}$/; + +export const OMNIROUTE_PEER_TRACE_HEADER = "X-OmniRoute-Peer-Trace"; + +type HeaderSource = Headers | Record | null | undefined; +type PeerEnvironment = { + OMNIROUTE_INSTANCE_ID?: string; + OMNIROUTE_PEER_URLS?: string; + OMNIROUTE_PEER_MAX_HOPS?: string; +}; + +export type PeerRequestRejection = { + code: "peer_loop_detected" | "peer_hop_limit_exceeded"; + message: string; +}; + +function readHeader(headers: HeaderSource, name: string): string | null { + if (!headers) return null; + if (headers instanceof Headers) return headers.get(name); + + const expected = name.toLowerCase(); + for (const [key, value] of Object.entries(headers)) { + if (key.toLowerCase() === expected && typeof value === "string") return value; + } + return null; +} + +function getInstanceId(env: PeerEnvironment): string | null { + const value = env.OMNIROUTE_INSTANCE_ID?.trim() ?? ""; + return INSTANCE_ID_PATTERN.test(value) ? value : null; +} + +function getMaxPeerHops(env: PeerEnvironment): number { + const parsed = Number.parseInt(env.OMNIROUTE_PEER_MAX_HOPS ?? "", 10); + return Number.isInteger(parsed) && parsed > 0 && parsed <= 32 ? parsed : DEFAULT_MAX_PEER_HOPS; +} + +export function parsePeerTrace(value: string | null | undefined): string[] { + if (!value || value.length > MAX_TRACE_HEADER_LENGTH) return []; + return value + .split(",") + .map((part) => part.trim()) + .filter((part) => INSTANCE_ID_PATTERN.test(part)); +} + +function normalizePeerUrl(value: string): URL | null { + try { + const url = new URL(value.trim()); + if (url.protocol !== "http:" && url.protocol !== "https:") return null; + return url; + } catch { + return null; + } +} + +export function isConfiguredOmniRoutePeer( + targetUrl: string, + env: PeerEnvironment = process.env +): boolean { + const target = normalizePeerUrl(targetUrl); + if (!target) return false; + + return (env.OMNIROUTE_PEER_URLS ?? "") + .split(",") + .map(normalizePeerUrl) + .some((peer) => { + if (!peer || peer.origin !== target.origin) return false; + const peerPath = peer.pathname.replace(/\/$/, ""); + const targetPath = target.pathname.replace(/\/$/, ""); + return targetPath === peerPath || targetPath.startsWith(`${peerPath}/`); + }); +} + +/** Reject a request that has already visited this instance or exhausted its peer-hop budget. */ +export function inspectPeerRequest( + headers: HeaderSource, + env: PeerEnvironment = process.env +): PeerRequestRejection | null { + const instanceId = getInstanceId(env); + if (!instanceId) return null; + + const trace = parsePeerTrace(readHeader(headers, OMNIROUTE_PEER_TRACE_HEADER)); + if (trace.includes(instanceId)) { + return { + code: "peer_loop_detected", + message: "OmniRoute peer routing loop detected", + }; + } + if (trace.length >= getMaxPeerHops(env)) { + return { + code: "peer_hop_limit_exceeded", + message: "OmniRoute peer routing hop limit exceeded", + }; + } + return null; +} + +/** Convenience for HTTP handlers: inspect + log + build the 508 response in one call. */ +export function rejectPeerRequest( + headers: HeaderSource, + warn: (tag: string, msg: string) => void, + respond: (status: number, msg: string) => T +): T | null { + const rejection = inspectPeerRequest(headers); + if (!rejection) return null; + warn("PEER_ROUTING", rejection.message); + return respond(508, rejection.message); +} + +/** + * Append this instance to the peer trace for an explicitly allowlisted OmniRoute URL. + * Returns true when the header was applied. Other upstream providers are untouched. + */ +export function applyPeerTraceHeader( + outgoingHeaders: Record, + clientHeaders: HeaderSource, + targetUrl: string, + env: PeerEnvironment = process.env +): boolean { + const instanceId = getInstanceId(env); + if (!instanceId || !isConfiguredOmniRoutePeer(targetUrl, env)) return false; + + const trace = parsePeerTrace(readHeader(clientHeaders, OMNIROUTE_PEER_TRACE_HEADER)); + if (!trace.includes(instanceId)) trace.push(instanceId); + const traceHeaderLower = OMNIROUTE_PEER_TRACE_HEADER.toLowerCase(); + for (const key of Object.keys(outgoingHeaders)) { + if (key.toLowerCase() === traceHeaderLower) delete outgoingHeaders[key]; + } + outgoingHeaders[OMNIROUTE_PEER_TRACE_HEADER] = trace.join(","); + return true; +} diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 83487511c2..e7975dabc8 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -50,6 +50,7 @@ import { import * as log from "../utils/logger"; import { checkAndRefreshToken } from "../services/tokenRefresh"; import { createHookContext, runHooks, initPreRequestRegistry } from "@/lib/middleware/registry"; +import { rejectPeerRequest } from "@/shared/resilience/peerRouting"; import { deleteHandoff, getHandoff } from "@/lib/db/contextHandoffs"; import { updateCombo } from "@/lib/db/combos"; import { isModelAllowedForKey } from "@/lib/db/apiKeys"; @@ -221,6 +222,9 @@ export async function handleChat( preParsedBody: any = null, correlationId?: string ) { + const peerRejection = rejectPeerRequest(request?.headers, log.warn, errorResponse); + if (peerRejection) return peerRejection; + // Pipeline: Start request telemetry const reqId = correlationId || generateRequestId(); const telemetry = new RequestTelemetry(reqId); diff --git a/tests/unit/peer-routing.test.ts b/tests/unit/peer-routing.test.ts new file mode 100644 index 0000000000..b448b64896 --- /dev/null +++ b/tests/unit/peer-routing.test.ts @@ -0,0 +1,170 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { + applyPeerTraceHeader, + inspectPeerRequest, + isConfiguredOmniRoutePeer, + parsePeerTrace, +} from "../../src/shared/resilience/peerRouting"; + +const env = { + OMNIROUTE_INSTANCE_ID: "gateway-a", + OMNIROUTE_PEER_URLS: "http://gateway-b:20128/v1,https://peer.example/api/v1/", + OMNIROUTE_PEER_MAX_HOPS: "4", +}; + +test("peer routing stays disabled without an instance id", () => { + const headers: Record = {}; + assert.equal( + applyPeerTraceHeader(headers, null, "http://gateway-b:20128/v1/chat/completions", { + ...env, + OMNIROUTE_INSTANCE_ID: undefined, + }), + false + ); + assert.deepEqual(headers, {}); + assert.equal( + inspectPeerRequest(new Headers({ "X-OmniRoute-Peer-Trace": "gateway-a" }), {}), + null + ); +}); + +test("peer URL matching requires the configured origin and path boundary", () => { + assert.equal(isConfiguredOmniRoutePeer("http://gateway-b:20128/v1/chat/completions", env), true); + assert.equal(isConfiguredOmniRoutePeer("https://peer.example/api/v1/responses", env), true); + assert.equal( + isConfiguredOmniRoutePeer("http://gateway-b:20128/v10/chat/completions", env), + false + ); + assert.equal( + isConfiguredOmniRoutePeer("http://gateway-b.evil:20128/v1/chat/completions", env), + false + ); +}); + +test("outbound peer calls append the local instance to the existing trace", () => { + const headers: Record = { Authorization: "Bearer test" }; + const applied = applyPeerTraceHeader( + headers, + { "x-omniroute-peer-trace": "edge,gateway-z" }, + "http://gateway-b:20128/v1/chat/completions", + env + ); + assert.equal(applied, true); + assert.equal(headers["X-OmniRoute-Peer-Trace"], "edge,gateway-z,gateway-a"); + assert.equal(headers.Authorization, "Bearer test"); +}); + +test("non-peer providers never receive peer metadata", () => { + const headers: Record = {}; + assert.equal( + applyPeerTraceHeader(headers, null, "https://api.openai.com/v1/chat/completions", env), + false + ); + assert.equal(headers["X-OmniRoute-Peer-Trace"], undefined); +}); + +test("ingress rejects a repeated instance and an exhausted hop budget", () => { + assert.deepEqual( + inspectPeerRequest(new Headers({ "X-OmniRoute-Peer-Trace": "edge,gateway-a" }), env), + { + code: "peer_loop_detected", + message: "OmniRoute peer routing loop detected", + } + ); + assert.deepEqual( + inspectPeerRequest( + { "X-OmniRoute-Peer-Trace": "gateway-w,gateway-x,gateway-y,gateway-z" }, + env + ), + { + code: "peer_hop_limit_exceeded", + message: "OmniRoute peer routing hop limit exceeded", + } + ); +}); + +test("trace parsing drops invalid IDs and oversized untrusted values", () => { + assert.deepEqual(parsePeerTrace("gateway-a, bad id, gateway_b"), ["gateway-a", "gateway_b"]); + assert.deepEqual(parsePeerTrace("x".repeat(2049)), []); +}); + +test("BaseExecutor adds the trace only on an allowlisted peer dispatch", async () => { + const previous = { + instanceId: process.env.OMNIROUTE_INSTANCE_ID, + peerUrls: process.env.OMNIROUTE_PEER_URLS, + maxHops: process.env.OMNIROUTE_PEER_MAX_HOPS, + }; + let capturedTrace: string | undefined; + const server = createServer((request, response) => { + capturedTrace = request.headers["x-omniroute-peer-trace"]; + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ choices: [] })); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + assert.ok(address && typeof address === "object"); + const peerBaseUrl = `http://127.0.0.1:${address.port}/v1`; + + process.env.OMNIROUTE_INSTANCE_ID = "gateway-a"; + process.env.OMNIROUTE_PEER_URLS = peerBaseUrl; + process.env.OMNIROUTE_PEER_MAX_HOPS = "4"; + + try { + const { BaseExecutor } = await import("../../open-sse/executors/base.ts"); + const executor = new BaseExecutor("openai-compatible-peer", { + baseUrl: peerBaseUrl, + }); + await executor.execute({ + model: "peer-model", + body: { model: "peer-model", messages: [{ role: "user", content: "ping" }] }, + stream: false, + credentials: { + apiKey: "peer-key", + providerSpecificData: { baseUrl: peerBaseUrl }, + }, + clientHeaders: { "x-omniroute-peer-trace": "edge" }, + upstreamExtraHeaders: { "x-omniroute-peer-trace": "overridden" }, + }); + + assert.equal(capturedTrace, "edge,gateway-a"); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + if (previous.instanceId === undefined) delete process.env.OMNIROUTE_INSTANCE_ID; + else process.env.OMNIROUTE_INSTANCE_ID = previous.instanceId; + if (previous.peerUrls === undefined) delete process.env.OMNIROUTE_PEER_URLS; + else process.env.OMNIROUTE_PEER_URLS = previous.peerUrls; + if (previous.maxHops === undefined) delete process.env.OMNIROUTE_PEER_MAX_HOPS; + else process.env.OMNIROUTE_PEER_MAX_HOPS = previous.maxHops; + } +}); + +test("handleChat rejects a reciprocal peer loop before provider routing", async () => { + const previous = process.env.OMNIROUTE_INSTANCE_ID; + process.env.OMNIROUTE_INSTANCE_ID = "gateway-a"; + try { + const { handleChat } = await import("../../src/sse/handlers/chat.ts"); + const response = await handleChat( + new Request("http://gateway-a:20128/v1/chat/completions", { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-OmniRoute-Peer-Trace": "gateway-b,gateway-a", + }, + body: JSON.stringify({ + model: "steady-free", + messages: [{ role: "user", content: "ping" }], + }), + }) + ); + + assert.equal(response.status, 508); + assert.match(await response.text(), /peer routing loop detected/i); + } finally { + if (previous === undefined) delete process.env.OMNIROUTE_INSTANCE_ID; + else process.env.OMNIROUTE_INSTANCE_ID = previous; + } +});