fix(security): address u9up assessment findings (WEB-01–07) (#2207)

Hardens client-selected upstreams, memory identity resolution, downloaded binary integrity, telemetry import, Docker defaults, Neo4j credentials, and archive extraction. Refreshes the branch against current main and preserves newer same-origin and loopback protections.

.env.example

@@ -1,3 +1,14 @@
-# Copy this file to .env and fill in real values before running in production.
-# IMPORTANT: Change NEO4J_AUTH before deploying — default credentials are insecure.
+# Copy this file to .env and fill in real values before running.
+# docker-compose.yml requires these — it will refuse to start with defaults.
+
+# Neo4j credentials for the graph memory backend (format: user/password).
 NEO4J_AUTH=neo4j/CHANGEME
+# Password only, for library / non-Docker use of the Neo4j memory backend.
+NEO4J_PASSWORD=CHANGEME
+
+# Proxy token — gates the data plane whenever the proxy is not loopback-only.
+# Generate: openssl rand -hex 32
+HEADROOM_PROXY_TOKEN=CHANGEME
+
+# Optional: set to 0.0.0.0 to expose the proxy on the network (requires a token).
+# HEADROOM_BIND_ADDR=127.0.0.1

.github/workflows/tools-hash-refresh.yml

@@ -0,0 +1,29 @@
+name: tools-hash-refresh
+
+# Enforce that headroom/tools.json SHA-256 pins match the published assets for
+# the currently pinned tool versions. Fails if a version was bumped without
+# refreshing pins (run scripts/refresh_tool_hashes.py locally). See WEB-03.
+
+on:
+  pull_request:
+    paths:
+      - "headroom/tools.json"
+      - "scripts/refresh_tool_hashes.py"
+      - ".github/workflows/tools-hash-refresh.yml"
+  schedule:
+    - cron: "0 6 * * 1" # Mondays 06:00 UTC
+  workflow_dispatch: {}
+
+permissions:
+  contents: read
+
+jobs:
+  verify-pins:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Verify tool SHA-256 pins
+        run: python scripts/refresh_tool_hashes.py --check

.gitignore

@@ -38,6 +38,7 @@ scripts/*
 !scripts/audit_wheel_glibc_symbols.py
 !scripts/replay_codex_ws_load.py
 !scripts/export_kompress_v2_onnx.py
+!scripts/refresh_tool_hashes.py
 !scripts/record_kompress_fixtures.py
 !scripts/record_code_compressor_fixtures.py
 

docker-compose.yml

@@ -44,6 +44,11 @@ services:
     command: ["--host", "0.0.0.0"]
     environment:
       - HEADROOM_HOST=0.0.0.0
+      # The proxy binds 0.0.0.0 *inside* the container (required for Docker port
+      # forwarding); it is confined to host loopback by the published port below.
+      # A proxy token is required so the data plane is never open if you widen the
+      # bind. Generate one with: openssl rand -hex 32
+      - HEADROOM_PROXY_TOKEN=${HEADROOM_PROXY_TOKEN:?set HEADROOM_PROXY_TOKEN (see .env.example; e.g. openssl rand -hex 32)}
       - HOME=/home/nonroot
       # Keep all Headroom read/write state on the named volume below.
       - HEADROOM_WORKSPACE_DIR=/home/nonroot/.headroom
@@ -96,22 +101,19 @@ services:
   neo4j:
     image: neo4j:5.26
     ports:
-      # Loopback-only: NEO4J_AUTH below defaults to a password published in
-      # this file, so an exposed Bolt port is an open database.
+      # Loopback-only to keep the graph store off the network.
       - "127.0.0.1:7474:7474"  # HTTP (Browser)
       - "127.0.0.1:7687:7687"  # Bolt
     # Named volume persists the graph data across container restarts/recreates.
     volumes:
       - neo4j_data:/data
     environment:
-      # Credentials come from .env (NEO4J_AUTH=user/password). The default here
-      # is for LOCAL DEV ONLY — override it before exposing Neo4j anywhere.
-      - NEO4J_AUTH=${NEO4J_AUTH:-neo4j/devpassword}
+      # No default credential — must be supplied (see .env.example).
+      - NEO4J_AUTH=${NEO4J_AUTH:?set NEO4J_AUTH, e.g. neo4j/<strong-password>}
       # APOC: Neo4j's standard procedure library, needed by Headroom's queries.
       - NEO4J_PLUGINS=["apoc"]
-      - NEO4J_apoc_export_file_enabled=true
-      - NEO4J_apoc_import_file_enabled=true
-      - NEO4J_apoc_import_file_use__neo4j__config=true
+      # APOC file import/export stays disabled (its Neo4j default) — it grants
+      # filesystem read/write via stored procedures. Do not enable unless required.
 
 # Named volumes — managed by Docker, survive `docker compose down` (use
 # `docker compose down -v` to delete the stored data as well).

headroom/binaries.py

@@ -226,6 +226,8 @@ def _mirror_url(url: str) -> str:
     mirror = os.environ.get("HEADROOM_BINARIES_MIRROR")
     if not mirror:
         return url
+    if not mirror.startswith("https://"):
+        raise BinaryFetchError(f"HEADROOM_BINARIES_MIRROR must use https:// (got {mirror!r})")
     # Only substitute the github.com host so that paths remain intact.
     for prefix in ("https://github.com", "https://objects.githubusercontent.com"):
         if url.startswith(prefix):
@@ -245,6 +247,8 @@ def _download(url: str, dest: Path, *, progress: bool = True) -> None:
     if not _is_writable_dir(dest.parent):
         raise OSError(f"binary cache directory is not writable: {dest.parent}")
     final_url = _mirror_url(url)
+    if not final_url.startswith("https://"):
+        raise BinaryFetchError(f"refusing non-https download URL: {final_url!r}")
     req = urllib.request.Request(final_url, headers={"User-Agent": "headroom-binaries/1"})
     attempts = 3
     for attempt in range(1, attempts + 1):
@@ -307,10 +311,16 @@ def _sha256_file(path: Path) -> str:
 
 
 def _verify_sha256(path: Path, expected: str | None) -> None:
+    if os.environ.get("HEADROOM_BINARIES_ALLOW_UNVERIFIED"):
+        logger.warning(
+            "skipping sha256 verification for %s (HEADROOM_BINARIES_ALLOW_UNVERIFIED=1)",
+            path.name,
+        )
+        return
     if not expected:
-        # Upstream release not SHA-pinned in registry. HTTPS + the GitHub CDN
-        # is the only integrity check. Log at INFO so verbose runs can see
-        # this state; `doctor` surfaces the same fact via `sha_pinned=False`.
+        # No pin in the registry (e.g. an off-registry version override). All
+        # shipped assets ARE pinned — enforced by the tools-hash-refresh CI gate —
+        # so a missing pin means an off-registry fetch; fall back to HTTPS trust.
         logger.info("binary %s downloaded without sha256 pin (HTTPS trust only)", path.name)
         return
     got = _sha256_file(path)
@@ -319,6 +329,40 @@ def _verify_sha256(path: Path, expected: str | None) -> None:
         raise Sha256Mismatch(f"sha256 mismatch for {path.name}: expected {expected}, got {got}")
 
 
+def sha256_for_url(url: str) -> str | None:
+    """Return the registry's pinned sha256 for a download URL, if present."""
+    for tool in _registry().get("tools", {}).values():
+        for asset in tool.get("assets", {}).values():
+            if asset.get("url") == url:
+                pin = asset.get("sha256")
+                return pin if isinstance(pin, str) else None
+    return None
+
+
+def verify_download_bytes(data: bytes, *, url: str, name: str) -> None:
+    """Fail-closed integrity check for an in-memory downloaded archive.
+
+    Used by installers (rtk, lean-ctx, codebase-memory-mcp) that download and
+    extract on their own instead of going through the fetch path above. Verifies
+    the bytes against the tools.json pin for ``url`` and refuses an unpinned URL
+    unless HEADROOM_BINARIES_ALLOW_UNVERIFIED=1.
+    """
+    if os.environ.get("HEADROOM_BINARIES_ALLOW_UNVERIFIED"):
+        logger.warning(
+            "skipping sha256 verification for %s (HEADROOM_BINARIES_ALLOW_UNVERIFIED=1)", name
+        )
+        return
+    expected = sha256_for_url(url)
+    if not expected:
+        # Off-registry URL (e.g. a version override); shipped assets are all
+        # pinned via the CI gate, so fall back to HTTPS trust here.
+        logger.info("%s downloaded without sha256 pin (HTTPS trust only)", name)
+        return
+    got = hashlib.sha256(data).hexdigest()
+    if got.lower() != expected.lower():
+        raise Sha256Mismatch(f"sha256 mismatch for {name}: expected {expected}, got {got}")
+
+
 # ---------- Archive extraction ------------------------------------------- #
 
 

headroom/graph/installer.py

@@ -77,13 +77,17 @@ def download_cbm(version: str | None = None) -> Path:
     except Exception as e:
         raise RuntimeError(f"Failed to download codebase-memory-mcp from {url}: {e}") from e
 
+    from headroom.binaries import verify_download_bytes
+
+    verify_download_bytes(data, url=url, name="codebase-memory-mcp")
+
     # Extract binary from tar.gz
     try:
         with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
             for member in tar.getmembers():
                 if member.name.endswith(CBM_BIN_NAME) or member.name == CBM_BIN_NAME:
                     member.name = target_path.name
-                    tar.extract(member, CBM_BIN_DIR)
+                    tar.extract(member, CBM_BIN_DIR, filter="data")
                     break
             else:
                 raise RuntimeError("codebase-memory-mcp binary not found in archive")

headroom/memory/backends/direct_mem0.py

@@ -51,6 +51,7 @@ import asyncio
 import hashlib
 import inspect
 import logging
+import os
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -100,7 +101,7 @@ class Mem0Config:
     # Neo4j settings
     neo4j_uri: str = "neo4j://localhost:7687"
     neo4j_user: str = "neo4j"
-    neo4j_password: str = "password"
+    neo4j_password: str = field(default_factory=lambda: os.environ.get("NEO4J_PASSWORD", ""))
 
     # Qdrant settings (defaults resolve from HEADROOM_QDRANT_* env vars)
     qdrant_url: str | None = field(default_factory=qdrant_env.qdrant_env_url)

headroom/memory/backends/mem0.py

@@ -12,6 +12,7 @@ Supports both local mode (embedded services) and cloud mode (Mem0 API).
 from __future__ import annotations
 
 import asyncio
+import os
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -57,7 +58,7 @@ class Mem0Config:
     # Local mode settings - Neo4j and Qdrant config
     neo4j_uri: str = "neo4j://localhost:7687"
     neo4j_user: str = "neo4j"
-    neo4j_password: str = "password"
+    neo4j_password: str = field(default_factory=lambda: os.environ.get("NEO4J_PASSWORD", ""))
     # Qdrant settings (defaults resolve from HEADROOM_QDRANT_* env vars)
     qdrant_url: str | None = field(default_factory=qdrant_env.qdrant_env_url)
     qdrant_host: str = field(default_factory=qdrant_env.qdrant_env_host)

headroom/memory/easy.py

@@ -33,6 +33,7 @@ Backends:
 
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -115,7 +116,7 @@ class Memory:
         qdrant_api_key: str | None = None,
         neo4j_uri: str = "neo4j://localhost:7687",
         neo4j_user: str = "neo4j",
-        neo4j_password: str = "password",
+        neo4j_password: str | None = None,
     ) -> None:
         from headroom.memory import qdrant_env
 
@@ -145,7 +146,9 @@ class Memory:
         )
         self._neo4j_uri = neo4j_uri
         self._neo4j_user = neo4j_user
-        self._neo4j_password = neo4j_password
+        self._neo4j_password = (
+            neo4j_password if neo4j_password is not None else os.environ.get("NEO4J_PASSWORD", "")
+        )
 
     async def _ensure_initialized(self) -> None:
         """Initialize the backend on first use."""

headroom/providers/proxy_routes.py

@@ -6,7 +6,7 @@ from __future__ import annotations
 import logging
 from typing import Any
 
-from fastapi import FastAPI, Request, WebSocket
+from fastapi import FastAPI, HTTPException, Request, WebSocket
 from fastapi.responses import Response
 
 from headroom.providers.cloudcode import normalize_cloudcode_passthrough_path
@@ -67,6 +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
 
 logger = logging.getLogger("headroom.proxy.routes")
 
@@ -266,6 +267,9 @@ 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):
+                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(
                 request, upstream_base_url=custom_base.rstrip("/")
             )
@@ -506,6 +510,9 @@ 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):
+                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("/")
             endpoint_name, provider_name = _custom_base_passthrough_telemetry(
                 request.method,

headroom/proxy/handlers/anthropic.py

@@ -47,6 +47,7 @@ from headroom.proxy.helpers import (
     relocate_system_messages_to_top_level,
     sanitize_forwarded_response_headers,
 )
+from headroom.proxy.identity import resolve_memory_identity
 from headroom.proxy.image_isolation import run_image_compression_isolated
 from headroom.proxy.memory_decision import MemoryDecision
 from headroom.proxy.memory_query import MemoryQuery
@@ -331,7 +332,7 @@ class AnthropicHandlerMixin:
             ctx = _CtxFor(
                 headers=dict(request.headers),
                 system_prompt=_extract_sys_prompt(body),
-                base_user_id=request.headers.get("x-headroom-user-id", ""),
+                base_user_id=resolve_memory_identity(request, default=""),
                 project_root_override=None,
             )
             ident = ProjectResolver().resolve(ctx)
@@ -1223,10 +1224,7 @@ class AnthropicHandlerMixin:
             memory_user_id: str | None = None
             memory_request_ctx = None
             if self.memory_handler:
-                memory_user_id = request.headers.get(
-                    "x-headroom-user-id",
-                    os.environ.get("USER", os.environ.get("USERNAME", "default")),
-                )
+                memory_user_id = resolve_memory_identity(request)
                 # Per-project memory routing (GH #462). Build the context
                 # once here so save / search / inject all resolve against
                 # the same workspace. Tier order: explicit project-id /

headroom/proxy/handlers/gemini.py

@@ -8,7 +8,6 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-import os
 import time
 from typing import TYPE_CHECKING, Any
 
@@ -21,6 +20,7 @@ from headroom.copilot_auth import build_copilot_upstream_url
 from headroom.proxy.auth_mode import classify_client
 from headroom.proxy.compression_decision import CompressionDecision
 from headroom.proxy.helpers import COMPRESSION_TIMEOUT_SECONDS, extract_tags
+from headroom.proxy.identity import resolve_memory_identity
 from headroom.proxy.outcome import RequestOutcome
 from headroom.proxy.token_counting import gemini_output_tokens
 
@@ -342,10 +342,7 @@ class GeminiHandlerMixin:
         memory_user_id: str | None = None
         memory_request_ctx = None
         if self.memory_handler:
-            memory_user_id = request.headers.get(
-                "x-headroom-user-id",
-                os.environ.get("USER", os.environ.get("USERNAME", "default")),
-            )
+            memory_user_id = resolve_memory_identity(request)
             # Per-project memory routing (GH #462). Gemini's
             # ``systemInstruction`` field carries the system prompt;
             # ``extract_system_prompt`` doesn't know that shape, so we

headroom/proxy/handlers/openai.py

@@ -30,8 +30,10 @@ from headroom.proxy.helpers import (
     jitter_delay_ms,
     sanitize_forwarded_response_headers,
 )
+from headroom.proxy.identity import resolve_memory_identity
 from headroom.proxy.loopback_guard import is_loopback_host
 from headroom.proxy.stage_timer import StageTimer, emit_stage_timings_log
+from headroom.proxy.upstream_guard import is_safe_upstream_url
 from headroom.proxy.ws_headers import WS_HOP_BY_HOP_HEADERS
 from headroom.proxy.ws_session_registry import (
     TerminationCause,
@@ -382,6 +384,13 @@ def _resolve_openai_upstream_base(request_headers: dict[str, str]) -> str | None
 
     if urlparse(normalized).scheme not in {"http", "https"}:
         return None
+    if not is_safe_upstream_url(normalized):
+        # Client-supplied upstream resolves to a private/loopback/link-local or
+        # cloud-metadata address (SSRF). Ignore the override and fall back to the
+        # configured upstream; set HEADROOM_ALLOWED_BASE_URLS to permit specific
+        # internal endpoints.
+        logger.warning("ignoring unsafe x-headroom-base-url override: %r", raw_base_url)
+        return None
     return normalized
 
 
@@ -3160,10 +3169,7 @@ class OpenAIHandlerMixin:
         memory_user_id: str | None = None
         memory_request_ctx = None
         if self.memory_handler:
-            memory_user_id = request.headers.get(
-                "x-headroom-user-id",
-                os.environ.get("USER", os.environ.get("USERNAME", "default")),
-            )
+            memory_user_id = resolve_memory_identity(request)
             # Per-project memory routing (GH #462). Built once per request
             # so every save/search/inject resolves to the same workspace.
             from headroom.memory.storage_router import (
@@ -5197,10 +5203,7 @@ class OpenAIHandlerMixin:
         memory_user_id: str | None = None
         memory_request_ctx = None
         if self.memory_handler:
-            memory_user_id = request.headers.get(
-                "x-headroom-user-id",
-                os.environ.get("USER", os.environ.get("USERNAME", "default")),
-            )
+            memory_user_id = resolve_memory_identity(request)
             from headroom.memory.storage_router import (
                 RequestContext as _MemRequestContext,
             )
@@ -6979,12 +6982,7 @@ class OpenAIHandlerMixin:
                 nonlocal memory_user_id, memory_request_ctx
 
                 memory_user_id_candidate = (
-                    ws_headers.get(
-                        "x-headroom-user-id",
-                        os.environ.get("USER", os.environ.get("USERNAME", "default")),
-                    )
-                    if self.memory_handler
-                    else None
+                    resolve_memory_identity(websocket) if self.memory_handler else None
                 )
                 memory_decision = MemoryDecision.decide(
                     headers=ws_headers,

headroom/proxy/identity.py

@@ -0,0 +1,92 @@
+"""Memory partition identity resolution (WEB-02).
+
+``x-headroom-user-id`` is a partition *hint*, not an authenticated identity. In
+the OSS proxy it is honored only for loopback callers (the single-user local
+model). For other callers the identity is bound to the proxy token (or the
+server's OS user) so a network client cannot select another user's memory.
+
+Multi-tenant deployments (e.g. headroom-managed) replace the default with an
+authenticated resolver via :func:`set_identity_resolver`, typically from a
+``headroom.proxy_extension`` install hook. This keeps real per-tenant identity —
+the enterprise differentiator — out of the OSS proxy while giving it a clean,
+secure single-user default.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+from typing import Any, Protocol
+
+from headroom.proxy.loopback_guard import is_loopback_host
+
+USER_ID_HEADER = "x-headroom-user-id"
+
+
+class IdentityResolver(Protocol):
+    def __call__(self, request: Any, *, default: str) -> str: ...
+
+
+_resolver: IdentityResolver | None = None
+
+
+def set_identity_resolver(resolver: IdentityResolver | None) -> None:
+    """Install (or clear) a custom identity resolver — the enterprise hook."""
+    global _resolver
+    _resolver = resolver
+
+
+def _default_os_user() -> str:
+    return os.environ.get("USER", os.environ.get("USERNAME", "default"))
+
+
+def _client_host(request: Any) -> str | None:
+    client = getattr(request, "client", None)
+    host = getattr(client, "host", None) if client is not None else None
+    return host if isinstance(host, str) else None
+
+
+def _token_identity() -> str | None:
+    token = os.environ.get("HEADROOM_PROXY_TOKEN")
+    if not token:
+        return None
+    return "tok_" + hashlib.sha256(token.encode()).hexdigest()[:16]
+
+
+def resolve_memory_identity(request: Any, *, default: str | None = None) -> str:
+    """Resolve the memory partition id for a request.
+
+    A registered custom resolver wins. Otherwise the header is honored only for
+    loopback callers; every other caller is bound to the proxy token (or the OS
+    user), so it can never address another user's partition.
+    """
+    fallback = default if default is not None else _default_os_user()
+
+    if _resolver is not None:
+        return _resolver(request, default=fallback)
+
+    header_value: str | None
+    try:
+        header_value = request.headers.get(USER_ID_HEADER)
+    except Exception:
+        header_value = None
+    if header_value is not None:
+        header_value = header_value.strip() or None
+
+    host = _client_host(request)
+    # Missing peer metadata is not evidence of loopback. Fail closed so unusual
+    # ASGI transports or incomplete request doubles cannot opt into header trust.
+    is_local = host is not None and is_loopback_host(host)
+
+    if header_value is not None:
+        if is_local:
+            return header_value
+        # A caller-supplied value cannot authenticate its own authority to select
+        # that partition. Remote multi-tenant selection belongs in the custom
+        # resolver hook, where it can be bound to authenticated caller context.
+
+    if not is_local:
+        token_id = _token_identity()
+        if token_id is not None:
+            return token_id
+    return fallback

headroom/proxy/memory_handler.py

@@ -154,7 +154,7 @@ class MemoryConfig:
     qdrant_api_key: str | None = field(default_factory=qdrant_env.qdrant_env_api_key)
     neo4j_uri: str = "neo4j://localhost:7687"
     neo4j_user: str = "neo4j"
-    neo4j_password: str = "password"
+    neo4j_password: str = field(default_factory=lambda: os.environ.get("NEO4J_PASSWORD", ""))
     # Memory Bridge (bidirectional markdown <-> Headroom sync)
     bridge_enabled: bool = False
     bridge_md_paths: list[str] = field(default_factory=list)

headroom/proxy/upstream_guard.py

@@ -0,0 +1,110 @@
+"""SSRF guard for client-supplied upstream base URLs (WEB-01).
+
+Clients may redirect the proxy's upstream via the ``x-headroom-base-url`` header
+(BYOK / custom OpenAI-compatible endpoints). Without validation this lets a
+caller turn the proxy into a confused deputy — reaching cloud-metadata
+(``169.254.169.254``) or internal RFC1918 hosts the caller cannot reach directly.
+
+Policy:
+  * Default: reject destinations that resolve to private, loopback, link-local,
+    or otherwise non-public addresses. Public hosts (api.openai.com, api.x.ai,
+    Azure, ...) are allowed so ordinary BYOK keeps working.
+  * When ``HEADROOM_ALLOWED_BASE_URLS`` is set (comma-separated hosts or URLs),
+    bare hosts permit every safe scheme/port for that host, while URLs permit
+    only their exact normalized origin. Because that is an explicit operator
+    choice, allowlisted destinations may point at internal/on-prem endpoints.
+
+This module intentionally depends only on the standard library so it is safe to
+import from any handler without risking an import cycle.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import os
+import socket
+from urllib.parse import urlparse
+
+ALLOWED_BASE_URLS_ENV = "HEADROOM_ALLOWED_BASE_URLS"
+
+_SAFE_SCHEMES = {"http", "https", "ws", "wss"}
+
+
+def _allowlisted_destinations() -> tuple[set[str], set[tuple[str, str, int]]] | None:
+    raw = os.environ.get(ALLOWED_BASE_URLS_ENV)
+    if not raw or not raw.strip():
+        return None
+    hosts: set[str] = set()
+    origins: set[tuple[str, str, int]] = set()
+    for item in raw.split(","):
+        item = item.strip()
+        if not item:
+            continue
+        if "://" not in item:
+            parsed = urlparse(f"//{item}")
+            if parsed.hostname:
+                hosts.add(parsed.hostname.lower())
+            continue
+        parsed = urlparse(item)
+        if parsed.scheme.lower() not in _SAFE_SCHEMES or not parsed.hostname:
+            continue
+        try:
+            port = parsed.port
+        except ValueError:
+            continue
+        if port is None:
+            port = 443 if parsed.scheme.lower() in {"https", "wss"} else 80
+        origins.add((parsed.scheme.lower(), parsed.hostname.lower(), port))
+    return hosts, origins
+
+
+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 (
+        addr.is_private
+        or addr.is_loopback
+        or addr.is_link_local
+        or addr.is_reserved
+        or addr.is_multicast
+        or addr.is_unspecified
+    )
+
+
+def is_safe_upstream_url(url: str) -> bool:
+    """Return True if ``url`` is a safe client-chosen upstream destination.
+
+    In allowlist mode only allowlisted hosts pass. Otherwise the host is
+    resolved and rejected if any resolved address is internal/metadata, which
+    also catches DNS names that point at private space.
+    """
+    parsed = urlparse((url or "").strip())
+    if parsed.scheme.lower() not in _SAFE_SCHEMES:
+        return False
+    host = parsed.hostname
+    if not host:
+        return False
+
+    allow = _allowlisted_destinations()
+    if allow is not None:
+        hosts, origins = allow
+        if host.lower() in hosts:
+            return True
+        try:
+            port = parsed.port
+        except ValueError:
+            return False
+        if port is None:
+            port = 443 if parsed.scheme.lower() in {"https", "wss"} else 80
+        return (parsed.scheme.lower(), host.lower(), port) in origins
+
+    try:
+        infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
+    except OSError:
+        # Resolution and connection are separate operations, so allowing a DNS
+        # miss here would fail open if the name resolves on the later lookup.
+        # Operators can explicitly allowlist split-horizon/internal endpoints.
+        return False
+    return all(not _is_internal_address(str(info[4][0])) for info in infos)

headroom/tools.json

@@ -1,5 +1,5 @@
 {
-  "_comment": "Registry of externally fetched CLI tool binaries. Bump versions and SHA256s via the weekly tools-version-check CI job (see .github/workflows/). sha256=null means HTTPS-trust-only (initial bootstrap); the CI job fills real SHAs per release.",
+  "_comment": "Registry of externally fetched CLI tool binaries. SHA-256 pins are enforced before execution (headroom/binaries.py). After bumping a version, regenerate pins with scripts/refresh_tool_hashes.py; the tools-hash-refresh CI workflow verifies they match the published assets.",
   "tools": {
     "difft": {
       "version": "0.64.0",
@@ -11,39 +11,39 @@
         "linux-x86_64-gnu": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-x86_64-unknown-linux-gnu.tar.gz",
           "member": "difft",
-          "sha256": null
+          "sha256": "9ec3aa9f784a54c8099d1af71b6bc18d3ad12ce50055565ffa2979aff93c5732"
         },
         "linux-x86_64-musl": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-x86_64-unknown-linux-gnu.tar.gz",
           "member": "difft",
-          "sha256": null,
+          "sha256": "9ec3aa9f784a54c8099d1af71b6bc18d3ad12ce50055565ffa2979aff93c5732",
           "_comment": "same asset as linux-x86_64-gnu; upstream has no dedicated musl build"
         },
         "linux-aarch64-gnu": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-aarch64-unknown-linux-gnu.tar.gz",
           "member": "difft",
-          "sha256": null
+          "sha256": "f657b0cc1baba3b5bcde4fdfbafbc58f2131476ed53d01e4880898d34a1cb124"
         },
         "linux-aarch64-musl": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-aarch64-unknown-linux-gnu.tar.gz",
           "member": "difft",
-          "sha256": null,
+          "sha256": "f657b0cc1baba3b5bcde4fdfbafbc58f2131476ed53d01e4880898d34a1cb124",
           "_comment": "same asset as linux-aarch64-gnu"
         },
         "darwin-x86_64": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-x86_64-apple-darwin.tar.gz",
           "member": "difft",
-          "sha256": null
+          "sha256": "d6b7f1c4a66495400f1f6e224efdaf82aa66e8930fcb257bc9357d69d52dd69d"
         },
         "darwin-aarch64": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-aarch64-apple-darwin.tar.gz",
           "member": "difft",
-          "sha256": null
+          "sha256": "e1e59ff3cf6c0837c94ddd4910d43351ec14f8c169abcfa51e63136d14c9e528"
         },
         "windows-x86_64": {
           "url": "https://github.com/Wilfred/difftastic/releases/download/0.64.0/difft-x86_64-pc-windows-msvc.zip",
           "member": "difft.exe",
-          "sha256": null
+          "sha256": "1ac0113adf9ade9417ee133bed38efe3085b7ccbb32ab620bdc36c4a9b365276"
         }
       }
     },
@@ -52,42 +52,42 @@
       "binary": "scc",
       "source": "boyter/scc",
       "homepage": "https://github.com/boyter/scc",
-      "_comment": "scc is a statically-linked Go binary, so the same asset is used for both libc variants. The duplicated gnu/musl entries are intentional — they document musl support for `headroom tools doctor`.",
+      "_comment": "scc is a statically-linked Go binary, so the same asset is used for both libc variants. The duplicated gnu/musl entries are intentional \u2014 they document musl support for `headroom tools doctor`.",
       "assets": {
         "linux-x86_64-gnu": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Linux_x86_64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "6c31f4d0cf3b7a8c5ca910fa4e451949434798f6541ec5dea4b83f4973e13772"
         },
         "linux-x86_64-musl": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Linux_x86_64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "6c31f4d0cf3b7a8c5ca910fa4e451949434798f6541ec5dea4b83f4973e13772"
         },
         "linux-aarch64-gnu": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Linux_arm64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "64446b1ca954aa1ac34984bbb4f098f46e6c69c84d64a7d096275ea6e50461eb"
         },
         "linux-aarch64-musl": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Linux_arm64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "64446b1ca954aa1ac34984bbb4f098f46e6c69c84d64a7d096275ea6e50461eb"
         },
         "darwin-x86_64": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Darwin_x86_64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "33fee1db983a6d22297d5f4e41bca25438c8a09d5c7c558acbfb5f9cd4deb19b"
         },
         "darwin-aarch64": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Darwin_arm64.tar.gz",
           "member": "scc",
-          "sha256": null
+          "sha256": "0e967d77ad564fa77bc5ed157d4e04bc6b56c5619edef317e5c99b15c9fca9d2"
         },
         "windows-x86_64": {
           "url": "https://github.com/boyter/scc/releases/download/v3.5.0/scc_Windows_x86_64.zip",
           "member": "scc.exe",
-          "sha256": null
+          "sha256": "9ce15936440f1680bd133905c9f6c99a5d258bcea6f73fc372c3f8b4b888200f"
         }
       }
     },
@@ -98,6 +98,34 @@
       "homepage": "https://ast-grep.github.io/",
       "_comment": "Installed via the ast-grep-cli PyPI wheel; we never fetch from GitHub for this tool. Listed here so `headroom tools doctor` can report it.",
       "assets": {}
+    },
+    "codebase-memory-mcp": {
+      "version": "v0.8.1",
+      "binary": "codebase-memory-mcp",
+      "source": "DeusData/codebase-memory-mcp",
+      "_comment": "No Windows build published for v0.8.1.",
+      "assets": {
+        "darwin-arm64": {
+          "url": "https://github.com/DeusData/codebase-memory-mcp/releases/download/v0.8.1/codebase-memory-mcp-darwin-arm64.tar.gz",
+          "member": "codebase-memory-mcp",
+          "sha256": "fbd047509852021b5446a11141bcb0a3d1dcaebf6e5112460960f29f052c1c58"
+        },
+        "darwin-amd64": {
+          "url": "https://github.com/DeusData/codebase-memory-mcp/releases/download/v0.8.1/codebase-memory-mcp-darwin-amd64.tar.gz",
+          "member": "codebase-memory-mcp",
+          "sha256": "fb62da3016ea12b948351208759b5c083fb1446cf6e78d6db8b7cd28fe86fd54"
+        },
+        "linux-arm64": {
+          "url": "https://github.com/DeusData/codebase-memory-mcp/releases/download/v0.8.1/codebase-memory-mcp-linux-arm64.tar.gz",
+          "member": "codebase-memory-mcp",
+          "sha256": "d2f842d1365da5c35d9c5796f57a821c9745267350994346735e1e6e04d46091"
+        },
+        "linux-amd64": {
+          "url": "https://github.com/DeusData/codebase-memory-mcp/releases/download/v0.8.1/codebase-memory-mcp-linux-amd64.tar.gz",
+          "member": "codebase-memory-mcp",
+          "sha256": "dbd3b92ea870ef240b63059f26bda15015f76ef9978931bebc3a0f9d09470973"
+        }
+      }
     }
   }
 }

scripts/refresh_tool_hashes.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Refresh SHA-256 pins for externally fetched tool binaries (WEB-03).
+
+Fetches every asset URL in ``headroom/tools.json``, computes its SHA-256, and
+writes the digests back into the registry. Run it locally after bumping a tool
+version, or let the ``tools-hash-refresh`` CI workflow run it.
+
+    python scripts/refresh_tool_hashes.py            # populate/update pins
+    python scripts/refresh_tool_hashes.py --check     # exit 1 if any pin drifts
+
+Only ``https://`` URLs are accepted; a plaintext URL is a hard error.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+import urllib.request
+from pathlib import Path
+
+REGISTRY = Path(__file__).resolve().parent.parent / "headroom" / "tools.json"
+
+
+def _fetch_sha256(url: str) -> str:
+    req = urllib.request.Request(url, headers={"User-Agent": "headroom-tools-refresh/1"})
+    digest = hashlib.sha256()
+    with urllib.request.urlopen(req, timeout=120) as resp:  # noqa: S310 - https enforced below
+        for chunk in iter(lambda: resp.read(1024 * 64), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--check", action="store_true", help="fail if any pin is missing/stale")
+    args = parser.parse_args()
+
+    data = json.loads(REGISTRY.read_text())
+    seen: dict[str, str] = {}
+    drift: list[str] = []
+
+    for tool_name, tool in data.get("tools", {}).items():
+        for platform, asset in tool.get("assets", {}).items():
+            url = asset.get("url")
+            if not url:
+                continue
+            if not url.startswith("https://"):
+                print(f"ERROR: {tool_name}/{platform}: non-https url {url!r}", file=sys.stderr)
+                return 2
+            if url not in seen:
+                print(f"fetching {tool_name}/{platform} …", file=sys.stderr)
+                seen[url] = _fetch_sha256(url)
+            digest = seen[url]
+            if asset.get("sha256") != digest:
+                drift.append(f"{tool_name}/{platform}")
+                if not args.check:
+                    asset["sha256"] = digest
+
+    if args.check:
+        if drift:
+            print("stale pins: " + ", ".join(drift), file=sys.stderr)
+            return 1
+        print("all tool pins up to date")
+        return 0
+
+    REGISTRY.write_text(json.dumps(data, indent=2) + "\n")
+    print(f"updated {len(drift)} pin(s) in {REGISTRY}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/conftest.py

@@ -154,6 +154,38 @@ def pytest_runtest_call(item):
             pytest.skip(reason)
 
 
+@pytest.fixture(autouse=True)
+def _null_binary_pins():
+    """Null the tools.json SHA-256 pins during tests.
+
+    Installer tests fetch small mock archives, whose digests can't match the
+    real published pins. Nulling the pins lets those download/extract mechanics
+    tests run (verification then falls back to HTTPS trust); the tests that
+    specifically exercise verification set their own pin explicitly. Production
+    keeps the real pins (this fixture is test-only) and the tools-hash-refresh
+    CI gate guarantees they stay correct.
+    """
+    try:
+        from headroom import binaries
+    except Exception:
+        # Lean CI environments (e.g. the native-installer jobs) omit heavy deps
+        # such as opentelemetry that importing `binaries` pulls in. There are no
+        # tool pins to null there, so skip cleanly rather than erroring at setup.
+        yield
+        return
+
+    saved = [
+        (asset, asset.get("sha256"))
+        for tool in binaries._registry().get("tools", {}).values()
+        for asset in tool.get("assets", {}).values()
+    ]
+    for asset, _original in saved:
+        asset["sha256"] = None
+    yield
+    for asset, original in saved:
+        asset["sha256"] = original
+
+
 @pytest.fixture(autouse=True)
 def _reset_headroom_logger_propagation():
     """Keep `headroom.*` log records flowing to pytest's caplog handler.

tests/test_binaries.py

@@ -23,6 +23,12 @@ def _clear_caches(monkeypatch, tmp_path):
     """Isolate every test from global state: cache dir, platform lru_cache, env."""
     binaries.detect_platform.cache_clear()
     binaries._registry.cache_clear()
+    # Null the shipped SHA-256 pins so mechanics tests can serve small mock
+    # archives (whose digests won't match production pins); the verification
+    # tests below set their own pin explicitly.
+    for _tool in binaries._registry().get("tools", {}).values():
+        for _asset in _tool.get("assets", {}).values():
+            _asset["sha256"] = None
     monkeypatch.setenv("HEADROOM_BINARIES_CACHE", str(tmp_path / "cache"))
     monkeypatch.delenv("HEADROOM_BINARIES_MIRROR", raising=False)
     monkeypatch.delenv("HEADROOM_BINARIES_OFFLINE", raising=False)

tests/test_custom_base_passthrough_telemetry.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from typing import Any
 
 import httpx
+import pytest
 from fastapi import FastAPI
 from fastapi.responses import JSONResponse
 from fastapi.testclient import TestClient
@@ -14,6 +15,12 @@ from headroom.providers.proxy_routes import register_provider_routes
 from headroom.proxy.handlers.openai import OpenAIHandlerMixin
 
 
+@pytest.fixture(autouse=True)
+def _allow_reserved_test_upstream(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Permit the reserved, intentionally unresolvable test origin."""
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "custom.example,opencode.ai,www.opencode.ai")
+
+
 class _Runtime:
     @staticmethod
     def api_target(provider: str) -> str:

tests/test_header_isolation.py

@@ -38,6 +38,13 @@ from headroom.proxy.helpers import (
 )
 from headroom.proxy.server import ProxyConfig, create_app
 
+
+@pytest.fixture(autouse=True)
+def _allow_reserved_test_upstream(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Permit the reserved override used by the end-to-end isolation test."""
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "override.example")
+
+
 # ---------------------------------------------------------------------------
 # Pure helper unit tests
 # ---------------------------------------------------------------------------

tests/test_identity_resolution.py

@@ -0,0 +1,68 @@
+"""Tests for memory-partition identity resolution (WEB-02)."""
+
+from __future__ import annotations
+
+import pytest
+
+from headroom.proxy import identity
+from headroom.proxy.identity import resolve_memory_identity, set_identity_resolver
+
+
+class _FakeRequest:
+    def __init__(self, headers: dict[str, str], host: str | None) -> None:
+        self.headers = headers
+        self.client = type("_Client", (), {"host": host})() if host is not None else None
+
+
+@pytest.fixture(autouse=True)
+def _clean_env(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delenv("HEADROOM_PROXY_TOKEN", raising=False)
+    set_identity_resolver(None)
+    yield
+    set_identity_resolver(None)
+
+
+def test_loopback_trusts_header() -> None:
+    req = _FakeRequest({"x-headroom-user-id": "alice"}, "127.0.0.1")
+    assert resolve_memory_identity(req) == "alice"
+
+
+def test_non_loopback_ignores_header() -> None:
+    req = _FakeRequest({"x-headroom-user-id": "victim@corp"}, "10.0.0.5")
+    assert resolve_memory_identity(req, default="me") == "me"
+
+
+def test_missing_peer_metadata_does_not_trust_header() -> None:
+    req = _FakeRequest({"x-headroom-user-id": "victim@corp"}, None)
+    assert resolve_memory_identity(req, default="me") == "me"
+
+
+def test_non_loopback_binds_to_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HEADROOM_PROXY_TOKEN", "s3cret")
+    req = _FakeRequest({"x-headroom-user-id": "victim@corp"}, "10.0.0.5")
+    got = resolve_memory_identity(req, default="me")
+    assert got.startswith("tok_")
+    assert got not in {"victim@corp", "me"}
+
+
+def test_legacy_user_id_allowlist_cannot_authorize_remote_claim(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("HEADROOM_USER_ID_ALLOWLIST", "alice,bob")
+    assert (
+        resolve_memory_identity(
+            _FakeRequest({"x-headroom-user-id": "alice"}, "10.0.0.5"), default="me"
+        )
+        == "me"
+    )
+
+
+def test_custom_resolver_wins() -> None:
+    set_identity_resolver(lambda request, *, default: "tenant-42")
+    req = _FakeRequest({"x-headroom-user-id": "whatever"}, "10.0.0.5")
+    assert resolve_memory_identity(req) == "tenant-42"
+
+
+def test_no_header_uses_default() -> None:
+    assert resolve_memory_identity(_FakeRequest({}, "127.0.0.1"), default="") == ""
+    assert identity._default_os_user()  # never empty

tests/test_memory_handler_native_ops.py

@@ -1357,7 +1357,7 @@ async def test_ensure_initialized_fast_paths_and_qdrant_variants(
         "qdrant_api_key": None,
         "neo4j_uri": "neo4j://localhost:7687",
         "neo4j_user": "neo4j",
-        "neo4j_password": "password",
+        "neo4j_password": "",
         "enable_graph": True,
     }
 

tests/test_provider_proxy_routes.py

@@ -30,6 +30,9 @@ def _app() -> Any:
 
 
 def test_provider_passthrough_routes_forward_expected_targets(monkeypatch) -> None:
+    # This routing test uses reserved, intentionally unresolvable hostnames.
+    # Explicitly allow them so the SSRF guard can remain fail-closed on DNS errors.
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "azure.example,custom.example,opencode.ai")
     calls: list[tuple[str, str, str, str]] = []
     gemini_calls: list[tuple[str, str, str, str]] = []
     gemini_count_calls: list[tuple[str, str, str, str]] = []

tests/test_proxy/test_openai_upstream_header.py

@@ -24,6 +24,12 @@ from starlette.datastructures import Headers  # noqa: E402
 from headroom.proxy.handlers.openai import OpenAIHandlerMixin  # noqa: E402
 
 
+@pytest.fixture(autouse=True)
+def _allow_reserved_test_upstream(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Permit the reserved, intentionally unresolvable test origin."""
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "gateway.example")
+
+
 class _FakeRequest:
     """Minimal stand-in exposing ``headers`` like a real Starlette request.
 

tests/test_upstream_credential_scoping.py

@@ -50,6 +50,12 @@ def _clear_warn_memo():
     reset_warning_state()
 
 
+@pytest.fixture(autouse=True)
+def _allow_reserved_test_upstreams(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Permit reserved test hosts while credential trust is tested separately."""
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "attacker.example,corp-gw.internal")
+
+
 class _Capturing(httpx.AsyncBaseTransport):
     def __init__(self) -> None:
         self.headers: dict[str, str] | None = None

tests/test_upstream_guard.py

@@ -0,0 +1,59 @@
+"""Tests for the SSRF upstream guard (WEB-01).
+
+All cases use IP literals or ``localhost`` so no external network is required.
+"""
+
+from __future__ import annotations
+
+import socket
+
+import pytest
+
+from headroom.proxy.upstream_guard import is_safe_upstream_url
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://169.254.169.254/latest/meta-data/",  # cloud metadata
+        "http://127.0.0.1:8080/admin",  # loopback
+        "http://10.0.0.1:8080/",  # RFC1918
+        "http://192.168.1.10/",  # RFC1918
+        "http://172.16.0.1/",  # RFC1918
+        "https://localhost/v1",  # resolves to loopback
+        "http://[::1]/",  # IPv6 loopback
+        "ftp://example.com/",  # non-http(s)/ws scheme
+        "not-a-url",
+        "",
+    ],
+)
+def test_blocks_internal_and_invalid(url: str) -> None:
+    assert is_safe_upstream_url(url) is False
+
+
+@pytest.mark.parametrize("url", ["https://8.8.8.8/v1", "https://1.1.1.1/", "wss://9.9.9.9/rt"])
+def test_allows_public(url: str) -> None:
+    assert is_safe_upstream_url(url) is True
+
+
+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")
+
+    monkeypatch.setattr(socket, "getaddrinfo", fail_resolution)
+    assert is_safe_upstream_url("https://temporarily-unresolved.example/v1") is False
+
+
+def test_allowlist_mode(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HEADROOM_ALLOWED_BASE_URLS", "api.internal.example, https://llm.corp:8443")
+    # Allowlisted hosts pass — including internal ones the operator opted into,
+    # without a DNS lookup.
+    assert is_safe_upstream_url("https://api.internal.example/v1") is True
+    assert is_safe_upstream_url("https://llm.corp:8443/v1") is True
+    # URL entries are exact origins, not implicit host-wide grants.
+    assert is_safe_upstream_url("https://llm.corp:22/v1") is False
+    assert is_safe_upstream_url("http://llm.corp:8443/v1") is False
+    assert is_safe_upstream_url("https://llm.corp/v1") is False
+    # 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