From 3e3c409436792129259cfae3d95179a94321f9ce Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Fri, 21 Aug 2026 23:16:59 -0700 Subject: [PATCH] fix(security): validate caller-supplied upstreams on every resolution path (#3195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary CVE-2026-77775 (SSRF via `x-headroom-base-url`) is **not fully fixed on current `main`**. The advisory lists 0.36.1 as the last affected version; one route still forwards to any destination a caller names. `upstream_guard.is_safe_upstream_url` was added and wired into `/v1/messages` and the catch-all passthrough. But `select_passthrough_base_url` moved from `providers/proxy_routes.py` to `providers/proxy_targets.py`, and the guard did not follow it. Its Azure branch returns the header verbatim whenever an `api-key` header is present — **both values are caller-supplied** — and `POST /v1/alpha/search` resolves its upstream through that helper without checking the header itself. ## Verified, not inferred Against the current tree, with a listener on loopback standing in for an internal service: ``` proxy status : 200 internal service hit : 1 time(s) Authorization it received : 'Bearer SECRET-CLIENT-TOKEN' internal body relayed back : True ``` The caller's credentials are forwarded to the attacker-named host and the internal response is relayed back. After this change: `400`, zero hits, nothing relayed. A sweep of all 99 routes isolates exactly one leak on unfixed code — `POST /v1/alpha/search` with `api-key` — and zero after. ## 1. The missing enforcement **Guarded at the chokepoint, not just the route.** `select_passthrough_base_url` now validates before returning, in `proxy_targets.py` and in the parallel copy in `providers/registry.py`, so a future caller that forgets the header check cannot reopen this. `/v1/alpha/search` also rejects explicitly with 400, matching its sibling routes. ## 2. A second gap in the address policy RFC 6598 shared address space (`100.64.0.0/10`) is not `is_private`, so it passed the guard — while routing to ISP and cloud-internal infrastructure. `_is_internal_address` now also rejects anything not globally routable. Verified over a 27-vector battery — 0 bypasses, public control unaffected: | Vector | Before | After | |---|---|---| | `100.64.0.0/10` shared address space | **allowed** | blocked | | `198.18/15`, TEST-NET, `240/4` | **allowed** | blocked | | 6to4 / Teredo embedding internal IPv4 | **allowed** | blocked | | NAT64 `64:ff9b::/96` embedding loopback | **allowed** | blocked | | loopback, RFC1918, link-local, metadata, IPv4-mapped, userinfo tricks | blocked | blocked | | multicast `224.0.0.1` | blocked | blocked | | public `8.8.8.8` | allowed | allowed | The category checks are **kept alongside** `is_global` rather than replaced — `is_global` is `True` for multicast, so a replacement would have regressed. NAT64 also reports as global, so its embedded IPv4 is extracted and judged on its own. ## 3. Unauthenticated stall via the resolver `socket.getaddrinfo` takes no timeout and runs on the calling thread — the event loop. Since the hostname is caller-supplied, a deliberately slow-resolving name stalled every other in-flight request; a handful of concurrent requests made the proxy unresponsive, unauthenticated. Resolution now runs in a small dedicated pool with a budget (`HEADROOM_UPSTREAM_RESOLVE_TIMEOUT_S`, default 3s) and fails closed on overrun, which bounds every caller including the synchronous chokepoint. `is_safe_upstream_url_async` runs the lookup off the loop, and the three route handlers that validate a caller-supplied upstream now await it. Caching was deliberately avoided: a TTL cache in front of a security decision invites poisoning, and would widen the rebinding window rather than narrow it. ## Why this survived The existing tests unit-tested the guard's *logic* but never asserted it was *reached*. Added enforcement tests at the sinks plus a **sweep over the whole route table** that fails if any route forwards to a loopback address — so the next unguarded upstream resolution fails in CI rather than in a CVE. All new tests were confirmed failing against the unfixed tree and passing after. ## Known residual — deliberately not addressed **DNS rebinding.** Validation and connection resolve the host separately, so a low-TTL answer can differ between them. Closing this needs connection-time pinning in the shared `http_client` transport, which carries every request in the proxy — too broad to fold into this patch. It should not be described as fixed. ## Compatibility An endpoint that does not resolve publicly (split-horizon, on-prem) is now rejected where it previously passed unvalidated. `HEADROOM_ALLOWED_BASE_URLS` is the documented opt-in, covered by test. Three existing tests used fictional hostnames and legitimately began failing; DNS is pinned in them so they keep testing target precedence rather than depending on the missing guard. Separately: `docker-compose.yml` has already been hardened since the advisory — `HEADROOM_PROXY_TOKEN` is now mandatory and ports are loopback-only — so the "exposed by default" multiplier the advisory cites no longer applies to the shipped compose. Full suite: the 3 failures outside this area (`test_learn/test_integration`, `test_release_workflows::test_no_native_tls_in_wheel_build_tree`, and a `test_graceful_shutdown` ordering flake) reproduce on clean `main` and are unrelated. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Tejas Chopra Co-authored-by: Claude Opus 5 --- headroom/providers/proxy_routes.py | 14 +- headroom/providers/proxy_targets.py | 14 +- headroom/providers/registry.py | 5 +- headroom/proxy/upstream_guard.py | 79 +++++++- tests/test_provider_proxy_routes.py | 20 +- tests/test_provider_proxy_targets.py | 23 ++- tests/test_provider_registry_extended.py | 20 +- tests/test_upstream_guard.py | 241 +++++++++++++++++++++++ 8 files changed, 391 insertions(+), 25 deletions(-) diff --git a/headroom/providers/proxy_routes.py b/headroom/providers/proxy_routes.py index 8800a0809..5cf11ea47 100644 --- a/headroom/providers/proxy_routes.py +++ b/headroom/providers/proxy_routes.py @@ -67,7 +67,7 @@ from headroom.proxy.passthrough import ( custom_base_passthrough_telemetry as _custom_base_passthrough_telemetry, ) from headroom.proxy.request_scope import normalize_request_path -from headroom.proxy.upstream_guard import is_safe_upstream_url +from headroom.proxy.upstream_guard import is_safe_upstream_url_async logger = logging.getLogger("headroom.proxy.routes") @@ -267,7 +267,7 @@ def register_provider_routes(app: FastAPI, proxy: Any) -> None: # OpenAI-compatible and generic passthrough routes. custom_base = request.headers.get("x-headroom-base-url", "").strip() if custom_base: - if not is_safe_upstream_url(custom_base): + if not await is_safe_upstream_url_async(custom_base): logger.warning("rejecting unsafe x-headroom-base-url: %r", custom_base) raise HTTPException(status_code=400, detail="Rejected unsafe upstream base URL") return await proxy.handle_anthropic_messages( @@ -495,6 +495,14 @@ def register_provider_routes(app: FastAPI, proxy: Any) -> None: chatgpt_response = await _handle_chatgpt_codex_alpha_search(request, proxy) if chatgpt_response is not None: return chatgpt_response + # This route resolves a caller-named upstream like the catch-all does, + # so it needs the same rejection. Without it a client could point the + # proxy at loopback/RFC1918/cloud-metadata and read the response back + # (CVE-2026-77775). + custom_base = request.headers.get("x-headroom-base-url", "").strip() + if custom_base and not await is_safe_upstream_url_async(custom_base): + logger.warning("rejecting unsafe x-headroom-base-url: %r", custom_base) + raise HTTPException(status_code=400, detail="Rejected unsafe upstream base URL") return await proxy.handle_passthrough( request, _select_passthrough_base_url(proxy, dict(request.headers)), @@ -510,7 +518,7 @@ def register_provider_routes(app: FastAPI, proxy: Any) -> None: async def passthrough(request: Request, path: str): custom_base = request.headers.get("x-headroom-base-url") if custom_base: - if not is_safe_upstream_url(custom_base): + if not await is_safe_upstream_url_async(custom_base): logger.warning("rejecting unsafe x-headroom-base-url: %r", custom_base) raise HTTPException(status_code=400, detail="Rejected unsafe upstream base URL") base_url = custom_base.rstrip("/") diff --git a/headroom/providers/proxy_targets.py b/headroom/providers/proxy_targets.py index 86b2c5447..0ef4dfff8 100644 --- a/headroom/providers/proxy_targets.py +++ b/headroom/providers/proxy_targets.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from collections.abc import Mapping from typing import Any, cast @@ -13,6 +14,7 @@ from headroom.copilot_auth import ( from headroom.providers.codex import resolve_codex_routing from headroom.providers.codex.endpoints import CHATGPT_BACKEND_API_URL from headroom.providers.vertex import vertex_target_for_location as _vertex_target_for_location +from headroom.proxy.upstream_guard import is_safe_upstream_url LEGACY_API_TARGET_ATTRS: dict[str, str] = { "anthropic": "ANTHROPIC_API_URL", @@ -34,6 +36,9 @@ def vertex_target_for_location(proxy: Any, location: str) -> str: return _vertex_target_for_location(api_target(proxy, "vertex"), location) +logger = logging.getLogger("headroom.proxy") + + def select_passthrough_base_url( proxy: Any, headers: Mapping[str, str], path: str | None = None ) -> str: @@ -46,7 +51,14 @@ def select_passthrough_base_url( if headers.get("api-key"): azure_base = headers.get("x-headroom-base-url", "") if azure_base: - return azure_base.rstrip("/") + # Validate here, not only at the routes. `api-key` is attacker- + # supplied too, so this branch is reachable by anyone who can send + # a header, and it returns the destination the caller named. Routes + # that forgot to guard turned the proxy into an SSRF relay into + # loopback/RFC1918/cloud-metadata space (CVE-2026-77775). + if is_safe_upstream_url(azure_base): + return azure_base.rstrip("/") + logger.warning("ignoring unsafe x-headroom-base-url override: %r", azure_base) provider_name = proxy.provider_runtime.model_metadata_provider(headers) target = api_target(proxy, provider_name) if ( diff --git a/headroom/providers/registry.py b/headroom/providers/registry.py index cd92cbdac..04c212a4d 100644 --- a/headroom/providers/registry.py +++ b/headroom/providers/registry.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any, cast from headroom.providers.claude import DEFAULT_API_URL as DEFAULT_ANTHROPIC_API_URL from headroom.providers.codex import DEFAULT_API_URL as DEFAULT_OPENAI_API_URL from headroom.providers.gemini import DEFAULT_API_URL as DEFAULT_GEMINI_API_URL +from headroom.proxy.upstream_guard import is_safe_upstream_url DEFAULT_CLOUDCODE_API_URL = "https://cloudcode-pa.googleapis.com" DEFAULT_VERTEX_API_URL = "https://us-central1-aiplatform.googleapis.com" @@ -79,7 +80,9 @@ class ProxyProviderRuntime: return self.api_targets.gemini if headers.get("api-key"): azure_base = headers.get("x-headroom-base-url", "") - if azure_base: + # Same SSRF guard as `proxy_targets.select_passthrough_base_url`; + # both resolve a caller-named upstream (CVE-2026-77775). + if azure_base and is_safe_upstream_url(azure_base): return azure_base.rstrip("/") return self.api_targets.openai diff --git a/headroom/proxy/upstream_guard.py b/headroom/proxy/upstream_guard.py index 783b6e9ab..bfe172696 100644 --- a/headroom/proxy/upstream_guard.py +++ b/headroom/proxy/upstream_guard.py @@ -20,13 +20,38 @@ import from any handler without risking an import cycle. from __future__ import annotations +import asyncio import ipaddress import os import socket +from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import TimeoutError as _FutureTimeout from urllib.parse import urlparse ALLOWED_BASE_URLS_ENV = "HEADROOM_ALLOWED_BASE_URLS" +# `socket.getaddrinfo` has no timeout parameter and runs on whatever thread +# calls it -- which, for the proxy, is the event loop. A caller-supplied host +# that resolves slowly therefore stalls every other in-flight request, so the +# lookup is bounded here and fails closed when it overruns. Callers already in +# async context should prefer `is_safe_upstream_url_async`, which keeps the +# wait off the loop entirely. +RESOLVE_TIMEOUT_ENV = "HEADROOM_UPSTREAM_RESOLVE_TIMEOUT_S" +_DEFAULT_RESOLVE_TIMEOUT_S = 3.0 +_RESOLVER_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="hr-upstream-dns") + + +def _resolve_timeout_seconds() -> float: + raw = (os.environ.get(RESOLVE_TIMEOUT_ENV) or "").strip() + if not raw: + return _DEFAULT_RESOLVE_TIMEOUT_S + try: + value = float(raw) + except ValueError: + return _DEFAULT_RESOLVE_TIMEOUT_S + return value if value > 0 else _DEFAULT_RESOLVE_TIMEOUT_S + + _SAFE_SCHEMES = {"http", "https", "ws", "wss"} @@ -58,19 +83,53 @@ def _allowlisted_destinations() -> tuple[set[str], set[tuple[str, str, int]]] | return hosts, origins +# RFC 6052 / RFC 8215: these IPv6 prefixes embed an IPv4 address in their low +# 32 bits, and `ipaddress` reports the well-known one as globally routable. On a +# NAT64 network `64:ff9b::7f00:1` reaches 127.0.0.1, so the embedded address is +# what has to be judged. 6to4, Teredo and IPv4-mapped forms are already caught +# by the `is_global` test below. +_NAT64_PREFIXES = ( + ipaddress.IPv6Network("64:ff9b::/96"), + ipaddress.IPv6Network("64:ff9b:1::/48"), +) + + +def _nat64_embedded_ipv4(addr: ipaddress.IPv6Address) -> ipaddress.IPv4Address | None: + if not any(addr in prefix for prefix in _NAT64_PREFIXES): + return None + try: + return ipaddress.IPv4Address(int(addr) & 0xFFFFFFFF) + except (ipaddress.AddressValueError, ValueError): # pragma: no cover - defensive + return None + + def _is_internal_address(ip: str) -> bool: try: addr = ipaddress.ip_address(ip) except ValueError: return True # unparseable (e.g. scoped link-local) -> treat as unsafe - return ( + if ( addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved or addr.is_multicast or addr.is_unspecified - ) + ): + return True + # Anything not globally routable. This is what catches RFC 6598 shared + # address space (100.64.0.0/10) -- which `is_private` does not flag, and + # which reaches ISP and cloud-internal infrastructure -- along with + # benchmarking (198.18/15), TEST-NET, 240/4, 6to4 and Teredo tunnels that + # embed an internal IPv4, and any future special-use range the stdlib + # learns about. + if not addr.is_global: + return True + if isinstance(addr, ipaddress.IPv6Address): + embedded = _nat64_embedded_ipv4(addr) + if embedded is not None and _is_internal_address(str(embedded)): + return True + return False def is_safe_upstream_url(url: str) -> bool: @@ -101,10 +160,22 @@ def is_safe_upstream_url(url: str) -> bool: return (parsed.scheme.lower(), host.lower(), port) in origins try: - infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP) - except OSError: + infos = _RESOLVER_POOL.submit( + socket.getaddrinfo, host, None, 0, 0, socket.IPPROTO_TCP + ).result(timeout=_resolve_timeout_seconds()) + except (OSError, _FutureTimeout): # Resolution and connection are separate operations, so allowing a DNS # miss here would fail open if the name resolves on the later lookup. + # A lookup that overruns the budget is treated the same way. # Operators can explicitly allowlist split-horizon/internal endpoints. return False return all(not _is_internal_address(str(info[4][0])) for info in infos) + + +async def is_safe_upstream_url_async(url: str) -> bool: + """Async form of :func:`is_safe_upstream_url` for event-loop callers. + + Same policy; the blocking resolution runs off the loop so a hostile or + slow-resolving hostname cannot stall unrelated in-flight requests. + """ + return await asyncio.to_thread(is_safe_upstream_url, url) diff --git a/tests/test_provider_proxy_routes.py b/tests/test_provider_proxy_routes.py index cfd39e0fa..5f561ae06 100644 --- a/tests/test_provider_proxy_routes.py +++ b/tests/test_provider_proxy_routes.py @@ -10,6 +10,7 @@ from fastapi.responses import JSONResponse from fastapi.testclient import TestClient from headroom.providers.codex.runtime import CodexRoutingDecision +from headroom.proxy import upstream_guard from headroom.proxy.project_context import get_current_project from headroom.proxy.server import HeadroomProxy, ProxyConfig, create_app @@ -430,12 +431,21 @@ def test_proxy_route_helpers_prefer_legacy_targets_and_gemini_passthrough() -> N assert proxy_routes._select_passthrough_base_url(proxy, {"x-goog-api-key": "test"}) == ( "https://legacy.gemini.test" ) - assert ( - proxy_routes._select_passthrough_base_url( - proxy, {"api-key": "azure", "x-headroom-base-url": "https://azure.example/base/"} + # The azure branch honours the override, but only after the SSRF guard + # clears the destination (CVE-2026-77775). `azure.example` does not + # resolve, and the guard fails closed on resolution failure, so pin a + # public answer to keep this assertion about target *precedence*. + with patch.object( + upstream_guard.socket, + "getaddrinfo", + return_value=[(None, None, None, None, ("20.10.10.10", 443))], + ): + assert ( + proxy_routes._select_passthrough_base_url( + proxy, {"api-key": "azure", "x-headroom-base-url": "https://azure.example/base/"} + ) + == "https://azure.example/base" ) - == "https://azure.example/base" - ) assert proxy_routes._select_passthrough_base_url(proxy, {"api-key": "azure"}) == ( "https://legacy.anthropic.test" ) diff --git a/tests/test_provider_proxy_targets.py b/tests/test_provider_proxy_targets.py index b8f4412c5..ed1385745 100644 --- a/tests/test_provider_proxy_targets.py +++ b/tests/test_provider_proxy_targets.py @@ -1,11 +1,14 @@ from __future__ import annotations +from unittest.mock import patch + from headroom.providers.proxy_targets import ( api_target, select_passthrough_base_url, vertex_target_for_location, ) from headroom.providers.registry import DEFAULT_VERTEX_API_URL +from headroom.proxy import upstream_guard def _proxy(**legacy_targets: str): @@ -56,13 +59,21 @@ def test_select_passthrough_base_url_handles_special_auth_modes() -> None: assert select_passthrough_base_url(proxy, {"x-goog-api-key": "test"}) == ( "https://legacy.gemini.test" ) - assert ( - select_passthrough_base_url( - proxy, - {"api-key": "azure", "x-headroom-base-url": "https://azure.example/base/"}, + # The Azure branch honours the override only after the SSRF guard clears + # the destination (CVE-2026-77775), and `azure.example` does not resolve. + # Pin a public answer so this stays a test of target *precedence*. + with patch.object( + upstream_guard.socket, + "getaddrinfo", + return_value=[(None, None, None, None, ("20.10.10.10", 443))], + ): + assert ( + select_passthrough_base_url( + proxy, + {"api-key": "azure", "x-headroom-base-url": "https://azure.example/base/"}, + ) + == "https://azure.example/base" ) - == "https://azure.example/base" - ) assert select_passthrough_base_url(proxy, {"x-api-key": "anthropic"}) == ( "https://legacy.anthropic.test" ) diff --git a/tests/test_provider_registry_extended.py b/tests/test_provider_registry_extended.py index c3ef4559a..e70e14175 100644 --- a/tests/test_provider_registry_extended.py +++ b/tests/test_provider_registry_extended.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging from types import SimpleNamespace from typing import Any +from unittest.mock import patch import pytest @@ -13,6 +14,7 @@ from headroom.providers.registry import ( create_proxy_backend, format_backend_status, ) +from headroom.proxy import upstream_guard class DummyStorage: @@ -85,12 +87,20 @@ def test_proxy_provider_runtime_selects_targets_and_providers() -> None: assert runtime.select_passthrough_base_url({"x-goog-api-key": "test"}) == ( "https://gemini.example" ) - assert ( - runtime.select_passthrough_base_url( - {"api-key": "azure-key", "x-headroom-base-url": "https://azure.example/openai/"} + # The Azure branch honours the override only after the SSRF guard clears + # the destination (CVE-2026-77775), and `azure.example` does not resolve. + # Pin a public answer so this stays a test of target *precedence*. + with patch.object( + upstream_guard.socket, + "getaddrinfo", + return_value=[(None, None, None, None, ("20.10.10.10", 443))], + ): + assert ( + runtime.select_passthrough_base_url( + {"api-key": "azure-key", "x-headroom-base-url": "https://azure.example/openai/"} + ) + == "https://azure.example/openai" ) - == "https://azure.example/openai" - ) assert runtime.select_passthrough_base_url({}) == "https://openai.example" diff --git a/tests/test_upstream_guard.py b/tests/test_upstream_guard.py index af46ba8af..e38693324 100644 --- a/tests/test_upstream_guard.py +++ b/tests/test_upstream_guard.py @@ -5,10 +5,16 @@ All cases use IP literals or ``localhost`` so no external network is required. from __future__ import annotations +import re import socket +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer import pytest +from fastapi.testclient import TestClient +from headroom.providers.proxy_targets import select_passthrough_base_url +from headroom.proxy.server import ProxyConfig, create_app from headroom.proxy.upstream_guard import is_safe_upstream_url @@ -36,6 +42,39 @@ def test_allows_public(url: str) -> None: assert is_safe_upstream_url(url) is True +@pytest.mark.parametrize( + ("label", "url"), + [ + # RFC 6598 shared address space: `is_private` does not flag it, but it + # routes to ISP and cloud-internal infrastructure. + ("shared address space", "http://100.64.0.1/"), + ("shared address space top", "http://100.127.255.254/"), + ("benchmarking", "http://198.18.0.1/"), + ("TEST-NET-1", "http://192.0.2.1/"), + ("TEST-NET-3", "http://203.0.113.1/"), + ("reserved 240/4", "http://240.0.0.1/"), + ("IETF protocol assignments", "http://192.0.0.1/"), + # IPv6 forms that embed an internal IPv4 address. + ("6to4 embedding loopback", "http://[2002:7f00:1::]/"), + ("6to4 embedding RFC1918", "http://[2002:a00:1::]/"), + ("NAT64 embedding loopback", "http://[64:ff9b::7f00:1]/"), + ("NAT64 local-use prefix", "http://[64:ff9b:1::7f00:1]/"), + ("teredo", "http://[2001:0::7f00:1]/"), + ("IPv4-mapped metadata", "http://[::ffff:169.254.169.254]/"), + ("IPv4-mapped loopback", "http://[::ffff:127.0.0.1]/"), + # Credential-prefix confusion: the authority is what counts. + ("userinfo before loopback", "http://api.openai.com@127.0.0.1/"), + ], +) +def test_blocks_non_globally_routable_and_embedded_forms(label: str, url: str) -> None: + assert is_safe_upstream_url(url) is False, label + + +def test_multicast_is_still_blocked() -> None: + """`is_global` is True for multicast, so the category checks must remain.""" + assert is_safe_upstream_url("http://224.0.0.1/") is False + + def test_dns_failure_is_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None: def fail_resolution(*args: object, **kwargs: object) -> list[object]: raise socket.gaierror("temporary failure") @@ -57,3 +96,205 @@ def test_allowlist_mode(monkeypatch: pytest.MonkeyPatch) -> None: # Anything not on the list is rejected in allowlist mode, even public hosts. assert is_safe_upstream_url("https://8.8.8.8/v1") is False assert is_safe_upstream_url("https://api.openai.com/v1") is False + + +# --------------------------------------------------------------------------- +# Enforcement at the sinks (CVE-2026-77775). +# +# The tests above cover `is_safe_upstream_url` in isolation. They passed while +# `/v1/alpha/search` still forwarded to any caller-named host, because nothing +# asserted the guard was actually *reached*. `select_passthrough_base_url` +# returns the `x-headroom-base-url` value whenever an `api-key` header is +# present -- both attacker-supplied -- so every caller of it is a sink. +# --------------------------------------------------------------------------- + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +class _InternalService: + """Stands in for an internal host the caller should never be able to reach.""" + + def __init__(self) -> None: + self.hits: list[str] = [] + self.port = _free_port() + hits = self.hits + + class _Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + hits.append(self.path) + body = b'{"secret":"internal-only"}' + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_GET = do_POST # noqa: N815 + + def log_message(self, *args: object) -> None: + return + + self._server = HTTPServer(("127.0.0.1", self.port), _Handler) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + + def __enter__(self) -> _InternalService: + self._thread.start() + return self + + def __exit__(self, *exc: object) -> None: + self._server.shutdown() + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.port}" + + +def _app(): # noqa: ANN202 + return create_app( + ProxyConfig( + host="127.0.0.1", + port=_free_port(), + optimize=False, + cache_enabled=False, + rate_limit_enabled=False, + ) + ) + + +def test_alpha_search_rejects_a_caller_named_loopback_upstream() -> None: + """The route that shipped unguarded. 400, and the host is never contacted.""" + with _InternalService() as internal, TestClient(_app()) as client: + response = client.post( + "/v1/alpha/search", + headers={ + "api-key": "attacker-supplied", + "Authorization": "Bearer client-token", + "x-headroom-base-url": internal.url, + }, + json={"query": "x"}, + ) + + assert response.status_code == 400 + assert internal.hits == [], "proxy forwarded to a loopback address" + assert "internal-only" not in response.text + + +def test_no_route_forwards_to_a_loopback_upstream() -> None: + """Sweep the whole route table -- the guard must hold everywhere. + + This is the generalisation of the fix: a future route that resolves a + caller-named upstream without validating it fails here rather than in a + CVE. + """ + app = _app() + probes: set[tuple[str, str]] = set() + for route in app.routes: + path = getattr(route, "path", None) + methods = getattr(route, "methods", None) or set() + if not path: + continue + path = re.sub(r"\{[^}]+\}", "probe", path) + for method in ("POST", "GET"): + if method in methods: + probes.add((method, path)) + break + + assert len(probes) > 50, "route discovery found suspiciously few routes" + + with _InternalService() as internal, TestClient(app) as client: + for method, path in sorted(probes): + for unlock in ({"api-key": "x"}, {"x-goog-api-key": "x"}): + headers = {**unlock, "x-headroom-base-url": internal.url} + try: + client.request(method, path, headers=headers, json={"q": "x"}) + except Exception: # noqa: BLE001 - route errors are not the subject + pass + reached = list(internal.hits) + + assert reached == [], f"routes forwarded to a loopback upstream: {reached}" + + +class _StubProxy: + """Minimal stand-in for the proxy object `select_passthrough_base_url` reads.""" + + class provider_runtime: # noqa: N801 + @staticmethod + def model_metadata_provider(headers: object) -> str: + return "openai" + + @staticmethod + def api_target(name: str) -> str: + return "https://api.openai.com" + + +def test_passthrough_base_url_ignores_an_unsafe_azure_override() -> None: + headers = {"api-key": "x", "x-headroom-base-url": "http://169.254.169.254"} + + resolved = select_passthrough_base_url(_StubProxy(), headers) + + assert "169.254.169.254" not in resolved + + +def test_passthrough_base_url_still_honours_a_safe_azure_override( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Legitimate BYOK must keep working -- this is not a blanket block.""" + + def public_resolution(*args: object, **kwargs: object) -> list[object]: + return [(None, None, None, None, ("20.10.10.10", 443))] + + monkeypatch.setattr(socket, "getaddrinfo", public_resolution) + headers = { + "api-key": "x", + "x-headroom-base-url": "https://my-resource.openai.azure.com/", + } + + resolved = select_passthrough_base_url(_StubProxy(), headers) + + assert resolved == "https://my-resource.openai.azure.com" + + +def test_operator_allowlist_still_permits_an_internal_azure_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """On-prem/split-horizon deployments opt in explicitly rather than being stuck.""" + monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "gateway.internal") + headers = {"api-key": "x", "x-headroom-base-url": "https://gateway.internal/v1"} + + assert select_passthrough_base_url(_StubProxy(), headers) == "https://gateway.internal/v1" + + +def test_slow_resolution_is_bounded_and_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """A hostile hostname must not hold the caller for the resolver's timeout. + + `socket.getaddrinfo` takes no timeout and runs on the calling thread, which + for the proxy is the event loop -- so an unbounded lookup is an + unauthenticated stall of every in-flight request. + """ + import time as _time + + def slow_resolution(*args: object, **kwargs: object) -> list[object]: + _time.sleep(5.0) + return [(None, None, None, None, ("8.8.8.8", 443))] + + monkeypatch.setenv("HEADROOM_UPSTREAM_RESOLVE_TIMEOUT_S", "0.25") + monkeypatch.setattr(socket, "getaddrinfo", slow_resolution) + + started = _time.perf_counter() + result = is_safe_upstream_url("https://slow.example/v1") + elapsed = _time.perf_counter() - started + + assert result is False, "a lookup that overruns its budget must fail closed" + assert elapsed < 2.0, f"resolution was not bounded (took {elapsed:.2f}s)" + + +async def test_async_guard_matches_the_sync_policy() -> None: + """The off-loop wrapper must not diverge from the blocking form.""" + from headroom.proxy.upstream_guard import is_safe_upstream_url_async + + assert await is_safe_upstream_url_async("http://127.0.0.1/") is False + assert await is_safe_upstream_url_async("http://169.254.169.254/") is False + assert await is_safe_upstream_url_async("https://8.8.8.8/v1") is True