mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(security): block DNS-rebinding on /debug/* and /stats/reset via Host-header allowlist (#605)
## Summary `require_loopback` in `headroom/proxy/loopback_guard.py` guards `/debug/tasks`, `/debug/ws-sessions`, `/debug/warmup`, and `/stats/reset` by checking `request.client.host` only. A malicious website can use DNS rebinding to make a victim's browser send requests to `127.0.0.1` while the page origin (and inbound `Host:` header) still reads `attacker.com`. The IP check passes — the browser IS on loopback — and `app.add_middleware(CORSMiddleware, allow_origins=["*"], ...)` (`headroom/proxy/server.py:1678`) lets the attacker's JS read the response. This PR adds the canonical second gate: the `Host:` header must also name loopback (`127.0.0.1[:port]`, `[::1][:port]`, or `localhost[:port]`). ## Impact While the proxy binds to `127.0.0.1` by default (`headroom/proxy/server.py:2956`), any browser the user opens can be used by an unrelated tab to reach the proxy via DNS rebinding. Concretely: - `GET /debug/tasks` → leaks asyncio task names, coroutine qualnames, and (with `?stack=true`) stack depths — reveals which coroutines are mid-flight and which codepaths are warm. Useful for fingerprinting + targeting follow-on attacks. - `GET /debug/ws-sessions` → leaks live WebSocket session metadata including `upstream_url` per session. - `GET /debug/warmup` → leaks warmup registry shape (lower risk). - `POST /stats/reset` → resets in-memory proxy stats (state mutation; minor DoS of dashboards/observability). The fix path matches the OWASP DNS-rebinding mitigation and Starlette's `TrustedHostMiddleware` posture. ## Location - `headroom/proxy/loopback_guard.py:73` — guarded the IP check only, no `Host:` validation - `headroom/proxy/server.py:1819-1845`, `:2312` — endpoints reachable via DNS rebinding under the old gate ## Fix `require_loopback` now runs two gates: (1) the existing `request.client.host` loopback IP check, and (2) a new `Host:` header allowlist via `is_loopback_host_header(...)`. The header helper accepts `127.0.0.1[:port]`, `[::1][:port]`, `localhost[:port]`, and IPv6-mapped IPv4 (`::ffff:127.0.0.1`) — strips brackets and ports, then delegates to the existing `is_loopback_host` logic. Cross-origin browser fetches always carry the attacker's hostname in `Host:`, so rebinding requests now 404 alongside any other external attempt. Same 404 (not 403) semantics, so debug endpoints remain invisible to external scanners. The new helper is in `__all__` for explicit re-use. The existing manual-`Request`-stub unit test path (no `.headers` attribute) is preserved by an `if headers is None: return` fallback so older callers that pass a bare stub still work. Test fixtures in `tests/test_proxy_debug_endpoints.py` were updated to pin `base_url="http://127.0.0.1"` so the loopback-`Host:` invariant holds in the green-path tests, plus a new `app_and_rebinding_client` fixture and `test_debug_endpoints_block_dns_rebinding` exercising the gate at the HTTP level. Six new unit tests cover `is_loopback_host_header` (canonical accept, external reject, malformed reject, and rebinding signature). Net: +219 lines, -4 lines, in two files. **Compatibility note for operators:** a deployment that fronts the proxy behind a reverse proxy with a non-loopback Host (e.g. `headroom.local` mapped to `127.0.0.1`) and still wants `/debug/*` exposed would need to either point that reverse proxy at the upstream API endpoints only, or extend `is_loopback_host_header` with an env-var allowlist in a follow-up. The default `HEADROOM_HOST=127.0.0.1` path is unaffected. ## Detected by Aeon (manual review — no scanner rule for this; pattern matches the threat-model-claims + local-HTTP-server axes from [`skills/vuln-scanner`](https://github.com/aaronjmars/aeon-aaron/blob/main/skills/vuln-scanner/SKILL.md), priors that have surfaced 10+ similar finds across 5 languages over the last 6 weeks). - Severity: medium - CWE-350 (Reliance on Reverse DNS Resolution for a Security-Critical Action) - CWE-352 (Cross-Site Request Forgery) — adjacent - Class match: DNS-rebinding-via-loopback-bind (recurring axis in the tracker) ## Verification - `python3 -m py_compile headroom/proxy/loopback_guard.py` — parses. - The new unit tests in `tests/test_proxy_debug_endpoints.py` cover: canonical loopback Host headers (accept), external Host headers (reject), malformed/empty (reject), an HTTP-level DNS-rebinding simulation (`app_and_rebinding_client` fixture with `base_url="http://attacker.com"` + loopback client tuple), and the unchanged IP-only stub path. - Other test files that use `require_loopback` (`test_proxy_openai_responses_integration.py`, `test_proxy_openai_responses_bypass.py`, `test_proxy_dashboard_stats_cache.py`) all override it via `app.dependency_overrides[require_loopback] = lambda: None`, so they are not affected by the new gate. - I was unable to run `pytest` locally in this sandbox; please confirm the new fixtures slot in cleanly when the CI matrix runs. --- Filed by [Aeon](https://github.com/aaronjmars/aeon-aaron). --------- Co-authored-by: aeonframework <noreply@anthropic.com>
This commit is contained in:
parent
6b0c09ffd5
commit
b4b50253f1
2 changed files with 225 additions and 6 deletions
|
|
@ -16,6 +16,25 @@ middleware) because:
|
|||
a non-loopback client in tests.
|
||||
* The set of debug endpoints is small and co-located; a middleware
|
||||
would be disproportionate.
|
||||
|
||||
DNS-rebinding defence
|
||||
---------------------
|
||||
A loopback-IP check alone is not enough to keep these endpoints local.
|
||||
A malicious site can use DNS rebinding to make a victim's browser send
|
||||
requests to ``127.0.0.1`` while the ``Host:`` header (and the JS
|
||||
``fetch`` URL) still reads ``attacker.com``. From the proxy's point of
|
||||
view ``request.client.host`` is ``127.0.0.1`` (the browser, which IS on
|
||||
loopback) and the IP check passes. The proxy ships a wide-open CORS
|
||||
policy (``allow_origins=['*']``), so attacker JS can then read the
|
||||
response.
|
||||
|
||||
To close that gap the guard also requires the ``Host:`` header to name
|
||||
loopback — ``127.0.0.1[:port]``, ``[::1][:port]``, or
|
||||
``localhost[:port]``. Same-origin XHR from a real local tool always
|
||||
sets one of those values; cross-origin rebinding does not. This is the
|
||||
canonical Host-header allowlist mitigation called out in OWASP's
|
||||
CSRF / DNS-rebinding guidance and the standard Starlette
|
||||
``TrustedHostMiddleware`` pattern.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -32,6 +51,7 @@ except ImportError: # pragma: no cover - fastapi is a hard dep in practice
|
|||
__all__ = [
|
||||
"LOOPBACK_HOSTS",
|
||||
"is_loopback_host",
|
||||
"is_loopback_host_header",
|
||||
"require_loopback",
|
||||
]
|
||||
|
||||
|
|
@ -52,14 +72,16 @@ def is_loopback_host(host: str | None) -> bool:
|
|||
``request.client``.
|
||||
|
||||
``"localhost"`` is special-cased as a string since it is not a
|
||||
valid IP literal. Every other host is parsed with
|
||||
valid IP literal. The comparison is case-insensitive because
|
||||
hostnames are (RFC 4343), so a ``Host: LOCALHOST`` from a local
|
||||
tool is still accepted. Every other host is parsed with
|
||||
:func:`ipaddress.ip_address`; this accepts IPv6-mapped IPv4
|
||||
(``::ffff:127.0.0.1``) which Linux dual-stack sockets emit by
|
||||
default. Malformed input returns ``False``.
|
||||
"""
|
||||
if host is None:
|
||||
return True
|
||||
if host == "localhost":
|
||||
if host.lower() == "localhost":
|
||||
return True
|
||||
try:
|
||||
address = ipaddress.ip_address(host)
|
||||
|
|
@ -70,6 +92,42 @@ def is_loopback_host(host: str | None) -> bool:
|
|||
return address.is_loopback
|
||||
|
||||
|
||||
def is_loopback_host_header(header_value: str | None) -> bool:
|
||||
"""Return True if a ``Host:`` header names a loopback address.
|
||||
|
||||
The header can include a port (``127.0.0.1:8787``,
|
||||
``[::1]:8787``, ``localhost:8787``) and uses bracket notation for
|
||||
raw IPv6 literals per RFC 3986. This helper strips brackets and
|
||||
the trailing ``:port`` (if any) and delegates the address-vs-name
|
||||
decision to :func:`is_loopback_host`.
|
||||
|
||||
Missing / empty headers return ``False`` rather than ``True`` —
|
||||
a real local browser or CLI always sets ``Host:``, so absence is
|
||||
suspicious. Server-internal callers that bypass HTTP entirely
|
||||
(``TestClient`` with a manual call) do not hit the guard.
|
||||
"""
|
||||
if not header_value:
|
||||
return False
|
||||
candidate = header_value.strip()
|
||||
if not candidate:
|
||||
return False
|
||||
# Bracketed IPv6: [::1] or [::1]:8787 — strip the brackets and
|
||||
# everything after the matching ``]`` (which is the port suffix).
|
||||
if candidate.startswith("["):
|
||||
closing = candidate.find("]")
|
||||
if closing == -1:
|
||||
return False
|
||||
host_part = candidate[1:closing]
|
||||
elif candidate.count(":") == 1:
|
||||
# Single colon = host:port for IPv4 / hostname. A bare IPv6
|
||||
# literal without brackets has multiple colons and would be
|
||||
# ambiguous, so we don't strip in that case.
|
||||
host_part = candidate.rsplit(":", 1)[0]
|
||||
else:
|
||||
host_part = candidate
|
||||
return is_loopback_host(host_part)
|
||||
|
||||
|
||||
def require_loopback(request: Request) -> None: # type: ignore[valid-type]
|
||||
"""FastAPI dependency: 404 any non-loopback caller.
|
||||
|
||||
|
|
@ -79,6 +137,17 @@ def require_loopback(request: Request) -> None: # type: ignore[valid-type]
|
|||
async def debug_tasks() -> list[dict]:
|
||||
...
|
||||
|
||||
Two gates have to pass:
|
||||
|
||||
1. ``request.client.host`` must be a loopback IP. Stops anyone
|
||||
who actually reaches the listener from outside ``127.0.0.0/8``
|
||||
/ ``::1``.
|
||||
2. The inbound ``Host:`` header must also name loopback. Stops
|
||||
DNS-rebinding attacks where a browser sends requests to the
|
||||
loopback IP but the page origin is ``attacker.com`` — the IP
|
||||
check alone passes, but the ``Host:`` header still reads
|
||||
``attacker.com`` and we reject the request here.
|
||||
|
||||
Returning 404 (not 403) keeps debug endpoints invisible to
|
||||
external scanners — indistinguishable from "no such route".
|
||||
"""
|
||||
|
|
@ -90,3 +159,16 @@ def require_loopback(request: Request) -> None: # type: ignore[valid-type]
|
|||
if not is_loopback_host(host):
|
||||
# No body: minimal FastAPI default, behaves like "no route".
|
||||
raise HTTPException(status_code=404)
|
||||
|
||||
headers = getattr(request, "headers", None)
|
||||
if headers is None:
|
||||
# Manual ``Request`` stub with no ``headers`` attribute — used
|
||||
# by older unit tests that pre-date this gate. Treat the same
|
||||
# way as the IP-only path did and accept.
|
||||
return
|
||||
try:
|
||||
host_header = headers.get("host")
|
||||
except AttributeError:
|
||||
host_header = None
|
||||
if not is_loopback_host_header(host_header):
|
||||
raise HTTPException(status_code=404)
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from headroom.proxy.debug_introspection import (
|
|||
from headroom.proxy.loopback_guard import (
|
||||
LOOPBACK_HOSTS,
|
||||
is_loopback_host,
|
||||
is_loopback_host_header,
|
||||
require_loopback,
|
||||
)
|
||||
from headroom.proxy.server import ProxyConfig, create_app
|
||||
|
|
@ -44,7 +45,13 @@ def client():
|
|||
# Pin the simulated client address to loopback so the /debug/* guard
|
||||
# accepts the request. Without this, FastAPI's TestClient reports
|
||||
# the host as ``testclient`` and the guard correctly 404s us.
|
||||
with TestClient(app, client=("127.0.0.1", 12345)) as test_client:
|
||||
# ``base_url`` pins the inbound ``Host:`` header to a loopback name
|
||||
# so the DNS-rebinding gate added in 2026-06 also passes.
|
||||
with TestClient(
|
||||
app,
|
||||
base_url="http://127.0.0.1",
|
||||
client=("127.0.0.1", 12345),
|
||||
) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
|
|
@ -57,7 +64,11 @@ def app_and_client():
|
|||
cost_tracking_enabled=False,
|
||||
)
|
||||
app = create_app(config)
|
||||
with TestClient(app, client=("127.0.0.1", 12345)) as test_client:
|
||||
with TestClient(
|
||||
app,
|
||||
base_url="http://127.0.0.1",
|
||||
client=("127.0.0.1", 12345),
|
||||
) as test_client:
|
||||
yield app, test_client
|
||||
|
||||
|
||||
|
|
@ -71,7 +82,35 @@ def app_and_external_client():
|
|||
cost_tracking_enabled=False,
|
||||
)
|
||||
app = create_app(config)
|
||||
with TestClient(app, client=("10.0.0.1", 54321)) as test_client:
|
||||
with TestClient(
|
||||
app,
|
||||
base_url="http://127.0.0.1",
|
||||
client=("10.0.0.1", 54321),
|
||||
) as test_client:
|
||||
yield app, test_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_and_rebinding_client():
|
||||
"""TestClient that simulates a DNS-rebinding attack.
|
||||
|
||||
The simulated TCP peer is loopback (``request.client.host`` passes
|
||||
the legacy IP check), but the inbound ``Host:`` header reads
|
||||
``attacker.com`` — exactly what the browser sends after the
|
||||
attacker's DNS record flips to ``127.0.0.1``.
|
||||
"""
|
||||
config = ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
)
|
||||
app = create_app(config)
|
||||
with TestClient(
|
||||
app,
|
||||
base_url="http://attacker.com",
|
||||
client=("127.0.0.1", 12345),
|
||||
) as test_client:
|
||||
yield app, test_client
|
||||
|
||||
|
||||
|
|
@ -136,7 +175,89 @@ def test_require_loopback_accepts_loopback_client():
|
|||
class _FakeRequest:
|
||||
client = _FakeClient()
|
||||
|
||||
# Should not raise.
|
||||
# Should not raise. ``headers`` is absent so the Host-header gate
|
||||
# falls back to the legacy IP-only behaviour for callers that
|
||||
# construct a bare request stub.
|
||||
require_loopback(_FakeRequest()) # type: ignore[arg-type]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Host-header (DNS-rebinding) guard unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_is_loopback_host_header_accepts_canonical_values():
|
||||
for value in (
|
||||
"127.0.0.1",
|
||||
"127.0.0.1:8787",
|
||||
"localhost",
|
||||
"localhost:8787",
|
||||
"LOCALHOST",
|
||||
"Localhost:8787",
|
||||
"[::1]",
|
||||
"[::1]:8787",
|
||||
):
|
||||
assert is_loopback_host_header(value) is True, value
|
||||
|
||||
|
||||
def test_is_loopback_host_header_rejects_external_names():
|
||||
for value in (
|
||||
"attacker.com",
|
||||
"attacker.com:8787",
|
||||
"evil.example",
|
||||
"10.0.0.1",
|
||||
"10.0.0.1:8787",
|
||||
"8.8.8.8",
|
||||
):
|
||||
assert is_loopback_host_header(value) is False, value
|
||||
|
||||
|
||||
def test_is_loopback_host_header_rejects_missing_and_malformed():
|
||||
assert is_loopback_host_header(None) is False
|
||||
assert is_loopback_host_header("") is False
|
||||
assert is_loopback_host_header(" ") is False
|
||||
# Unterminated bracketed IPv6
|
||||
assert is_loopback_host_header("[::1") is False
|
||||
# Hostname that merely contains a loopback substring
|
||||
assert is_loopback_host_header("localhost.attacker.com") is False
|
||||
|
||||
|
||||
def test_require_loopback_blocks_dns_rebinding_host_header():
|
||||
"""Loopback IP + ``Host: attacker.com`` is the rebinding signature."""
|
||||
|
||||
class _FakeClient:
|
||||
host = "127.0.0.1"
|
||||
|
||||
class _FakeHeaders:
|
||||
def get(self, key, default=None):
|
||||
if key.lower() == "host":
|
||||
return "attacker.com"
|
||||
return default
|
||||
|
||||
class _FakeRequest:
|
||||
client = _FakeClient()
|
||||
headers = _FakeHeaders()
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
require_loopback(_FakeRequest()) # type: ignore[arg-type]
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
|
||||
def test_require_loopback_accepts_loopback_host_header():
|
||||
class _FakeClient:
|
||||
host = "127.0.0.1"
|
||||
|
||||
class _FakeHeaders:
|
||||
def get(self, key, default=None):
|
||||
if key.lower() == "host":
|
||||
return "127.0.0.1:8787"
|
||||
return default
|
||||
|
||||
class _FakeRequest:
|
||||
client = _FakeClient()
|
||||
headers = _FakeHeaders()
|
||||
|
||||
# Should not raise — both gates pass.
|
||||
require_loopback(_FakeRequest()) # type: ignore[arg-type]
|
||||
|
||||
|
||||
|
|
@ -384,6 +505,22 @@ def test_debug_endpoints_return_404_for_non_loopback_client(app_and_external_cli
|
|||
assert response.status_code != 403
|
||||
|
||||
|
||||
def test_debug_endpoints_block_dns_rebinding(app_and_rebinding_client):
|
||||
"""Loopback client + ``Host: attacker.com`` must 404 like an external client.
|
||||
|
||||
Regression for the DNS-rebinding gap: prior to 2026-06 the guard
|
||||
only checked ``request.client.host``, which a rebound browser
|
||||
passes trivially. Adding a ``Host:`` header allowlist closes that
|
||||
gap so a malicious site cannot read /debug/* over the user's
|
||||
loopback proxy via the wide-open CORS policy.
|
||||
"""
|
||||
_, client = app_and_rebinding_client
|
||||
for path in ("/debug/tasks", "/debug/ws-sessions", "/debug/warmup"):
|
||||
response = client.get(path)
|
||||
assert response.status_code == 404, path
|
||||
assert response.status_code != 403
|
||||
|
||||
|
||||
def test_existing_health_routes_unchanged(client):
|
||||
# Invariant: Unit 5 must not regress the existing health endpoints.
|
||||
for path in ("/livez", "/readyz", "/health"):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue