Require a local Host for loopback trust (DNS rebinding)

SECURITY.md

@@ -17,6 +17,15 @@ directly, bypassing Cerberus policy, so scope and rotate it as you would any
 provider key. An unpublished container port alone does not protect against other
 containers on the same network or a privileged host operator.
 
+Loopback peers are trusted as admins, and without `server.api_token_env` they may
+use the inference surface. That trust also requires the request's `Host` to name
+this machine (`localhost`, `*.localhost`, `127.0.0.0/8`, `::1`, or a name listed in
+`server.trusted_hosts`), which defeats DNS rebinding from a local browser. Do not
+put a reverse proxy on the same host in front of a bare `cerberus serve` that
+forwards to 127.0.0.1 with a loopback `Host`: every proxied request would then be
+a loopback peer. Configure admin SSO or tokens and have the proxy pass the public
+`Host` through.
+
 Admin sessions and pending OIDC logins contain identity and authentication
 material. Restrict state-volume access and protect backups. See
 `docs/persistence.md` for supported deployment limits and recovery behavior.

src/cerberus/app.py

@@ -79,6 +79,27 @@ def _release_id() -> str:
 
 _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"})
 
+
+def _host_name(header: str) -> str:
+    """The hostname from a Host header, lowercased, without port or brackets."""
+
+    value = header.strip().lower()
+    if value.startswith("["):
+        return value[1:].split("]", 1)[0]
+    return value.rsplit(":", 1)[0] if value.count(":") == 1 else value
+
+
+def is_local_host_header(header: str | None, trusted: frozenset[str] = frozenset()) -> bool:
+    """Whether a Host header names this machine (or an operator-trusted name)."""
+
+    if not header:
+        return False
+    name = _host_name(header)
+    if name in _LOOPBACK_HOSTS or name in trusted or name.endswith(".localhost"):
+        return True
+    parts = name.split(".")
+    return len(parts) == 4 and parts[0] == "127" and all(p.isdigit() and int(p) < 256 for p in parts)
+
 # The audit read is bounded: the table is lifetime history and a console asking
 # for all of it would grow without limit. No cursor yet — a bounded newest-first
 # window is what the screen needs, and pagination can be designed when something
@@ -296,11 +317,17 @@ def token_matches(token_env: str | None, request: Request) -> bool:
             # non-ASCII header input is a bad credential, never a server error
             return False
 
+    trusted_hosts = frozenset(name.lower() for name in boot_config.server.trusted_hosts)
+
+    def local_host(request: Request) -> bool:
+        return is_local_host_header(request.headers.get("host"), trusted_hosts)
+
     def authenticated(request: Request) -> bool:
-        """Inference-surface gate: the api token, or open when none is configured."""
+        """Inference-surface gate: the api token, or open to local clients when none
+        is configured (the schema only allows that on a loopback bind)."""
 
         if boot_config.server.api_token_env is None:
-            return True
+            return local_host(request)
         return token_matches(boot_config.server.api_token_env, request)
 
     @app.get("/", include_in_schema=False)
@@ -349,8 +376,14 @@ async def health() -> dict[str, Any]:
     # -- admin surface (control plane) ------------------------------------
 
     def is_loopback(request: Request) -> bool:
