fix(security): validate caller-supplied upstreams on every resolution path (#3195)

## 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 <tejas@Tejass-MacBook-Pro.local>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

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("/")

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 (

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
 

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)

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

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

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"
 
 

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