diff --git a/changelog.d/fixes/8041-forwarded-response-header-budget.md b/changelog.d/fixes/8041-forwarded-response-header-budget.md new file mode 100644 index 0000000000..39abd21147 --- /dev/null +++ b/changelog.d/fixes/8041-forwarded-response-header-budget.md @@ -0,0 +1 @@ +- **fix(sse):** Bound forwarded upstream response headers so large provider metadata cannot turn successful streaming requests into reverse-proxy errors ([#8041](https://github.com/diegosouzapw/OmniRoute/pull/8041)) — thanks @insoln diff --git a/open-sse/handlers/chatCore/responseHeaders.ts b/open-sse/handlers/chatCore/responseHeaders.ts index 3c09538716..60701d9495 100644 --- a/open-sse/handlers/chatCore/responseHeaders.ts +++ b/open-sse/handlers/chatCore/responseHeaders.ts @@ -3,14 +3,70 @@ import { buildOmniRouteResponseMetaHeaders, } from "@/domain/omnirouteResponseMeta"; import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers"; +import { defaultLogger } from "@omniroute/open-sse/utils/logger"; const STREAMING_RESPONSE_HEADER_DENYLIST = new Set([ "content-type", "content-encoding", "content-length", "transfer-encoding", + "cache-control", + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "upgrade", + "authorization", + "authentication-info", + "cookie", + "set-cookie", + "set-cookie2", + "www-authenticate", + "x-api-key", + "x-amz-security-token", + "x-auth-token", + "x-accel-buffering", ]); +/** + * Keep upstream-derived headers comfortably below common reverse-proxy response-header limits. + * This budget includes each header name, separator, value, and trailing CRLF. OmniRoute's own + * response metadata and framework/security headers are added separately. + */ +export const MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES = 768; +const MAX_LOGGED_DROPPED_RESPONSE_HEADERS = 20; +const responseHeaderEncoder = new TextEncoder(); + +type ResponseHeaderLogger = { + warn?: (tag: string, message: string, data?: Record) => void; +} | null; + +function responseHeaderWireBytes(name: string, value: string): number { + return responseHeaderEncoder.encode(`${name}: ${value}\r\n`).byteLength; +} + +function isOmniRouteInternalHeader(headerName: string): boolean { + return headerName.toLowerCase().startsWith("x-omniroute-"); +} + +function getForwardingPriority(headerName: string): number { + const normalized = headerName.toLowerCase(); + if ( + normalized === "x-request-id" || + normalized === "request-id" || + normalized === "x-correlation-id" || + normalized === "traceparent" || + normalized === "traceresponse" + ) { + return 0; + } + if (normalized === "retry-after") return 1; + if (normalized.includes("ratelimit") || normalized.includes("rate-limit")) return 2; + return 3; +} + /** * Prefix of Next.js internal middleware control headers. * @@ -20,8 +76,8 @@ const STREAMING_RESPONSE_HEADER_DENYLIST = new Set([ * `x-middleware-next`, `x-middleware-override-headers`, * `x-middleware-set-cookie`, and the `x-middleware-request-*` family. * - * OmniRoute forwards upstream response headers verbatim. If we re-emit those - * headers from an App Router route handler, Next 16's `app-route` runtime + * If OmniRoute re-emits those headers from an App Router route handler, Next + * 16's `app-route` runtime * interprets `x-middleware-rewrite` as a `NextResponse.rewrite()` call and * throws `NextResponse.rewrite() was used in a app route handler` — turning a * successful upstream call into a 500. This is provider-agnostic proxy @@ -58,18 +114,67 @@ export function stripNextMiddlewareControlHeaders(headers: Headers): void { export function buildStreamingResponseHeaders( providerHeaders: Headers, - meta: Parameters[0] + meta: Parameters[0], + log: ResponseHeaderLogger = defaultLogger ): Record { - const forwardedHeaders: [string, string][] = []; + const connectionScopedHeaders = new Set( + (providerHeaders.get("connection") || "") + .split(",") + .map((name) => name.trim().toLowerCase()) + .filter(Boolean) + ); + const candidates: Array<{ + key: string; + value: string; + bytes: number; + priority: number; + position: number; + }> = []; + let position = 0; + providerHeaders.forEach((value, key) => { + const normalized = key.toLowerCase(); if ( - !STREAMING_RESPONSE_HEADER_DENYLIST.has(key.toLowerCase()) && - !isNextMiddlewareControlHeader(key) + STREAMING_RESPONSE_HEADER_DENYLIST.has(normalized) || + connectionScopedHeaders.has(normalized) || + isNextMiddlewareControlHeader(normalized) || + isOmniRouteInternalHeader(normalized) ) { - forwardedHeaders.push([key, value]); + return; } + candidates.push({ + key, + value, + bytes: responseHeaderWireBytes(key, value), + priority: getForwardingPriority(key), + position: position++, + }); }); + candidates.sort((a, b) => a.priority - b.priority || a.position - b.position); + + const forwardedHeaders: [string, string][] = []; + const droppedHeaders: Array<{ name: string; bytes: number }> = []; + let forwardedBytes = 0; + + for (const candidate of candidates) { + if (forwardedBytes + candidate.bytes <= MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES) { + forwardedHeaders.push([candidate.key, candidate.value]); + forwardedBytes += candidate.bytes; + } else { + droppedHeaders.push({ name: candidate.key, bytes: candidate.bytes }); + } + } + + if (droppedHeaders.length > 0) { + log?.warn?.("HTTP", "Dropped upstream response headers that exceeded forwarding budget", { + budgetBytes: MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES, + forwardedBytes, + droppedCount: droppedHeaders.length, + droppedHeaders: droppedHeaders.slice(0, MAX_LOGGED_DROPPED_RESPONSE_HEADERS), + }); + } + const responseHeaders: Record = { ...Object.fromEntries(forwardedHeaders), "Content-Type": "text/event-stream", diff --git a/open-sse/services/compression/planResolution.ts b/open-sse/services/compression/planResolution.ts index b8687d23a3..ab68a89d89 100644 --- a/open-sse/services/compression/planResolution.ts +++ b/open-sse/services/compression/planResolution.ts @@ -9,6 +9,13 @@ import { /** Named-combo map: combo id → its stacked pipeline (operator-defined profiles). */ type NamedCombos = Record; +const MAX_COMPRESSION_ANNOTATION_BYTES = 768; +const NON_ASCII_HEADER_VALUE_CHARS = /[^\x20-\x7e]/g; + +function utf8ByteLength(value: string): number { + return new TextEncoder().encode(value).byteLength; +} + /** Tags a plan with the precedence layer that produced it (Phase 3 observability). */ export function withSource(plan: DerivedPlan, source: CompressionSource): DerivedPlan { return { ...plan, source }; @@ -69,12 +76,28 @@ export function formatCompressionAnnotation(stats: CompressionStats): string { const counts = new Map(); for (const rule of rules) { - counts.set(rule, (counts.get(rule) ?? 0) + 1); + const safeRule = rule.replace(NON_ASCII_HEADER_VALUE_CHARS, "?"); + counts.set(safeRule, (counts.get(safeRule) ?? 0) + 1); } const sorted = [...counts.entries()].sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0])); - const agg = sorted.map(([name, n]) => `${name}x${n}`).join(", "); - return `tokens=${stats.originalTokens}->${stats.compressedTokens}; rules: ${agg}`; + const prefix = `tokens=${stats.originalTokens}->${stats.compressedTokens}; rules: `; + const suffix = ", ..."; + const parts: string[] = []; + let bytes = utf8ByteLength(prefix); + for (const [name, n] of sorted) { + const part = `${name}x${n}`; + const separator = parts.length > 0 ? ", " : ""; + const partBytes = utf8ByteLength(separator + part); + if (bytes + partBytes > MAX_COMPRESSION_ANNOTATION_BYTES - utf8ByteLength(suffix)) { + if (parts.length === 0) return ""; + return `${prefix}${parts.join(", ")}${suffix}`; + } + parts.push(part); + bytes += partBytes; + } + const agg = parts.join(", "); + return `${prefix}${agg}`; } /** diff --git a/tests/unit/compression/compressionAnnotation.test.ts b/tests/unit/compression/compressionAnnotation.test.ts index 98adbc02af..fa5338de67 100644 --- a/tests/unit/compression/compressionAnnotation.test.ts +++ b/tests/unit/compression/compressionAnnotation.test.ts @@ -31,7 +31,18 @@ describe("formatCompressionAnnotation", () => { it("aggregates rulesApplied counts deterministically", () => { const stats = makeStats({ - rulesApplied: ["filler", "filler", "filler", "filler", "filler", "filler", "filler", "filler", "dedup", "dedup"], + rulesApplied: [ + "filler", + "filler", + "filler", + "filler", + "filler", + "filler", + "filler", + "filler", + "dedup", + "dedup", + ], techniquesUsed: ["caveman"], }); const result = formatCompressionAnnotation(stats); @@ -50,7 +61,10 @@ describe("formatCompressionAnnotation", () => { }); const value = `standard; source=auto; ${formatCompressionAnnotation(stats)}`; for (const ch of value) { - assert.ok(ch.codePointAt(0)! <= 0xff, `non-latin1 char ${JSON.stringify(ch)} in header value: ${value}`); + assert.ok( + ch.codePointAt(0)! <= 0xff, + `non-latin1 char ${JSON.stringify(ch)} in header value: ${value}` + ); } // Must not throw at real Headers/Response construction. assert.doesNotThrow(() => new Headers({ "X-OmniRoute-Compression": value })); @@ -66,7 +80,10 @@ describe("formatCompressionAnnotation", () => { const result = formatCompressionAnnotation(stats); const fillerIdx = result.indexOf("fillerx3"); const dedupIdx = result.indexOf("dedupx1"); - assert.ok(fillerIdx < dedupIdx, `filler (count=3) should appear before dedup (count=1): ${result}`); + assert.ok( + fillerIdx < dedupIdx, + `filler (count=3) should appear before dedup (count=1): ${result}` + ); }); it("is deterministic (same input → same output)", () => { @@ -77,6 +94,41 @@ describe("formatCompressionAnnotation", () => { assert.equal(formatCompressionAnnotation(stats), formatCompressionAnnotation(stats)); }); + it("bounds high-cardinality rule telemetry before it reaches an HTTP response header", () => { + const stats = makeStats({ + rulesApplied: Array.from( + { length: 1_000 }, + (_, index) => `rtk:custom-filter:${index.toString().padStart(4, "0")}` + ), + techniquesUsed: ["rtk-filter"], + }); + + const annotation = formatCompressionAnnotation(stats); + assert.ok( + Buffer.byteLength(annotation) <= 768, + `annotation is ${Buffer.byteLength(annotation)} bytes` + ); + assert.ok(annotation.endsWith(", ..."), `expected truncation marker in: ${annotation}`); + assert.doesNotThrow( + () => new Response(null, { headers: { "X-OmniRoute-Compression": annotation } }) + ); + }); + + it("replaces non-ASCII and control characters before constructing the header", () => { + const annotation = formatCompressionAnnotation( + makeStats({ rulesApplied: ["rtk:one\r\nx-injected: yes", "rtk:two\u0000", "rtk:café"] }) + ); + + assert.doesNotMatch(annotation, /[^\x20-\x7e]/); + assert.ok( + annotation.includes("rtk:caf?x1"), + `expected a sanitized rule name in: ${annotation}` + ); + assert.doesNotThrow( + () => new Response(null, { headers: { "X-OmniRoute-Compression": annotation } }) + ); + }); + it("prefix mode; source=X is never mutated by appending the annotation", () => { const plan = { mode: "standard" as const, stackedPipeline: [], source: "auto" as const }; const prefix = formatCompressionMeta(plan); diff --git a/tests/unit/middleware-header-strip-5849.test.ts b/tests/unit/middleware-header-strip-5849.test.ts index 3e087c77bb..7fbe9b18e3 100644 --- a/tests/unit/middleware-header-strip-5849.test.ts +++ b/tests/unit/middleware-header-strip-5849.test.ts @@ -4,6 +4,7 @@ import { test } from "node:test"; import { buildStreamingResponseHeaders, isNextMiddlewareControlHeader, + MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES, stripNextMiddlewareControlHeaders, } from "@omniroute/open-sse/handlers/chatCore/responseHeaders.ts"; @@ -22,6 +23,11 @@ const MIDDLEWARE_HEADERS: [string, string][] = [ ["x-middleware-request-foo", "bar"], ]; +function getHeaderValue(headers: Record, name: string): string | undefined { + const entry = Object.entries(headers).find(([key]) => key.toLowerCase() === name.toLowerCase()); + return entry?.[1]; +} + test("isNextMiddlewareControlHeader matches the whole x-middleware-* family (case-insensitive)", () => { for (const [name] of MIDDLEWARE_HEADERS) { assert.equal(isNextMiddlewareControlHeader(name), true, name); @@ -50,6 +56,84 @@ test("streaming path: buildStreamingResponseHeaders strips x-middleware-* and pr assert.ok(requestIdKey, "x-request-id must be preserved"); assert.equal(out[requestIdKey as string], "req-123"); }); +test("streaming path strips oversized and credential-bearing upstream response headers", () => { + const upstream = new Headers({ + "x-request-id": "req-oversized-guard", + "x-upstream-diagnostic": "x".repeat(MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES), + "set-cookie": "session=upstream-secret; HttpOnly", + }); + + const warnings: unknown[][] = []; + const out = buildStreamingResponseHeaders( + upstream, + {}, + { + warn: (...args: unknown[]) => warnings.push(args), + } + ); + const lowerKeys = Object.keys(out).map((key) => key.toLowerCase()); + assert.ok(lowerKeys.includes("x-request-id")); + assert.ok(!lowerKeys.includes("x-upstream-diagnostic")); + assert.ok(!lowerKeys.includes("set-cookie")); + assert.equal(warnings.length, 1); + assert.ok(!JSON.stringify(warnings).includes("session=upstream-secret")); +}); + +test("streaming path bounds the aggregate size of many small upstream response headers", () => { + const upstream = new Headers({ "x-request-id": "req-many-small-headers" }); + for (let index = 0; index < 40; index += 1) { + upstream.set(`x-upstream-diagnostic-${index.toString().padStart(2, "0")}`, "x".repeat(32)); + } + + const out = buildStreamingResponseHeaders(upstream, {}, null); + const upstreamEntries = Object.entries(out).filter( + ([name]) => + name.toLowerCase().startsWith("x-upstream-") || name.toLowerCase() === "x-request-id" + ); + const forwardedBytes = upstreamEntries.reduce( + (total, [name, value]) => total + Buffer.byteLength(`${name}: ${value}\r\n`), + 0 + ); + + assert.ok(forwardedBytes <= MAX_FORWARDED_UPSTREAM_RESPONSE_HEADER_BYTES); + assert.ok(upstreamEntries.length < 41, "at least one small header must be dropped"); + assert.equal(getHeaderValue(out, "x-request-id"), "req-many-small-headers"); +}); + +test("streaming path prioritizes request and rate-limit headers over diagnostics", () => { + const upstream = new Headers(); + for (let index = 0; index < 20; index += 1) { + upstream.set(`a-diagnostic-${index.toString().padStart(2, "0")}`, "x".repeat(48)); + } + upstream.set("retry-after", "30"); + upstream.set("x-ratelimit-remaining-requests", "12"); + upstream.set("x-request-id", "req-priority"); + + const out = buildStreamingResponseHeaders(upstream, {}, null); + + assert.equal(getHeaderValue(out, "x-request-id"), "req-priority"); + assert.equal(getHeaderValue(out, "retry-after"), "30"); + assert.equal(getHeaderValue(out, "x-ratelimit-remaining-requests"), "12"); +}); + +test("streaming path strips hop-by-hop and spoofed OmniRoute headers", () => { + const upstream = new Headers({ + connection: "keep-alive, x-remove-me", + "keep-alive": "timeout=5", + "proxy-authenticate": "Basic realm=upstream", + "x-remove-me": "connection-scoped", + "x-omniroute-provider": "spoofed-provider", + "x-request-id": "req-safe", + }); + + const out = buildStreamingResponseHeaders(upstream, {}, null); + + assert.equal(getHeaderValue(out, "x-request-id"), "req-safe"); + assert.ok(!Object.keys(out).some((name) => name.toLowerCase() === "keep-alive")); + assert.ok(!Object.keys(out).some((name) => name.toLowerCase() === "proxy-authenticate")); + assert.ok(!Object.keys(out).some((name) => name.toLowerCase() === "x-remove-me")); + assert.ok(!Object.values(out).includes("spoofed-provider")); +}); test("non-streaming JSON path: stripNextMiddlewareControlHeaders removes the family, keeps the rest", () => { const headers = new Headers();