-        # transport peer address only — forwarding headers are client-forgeable
-        return request.client is not None and request.client.host in _LOOPBACK_HOSTS
+        # transport peer address only — forwarding headers are client-forgeable.
+        # The Host must name this machine too: under DNS rebinding a hostile page
+        # in a local browser reaches 127.0.0.1 with its own hostname as Host.
+        return (
+            request.client is not None
+            and request.client.host in _LOOPBACK_HOSTS
+            and local_host(request)
+        )
 
     def admin_denied(request: Request, *, read_only: bool = False) -> JSONResponse | None:
         """Loopback-only or admin-scoped credential (PLAN Session 9).

src/cerberus/registry/schema.py

@@ -59,6 +59,13 @@ class ServerConfig(BaseModel):
     # only — never discovered — and empty by default, so the boundary does not
     # widen unless an operator says so.
     admin_config_roots: list[str] = Field(default_factory=list)
+    # Loopback trust (admin access without a credential, and the token-less
+    # inference surface) also requires the request's Host to name this machine:
+    # localhost, *.localhost, 127.0.0.0/8 or ::1. A DNS-rebinding page reaches
+    # 127.0.0.1 from the operator's browser under its own hostname, so the peer
+    # address alone cannot tell it from a local client. List any other name
+    # local clients use here.
+    trusted_hosts: list[str] = Field(default_factory=list)
 
     @model_validator(mode="after")
     def validate_credential_separation(self) -> "ServerConfig":

tests/test_admin_candidates.py

@@ -40,7 +40,7 @@ def build_app(tmp_path, **server):
 async def post(app, url, body):
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             return await client.post(url, json=body, headers=CSRF)
 
 
@@ -131,7 +131,7 @@ async def test_staged_candidate_round_trips_through_validate_and_activate(tmp_pa
     app = build_app(tmp_path)
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             staged = await client.post(
                 "/admin/config/stage",
                 json={"updates": {"providers.alpha.quota_cooldown_seconds": 1234}},
@@ -247,7 +247,7 @@ async def test_version_checksum_collision_keeps_its_cerberus_message(tmp_path):
 
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             await client.post("/admin/validate", json={"path": first}, headers=CSRF)
             clash = await client.post("/admin/validate", json={"path": collision}, headers=CSRF)
 

tests/test_admin_credentials.py

@@ -95,7 +95,7 @@ async def test_validated_inference_token_still_denied_on_all_mutations(monkeypat
     app = make_tokened_app(monkeypatch, tmp_path)
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app, client=REMOTE)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as remote:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as remote:
             for url in MUTATION_URLS:
                 response = await remote.post(url, json={}, headers={"authorization": "Bearer inference-token"})
                 assert response.status_code == 403, url
@@ -106,13 +106,13 @@ async def test_validated_admin_token_keeps_read_only_remote_and_loopback_mutatio
     app = make_tokened_app(monkeypatch, tmp_path)
     async with app.router.lifespan_context(app):
         remote_transport = httpx.ASGITransport(app=app, client=REMOTE)
-        async with httpx.AsyncClient(transport=remote_transport, base_url="http://test") as remote:
+        async with httpx.AsyncClient(transport=remote_transport, base_url="http://localhost") as remote:
             admin = {"authorization": "Bearer admin-secret"}
             assert (await remote.get("/admin/events", headers=admin)).status_code == 200
             for url in MUTATION_URLS:
                 assert (await remote.post(url, json={}, headers=admin)).status_code == 403, url
         local_transport = httpx.ASGITransport(app=app, client=("127.0.0.1", 40001))
-        async with httpx.AsyncClient(transport=local_transport, base_url="http://test") as local:
+        async with httpx.AsyncClient(transport=local_transport, base_url="http://localhost") as local:
             # auth + CSRF ok, body invalid
             assert (await local.post("/admin/validate", json={}, headers=CSRF)).status_code == 400
 

tests/test_admin_fields.py

@@ -93,7 +93,7 @@ async def test_config_schema_endpoint_is_read_only_gated(env, tmp_path):
     app = create_app(doc, http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             resp = await client.get("/admin/config/schema")
     assert resp.status_code == 200
     keys = {f["key"] for f in resp.json()["fields"]}
@@ -106,7 +106,7 @@ async def test_stage_endpoint_rejects_non_allow_listed_keys(env, tmp_path):
     app = create_app(doc, http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             bad_key = await client.post(
                 "/admin/config/stage",
                 json={"updates": {"providers.alpha.credentials.main.api_key_env": "EVIL"}},
@@ -127,7 +127,7 @@ async def test_stage_validate_activate_loop_changes_live_behavior(env, tmp_path)
     app = create_app(doc, http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             staged = await client.post(
                 "/admin/config/stage",
                 json={"updates": {"providers.alpha.quota_cooldown_seconds": "45"}},

tests/test_admin_ui.py

@@ -51,7 +51,7 @@ def make_app(monkeypatch, tmp_path, *, token: str | None = None, admin_token: st
 
 def client_for(app, client_addr):
     return httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app, client=client_addr), base_url="http://test"
+        transport=httpx.ASGITransport(app=app, client=client_addr), base_url="http://localhost"
     )
 
 

tests/test_app.py

@@ -50,7 +50,7 @@ def make_config(monkeypatch: pytest.MonkeyPatch, **server) -> CerberusConfig:
 async def call(app, method: str, path: str, **kwargs) -> httpx.Response:
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             return await client.request(method, path, **kwargs)
 
 
@@ -86,7 +86,7 @@ async def upstream(request: httpx.Request) -> httpx.Response:
     app = create_app(make_config(monkeypatch), http_transport=httpx.MockTransport(upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             response = await client.post("/v1/chat/completions", json={"model": "cerberus/main", "messages": []})
             health = await client.get("/admin/health")  # cooldowns are admin-only diagnostics
 
@@ -211,7 +211,7 @@ async def test_external_bind_rejects_unauthenticated_requests(monkeypatch):
     app = create_app(config, http_transport=httpx.MockTransport(lambda _r: httpx.Response(200, json={})))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             unauthorized = await client.get("/v1/models")
             authorized = await client.get("/v1/models", headers={"authorization": "Bearer router-token"})
             non_ascii = await client.get(
@@ -384,7 +384,7 @@ async def upstream(request: httpx.Request) -> httpx.Response:
     app = create_app(make_config(monkeypatch), http_transport=httpx.MockTransport(upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             await client.post("/v1/chat/completions", json={"model": "cerberus/main", "messages": []})
             public = await client.get("/health")
             admin = await client.get("/admin/health")

tests/test_audit.py

@@ -63,7 +63,7 @@ def build(path):
 
 def client_for(app, addr=LOOPBACK):
     return httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app, client=addr), base_url="http://test"
+        transport=httpx.ASGITransport(app=app, client=addr), base_url="http://localhost"
     )
 
 

tests/test_control.py

@@ -68,7 +68,7 @@ async def test_validate_reports_version_and_rejects_bad_configs(env, tmp_path):
     app = create_app(doc, http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             ok = await client.post("/admin/validate", json={"path": good}, headers=CSRF)
             broken = await client.post("/admin/validate", json={"path": bad}, headers=CSRF)
 
@@ -91,7 +91,7 @@ def upstream(request: httpx.Request) -> httpx.Response:
     app = create_app(load_config_document(v1), http_transport=httpx.MockTransport(upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             body = {"model": "cerberus/free", "messages": []}
             await client.post("/v1/chat/completions", json=body)
             activated = await client.post("/admin/activate", json={"path": v2}, headers=CSRF)
@@ -121,7 +121,7 @@ async def test_version_cannot_be_rebound_to_a_different_checksum(env, tmp_path):
 
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             validated = await client.post("/admin/validate", json={"path": collision_path}, headers=CSRF)
             shadowed = await client.post("/admin/shadow", json={"path": collision_path}, headers=CSRF)
             activated = await client.post("/admin/activate", json={"path": collision_path}, headers=CSRF)
@@ -149,7 +149,7 @@ async def test_rolled_back_version_binding_remains_reserved(env, tmp_path):
 
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             assert (await client.post("/admin/activate", json={"path": v2}, headers=CSRF)).status_code == 200
             assert (await client.post("/admin/rollback", headers=CSRF)).status_code == 200
             rebound = await client.post("/admin/activate", json={"path": v2_collision}, headers=CSRF)
@@ -164,7 +164,7 @@ async def test_rollback_without_history_is_conflict(env, tmp_path):
     app = create_app(doc, http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             response = await client.post("/admin/rollback", headers=CSRF)
     assert response.status_code == 409
 
@@ -185,7 +185,7 @@ async def telemetry(request: httpx.Request) -> httpx.Response:
     )
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             await client.post("/v1/chat/completions", json={"model": "cerberus/free", "messages": []})
 
     assert len(events) == 1
@@ -219,7 +219,7 @@ async def telemetry(request: httpx.Request) -> httpx.Response:
     )
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             armed = await client.post("/admin/shadow", json={"path": shadow}, headers=CSRF)
             response = await client.post("/v1/chat/completions", json={"model": "cerberus/free", "messages": []})
             cleared = await client.post("/admin/shadow", json={"path": None}, headers=CSRF)
@@ -241,7 +241,7 @@ async def test_admin_status_reports_active_and_shadow(env, tmp_path):
     app = create_app(load_config_document(active), http_transport=httpx.MockTransport(ok_upstream))
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             status = await client.get("/admin/status")
             config_dump = await client.get("/admin/config/active")
 
@@ -279,7 +279,7 @@ async def telemetry(request: httpx.Request) -> httpx.Response:
     )
     async with app.router.lifespan_context(app):
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
             await client.post("/admin/shadow", json={"path": shadow}, headers=CSRF)
             await client.post("/v1/chat/completions", json={"model": "cerberus/free", "messages": []})
 

tests/test_decisions.py

@@ -100,7 +100,7 @@ def build(tmp_path, raw, *, transport, telemetry_transport=None, name="config.ya
 
 def client_for(app):
     return httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app, client=("127.0.0.1", 40001)), base_url="http://test"
+        transport=httpx.ASGITransport(app=app, client=("127.0.0.1", 40001)), base_url="http://localhost"
     )