fix(proxy): validate Google API route hosts

headroom/proxy/handlers/anthropic.py

@@ -14,6 +14,7 @@ import time
 import uuid
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlsplit
 
 from headroom.proxy.stage_timer import StageTimer, emit_stage_timings_log
 
@@ -50,6 +51,23 @@ from headroom.proxy.outcome import RequestOutcome
 logger = logging.getLogger("headroom.proxy")
 
 
+def _is_googleapis_endpoint(value: object) -> bool:
+    """Return whether *value* targets Google APIs by parsed hostname.
+
+    A substring check would also trust attacker-controlled hosts such as
+    ``googleapis.com.example.test``. URL parsing plus a label-boundary suffix
+    check accepts Google API subdomains without widening the route gate.
+    """
+    raw = str(value).strip()
+    if not raw:
+        return False
+    try:
+        hostname = (urlsplit(raw).hostname or "").rstrip(".").lower()
+    except ValueError:
+        return False
+    return hostname == "googleapis.com" or hostname.endswith(".googleapis.com")
+
+
 class _AnthropicTurnHookUsage:
     """Usage from hook-triggered Anthropic calls the main response omits.
 
@@ -3020,7 +3038,7 @@ class AnthropicHandlerMixin:
                         if (
                             not upstream_base_url
                             or getattr(self, "anthropic_backend", None) is not None
-                            or "googleapis.com" in str(upstream_base_url)
+                            or _is_googleapis_endpoint(upstream_base_url)
                         )
                         else None
                     ),

tests/test_proxy_handler_helpers.py

@@ -11,7 +11,7 @@ import httpx
 import pytest
 from fastapi.responses import StreamingResponse
 
-from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
+from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin, _is_googleapis_endpoint
 from headroom.proxy.handlers.openai import (
     OpenAIHandlerMixin,
     _decode_openai_bearer_payload,
@@ -35,6 +35,23 @@ def _jwt(payload: object) -> str:
     return f"{encode(header)}.{encode(payload)}."
 
 
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("https://us-central1-aiplatform.googleapis.com/v1", True),
+        ("https://googleapis.com/v1", True),
+        ("https://AIPLATFORM.GOOGLEAPIS.COM./v1", True),
+        ("https://googleapis.com.example.test/v1", False),
+        ("https://notgoogleapis.com/v1", False),
+        ("https://googleapis.com@attacker.test/v1", False),
+        ("not a url", False),
+        ("", False),
+    ],
+)
+def test_googleapis_endpoint_gate_uses_hostname_boundary(url: str, expected: bool) -> None:
+    assert _is_googleapis_endpoint(url) is expected
+
+
 class _ImageCompressor:
     def __init__(self, compressed_message):
         self._compressed_message = compressed_message