fix(sse): bound forwarded response headers (#8041)

* fix(sse): bound forwarded response headers

* docs(changelog): record forwarded response header fix

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

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<string, unknown>) => 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<typeof buildOmniRouteResponseMetaHeaders>[0]
+  meta: Parameters<typeof buildOmniRouteResponseMetaHeaders>[0],
+  log: ResponseHeaderLogger = defaultLogger
 ): Record<string, string> {
-  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<string, string> = {
     ...Object.fromEntries(forwardedHeaders),
     "Content-Type": "text/event-stream",

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<string, CompressionPipelineStep[]>;
 
+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<string, number>();
   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}`;
 }
 
 /**

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);

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<string, string>, 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();