Fix BadHost path handling

backend/auth.py

@@ -113,8 +113,14 @@ def _scoped_admin_tokens() -> dict[str, list[str]]:
     return normalized
 
 
+def _request_scope_path(request: Request) -> str:
+    """Return the ASGI request-line path, not the Host-derived URL path."""
+    scope = getattr(request, "scope", {}) or {}
+    return str(scope.get("path") or "")
+
+
 def _required_scope_for_request(request: Request) -> str:
-    path = str(request.url.path or "")
+    path = _request_scope_path(request)
     if path.startswith("/api/wormhole/gate/"):
         return "gate"
     if path.startswith("/api/wormhole/dm/"):
@@ -443,7 +449,7 @@ async def _verify_openclaw_hmac(request: Request) -> bool:
 
     # Compute expected signature: HMAC-SHA256(secret, METHOD|path|ts|nonce|body_digest)
     method = str(request.method or "").upper()
-    path = str(request.url.path or "")
+    path = _request_scope_path(request)
     message = f"{method}|{path}|{ts_str}|{nonce}|{body_digest}"
     expected = hmac.new(
         secret.encode("utf-8"),
@@ -744,8 +750,7 @@ def _is_debug_test_request(request: Request) -> bool:
     if not _debug_mode_enabled():
         return False
     client_host = (request.client.host or "").lower() if request.client else ""
-    url_host = (request.url.hostname or "").lower() if request.url else ""
-    return client_host == "test" or url_host == "test"
+    return client_host == "test"
 
 
 # ---------------------------------------------------------------------------
@@ -1397,10 +1402,7 @@ def _peer_hmac_url_from_request(request: Request) -> str:
     header_url = normalize_peer_url(str(request.headers.get("x-peer-url", "") or ""))
     if header_url:
         return header_url
-    if not request.url:
-        return ""
-    base_url = f"{request.url.scheme}://{request.url.netloc}".rstrip("/")
-    return normalize_peer_url(base_url)
+    return ""
 
 
 def _verify_peer_push_hmac(request: Request, body_bytes: bytes) -> bool:

backend/main.py

@@ -310,6 +310,7 @@ from auth import (
     _private_plane_access_denied_payload,
     _private_infonet_policy_snapshot,
     _private_plane_refusal_response,
+    _request_scope_path,
     _scoped_admin_tokens,
     _scoped_view_authenticated as _scoped_view_authenticated_auth,
     _security_headers,
@@ -2728,7 +2729,7 @@ async def json_decode_error_handler(_request: Request, _exc: JSONDecodeError):
 
 @app.exception_handler(StarletteHTTPException)
 async def private_plane_http_exception_handler(request: Request, exc: StarletteHTTPException):
-    if exc.status_code == 403 and _is_private_plane_access_path(request.url.path, request.method):
+    if exc.status_code == 403 and _is_private_plane_access_path(_request_scope_path(request), request.method):
         return await _private_plane_refusal_response(
             request,
             status_code=403,
@@ -2762,7 +2763,7 @@ async def mesh_security_headers(request: Request, call_next):
 @app.middleware("http")
 async def mesh_no_store_headers(request: Request, call_next):
     response = await call_next(request)
-    if request.url.path.startswith("/api/mesh/"):
+    if _request_scope_path(request).startswith("/api/mesh/"):
         response.headers["Cache-Control"] = "no-store, max-age=0"
         response.headers["Pragma"] = "no-cache"
     return response
@@ -3464,7 +3465,7 @@ def _refresh_lookup_handle_rotation_background(*, reason: str) -> dict[str, Any]
 
 @app.middleware("http")
 async def enforce_high_privacy_mesh(request: Request, call_next):
-    path = request.url.path
+    path = _request_scope_path(request)
     private_mesh_path = path.startswith("/api/mesh") and not _is_public_meshtastic_lane_path(
         path,
         request.method,
@@ -3624,7 +3625,7 @@ async def enforce_high_privacy_mesh(request: Request, call_next):
 @app.middleware("http")
 async def apply_no_store_to_sensitive_paths(request: Request, call_next):
     response = await call_next(request)
-    if _is_sensitive_no_store_path(request.url.path):
+    if _is_sensitive_no_store_path(_request_scope_path(request)):
         for key, value in _NO_STORE_HEADERS.items():
             response.headers[key] = value
     return response

backend/pyproject.toml

@@ -15,7 +15,7 @@ dependencies = [
     "cachetools==5.5.2",
     "cryptography>=41.0.0",
     "defusedxml>=0.7.1",
-    "fastapi==0.115.12",
+    "fastapi==0.136.3",
     "feedparser==6.0.10",
     "httpx==0.28.1",
     "playwright==1.59.0",
@@ -33,6 +33,7 @@ dependencies = [
     "paho-mqtt>=1.6.0,<2.0.0",
     "PyNaCl>=1.5.0",
     "slowapi==0.1.9",
+    "starlette==1.0.1",
     "vaderSentiment>=3.3.0",
     "uvicorn==0.34.0",
     "yfinance==1.3.0",

backend/services/mesh/mesh_signed_events.py

@@ -38,6 +38,11 @@ _REVOCATION_TTL_CACHE: dict[str, dict[str, Any]] = {}
 _REVOCATION_TTL_LOCK = threading.Lock()
 _REVOCATION_REFRESH_LOCK = threading.Lock()
 _REVOCATION_REFRESH_FAIL_FAST_WINDOW_S = 5.0
+
+
+def _request_scope_path(request: Request) -> str:
+    scope = getattr(request, "scope", {}) or {}
+    return str(scope.get("path") or "")
 _REVOCATION_REFRESH_RETRY_AFTER_S = 5
 _REVOCATION_PRECHECK_UNAVAILABLE_DETAIL = "Signed event integrity preflight unavailable"
 
@@ -166,7 +171,7 @@ def _canonical_signed_write_retry_payload(
     signed_context = build_signed_context(
         event_type=prepared.event_type,
         kind=prepared.kind.value,
-        endpoint=str(request.url.path or ""),
+        endpoint=_request_scope_path(request),
         lane_floor=_content_private_required_transport_tier(prepared.kind),
         sequence_domain=_signed_context_sequence_domain(prepared),
         node_id=prepared.node_id,
@@ -540,7 +545,7 @@ def _apply_signed_context_policy(prepared: "PreparedSignedWrite", request: Reque
     ok, reason = validate_signed_context(
         event_type=prepared.event_type,
         kind=prepared.kind.value,
-        endpoint=str(request.url.path or ""),
+        endpoint=_request_scope_path(request),
         lane_floor=_content_private_required_transport_tier(prepared.kind),
         sequence_domain=_signed_context_sequence_domain(prepared),
         node_id=prepared.node_id,

backend/tests/test_badhost_hardening.py

@@ -0,0 +1,50 @@
+from starlette.requests import Request
+
+import auth
+
+
+async def _empty_receive():
+    return {"type": "http.request", "body": b"", "more_body": False}
+
+
+def _request(path: str, *, host: str = "example.com/health?x=", client_host: str = "203.0.113.10") -> Request:
+    return Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "scheme": "http",
+            "server": ("127.0.0.1", 8000),
+            "client": (client_host, 12345),
+            "path": path,
+            "raw_path": path.encode("ascii"),
+            "query_string": b"",
+            "headers": [(b"host", host.encode("ascii"))],
+        },
+        receive=_empty_receive,
+    )
+
+
+def test_scope_auth_uses_asgi_path_not_host_derived_url_path():
+    request = _request("/api/mesh/gate/alpha/message")
+
+    assert auth._request_scope_path(request) == "/api/mesh/gate/alpha/message"
+    assert auth._required_scope_for_request(request) == "mesh"
+
+
+def test_debug_test_request_does_not_trust_host_header(monkeypatch):
+    monkeypatch.setattr(auth, "_debug_mode_enabled", lambda: True)
+
+    request = _request("/api/admin", host="test/api/public?x=")
+
+    assert auth._is_debug_test_request(request) is False
+
+
+def test_peer_hmac_identity_requires_explicit_peer_url_header():
+    request = _request("/api/mesh/infonet/push", host="https://peer.example/api/public?x=")
+
+    assert auth._peer_hmac_url_from_request(request) == ""
+
+    request = _request("/api/mesh/infonet/push")
+    request.scope["headers"].append((b"x-peer-url", b"https://peer.example/"))
+
+    assert auth._peer_hmac_url_from_request(request) == "https://peer.example"