chore(proxy): resolve request route from ASGI scope

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -13,6 +13,7 @@
     SpecialHeaders,
     UserAPIKeyAuth,
 )
+from litellm.proxy.auth.auth_utils import get_request_route
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 
 
@@ -119,14 +120,13 @@ async def mock_body():
 
         request.body = mock_body  # type: ignore
         # Only OAuth metadata routes registered under /.well-known/ are public.
-        # Match on request.url.path (path-only, exact prefix) so the substring
-        # cannot be smuggled via query string, hostname, or a deeper URL segment.
-        if request.url.path.startswith("/.well-known/"):
+        request_path = get_request_route(request)
+        if request_path.startswith("/.well-known/"):
             validated_user_api_key_auth = UserAPIKeyAuth()
         elif (
             not litellm_api_key
             and MCPRequestHandler._target_servers_delegate_auth_to_upstream(  # noqa: E501
-                path=request.url.path, mcp_servers=mcp_servers
+                path=request_path, mcp_servers=mcp_servers
             )
         ):
             # Operator opted this oauth2 server into upstream-delegated auth
@@ -174,7 +174,7 @@ async def mock_body():
                     "401",
                     "403",
                 ) and MCPRequestHandler._target_servers_use_oauth2(
-                    path=request.url.path, mcp_servers=mcp_servers
+                    path=request_path, mcp_servers=mcp_servers
                 ):
                     verbose_logger.debug(
                         "MCP OAuth2: target server is OAuth2-mode, treating "

litellm/proxy/auth/auth_utils.py

@@ -486,21 +486,29 @@ def get_request_route(request: Request) -> str:
     """
     Helper to get the route from the request
 
-    remove base url from path if set e.g. `/genai/chat/completions` -> `/chat/completions
+    remove base url from path if set e.g. `/genai/chat/completions` -> `/chat/completions`
+
+    Reads ``request.scope["path"]`` (the authoritative ASGI path that
+    FastAPI uses for routing) rather than ``request.url.path``. Starlette
+    constructs ``request.url`` by interpolating the ``Host`` header into
+    a URL string and re-parsing with ``urlsplit``, so a Host containing
+    ``/?`` or ``/#`` collapses ``url.path`` to ``"/"`` — which is in
+    ``LiteLLMRoutes.public_routes`` and would skip auth.
     """
+    scope = getattr(request, "scope", None)
+    if isinstance(scope, dict):
+        path = scope.get("path") or ""
+        root_path = scope.get("root_path") or ""
+        if root_path and path.startswith(root_path):
+            path = path[len(root_path) :]
+        return path or "/"
+    # Non-dict scope only arises in unit tests that mock ``Request``
+    # without populating scope. Production ASGI scope is always a dict,
+    # so the secure path above is the one that runs in practice.
     try:
-        if hasattr(request, "base_url") and request.url.path.startswith(
-            request.base_url.path
-        ):
-            # remove base_url from path
-            return request.url.path[len(request.base_url.path) - 1 :]
-        else:
-            return request.url.path
-    except Exception as e:
-        verbose_proxy_logger.debug(
-            f"error on get_request_route: {str(e)}, defaulting to request.url.path={request.url.path}"
-        )
         return request.url.path
+    except Exception:
+        return "/"
 
 
 @lru_cache(maxsize=256)

litellm/proxy/auth/route_checks.py

@@ -14,6 +14,7 @@
 )
 
 from .auth_checks_organization import _user_is_org_admin
+from .auth_utils import get_request_route
 
 # Management write routes denied to PROXY_ADMIN_VIEW_ONLY. Adding a new write
 # endpoint to a management router REQUIRES adding it here too — the surrounding
@@ -625,7 +626,8 @@ def _is_assistants_api_request(request: Request) -> bool:
         Returns:
             bool: True if `thread` or `assistant` is in the request path, False otherwise
         """
-        if "thread" in request.url.path or "assistant" in request.url.path:
+        path = get_request_route(request)
+        if "thread" in path or "assistant" in path:
             return True
         return False
 

litellm/proxy/auth/user_api_key_auth.py

@@ -332,8 +332,21 @@ def _apply_budget_limits_to_end_user_params(
 async def user_api_key_auth_websocket(websocket: WebSocket):
     # Accept the WebSocket connection
 
-    scope_headers = list(websocket.scope.get("headers") or [])
-    request = Request(scope={"type": "http", "headers": scope_headers})
+    ws_scope = websocket.scope or {}
+    # Carry the routing fields from the WebSocket scope into the
+    # synthetic HTTP request. ``get_request_route`` reads ``scope["path"]``;
+    # without these the auth gate would see an empty path and treat the
+    # connection as the public ``/`` route.
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": list(ws_scope.get("headers") or []),
+            "path": ws_scope.get("path") or "/",
+            "raw_path": ws_scope.get("raw_path") or b"",
+            "root_path": ws_scope.get("root_path") or "",
+            "query_string": ws_scope.get("query_string") or b"",
+        }
+    )
 
     request._url = websocket.url
 

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -52,6 +52,7 @@
 from litellm.proxy._experimental.mcp_server.utils import (
     validate_and_normalize_mcp_server_payload as _base_validate_and_normalize_mcp_server_payload,
 )
+from litellm.proxy.auth.auth_utils import get_request_route
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
     decrypt_value_helper,
     encrypt_value_helper,
@@ -1585,7 +1586,7 @@ async def _mcp_oauth_user_api_key_auth(request: Request) -> UserAPIKeyAuth:
                 ):
                     # For /token, require PKCE authorization_code; refresh_token
                     # grants must NOT bypass auth (see comment above).
-                    path_lower = (request.url.path or "").rstrip("/").lower()
+                    path_lower = get_request_route(request).rstrip("/").lower()
                     if path_lower.endswith("/token"):
                         body_data = await _read_request_body(request=request)
                         grant_type = (body_data or {}).get("grant_type", "")

litellm/proxy/vector_store_endpoints/utils.py

@@ -10,6 +10,7 @@
     LitellmUserRoles,
     UserAPIKeyAuth,
 )
+from litellm.proxy.auth.auth_utils import get_request_route
 from litellm.types.utils import LlmProviders
 from litellm.types.vector_stores import LiteLLM_ManagedVectorStore
 from litellm.utils import ProviderConfigManager
@@ -330,19 +331,21 @@ def is_allowed_to_call_vector_store_endpoint(
         provider_config.get_vector_store_endpoints_by_type()
     )
 
+    request_path = get_request_route(request)
+
     # Determine the permission type based on the request
     permission_type = None
     for endpoint in provider_vector_store_endpoints["read"]:
         if request.method == endpoint[0] and _does_endpoint_match(
-            endpoint[1], request.url.path
+            endpoint[1], request_path
         ):
             permission_type = "read"
             break
 
     if permission_type is None:
         for endpoint in provider_vector_store_endpoints["write"]:
             if request.method == endpoint[0] and _does_endpoint_match(
-                endpoint[1], request.url.path
+                endpoint[1], request_path
             ):
                 permission_type = "write"
                 break
@@ -392,18 +395,20 @@ def is_allowed_to_call_vector_store_files_endpoint(
         provider_config.get_vector_store_file_endpoints_by_type()
     )
 
+    request_path = get_request_route(request)
+
     permission_type: Optional[str] = None
     for endpoint in provider_vector_store_endpoints.get("read", ()):
         if request.method == endpoint[0] and _does_endpoint_match(
-            endpoint[1], request.url.path
+            endpoint[1], request_path
         ):
             permission_type = "read"
             break
 
     if permission_type is None:
         for endpoint in provider_vector_store_endpoints.get("write", ()):
             if request.method == endpoint[0] and _does_endpoint_match(
-                endpoint[1], request.url.path
+                endpoint[1], request_path
             ):
                 permission_type = "write"
                 break

tests/test_litellm/proxy/auth/test_request_route_resolution.py

@@ -0,0 +1,180 @@
+"""
+Regression tests: a malformed ``Host`` header must not influence the
+route the auth gate sees.
+
+``request.url.path`` in Starlette is constructed by interpolating the
+``Host`` header into a URL string and re-parsing with ``urlsplit``, so a
+Host containing ``/?`` or ``/#`` collapses ``url.path`` to ``"/"`` (the
+real path falls into the query/fragment). ``"/"`` is in
+``LiteLLMRoutes.public_routes``, so route-based auth gates would treat
+protected routes as public. ``get_request_route`` reads
+``scope["path"]`` instead — the authoritative ASGI path FastAPI uses
+for routing.
+"""
+
+import os
+import sys
+from unittest.mock import patch
+
+import pytest
+from fastapi.testclient import TestClient
+from starlette.requests import Request
+
+sys.path.insert(
+    0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../.."))
+)
+
+from litellm.proxy._types import LiteLLMRoutes  # noqa: E402
+from litellm.proxy.auth.auth_utils import get_request_route  # noqa: E402
+
+
+def _make_request(path: str, host_header: bytes, root_path: bytes = b"") -> Request:
+    """Build a minimal Starlette request with the chosen Host header."""
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "method": "GET",
+        "scheme": "http",
+        "server": ("localhost", 4000),
+        "client": ("127.0.0.1", 12345),
+        "root_path": root_path.decode(),
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": b"",
+        "headers": [(b"host", host_header)],
+    }
+    return Request(scope=scope)
+
+
+# All four payload variants from the advisory + extra shapes I confirmed
+# in variant analysis. ``user@`` and ``[::1]`` both collapse url.path to
+# "/" via the same urlsplit reparse mechanism.
+_BYPASS_HOST_HEADERS = [
+    b"localhost/?x=1",
+    b"localhost:4000/?x=1",
+    b"localhost/#test",
+    b"localhost:4000/#test",
+    b"user@localhost/?x=1",
+    b"[::1]/?x=1",
+    b"localhost\\/?x=1",
+]
+
+
+@pytest.mark.parametrize("host_header", _BYPASS_HOST_HEADERS)
+def test_get_request_route_ignores_host_header(host_header):
+    protected = "/get/internal_user_settings"
+    req = _make_request(protected, host_header)
+    assert get_request_route(req) == protected, (
+        f"Host header {host_header!r} corrupted the auth-time path: "
+        f"got {get_request_route(req)!r}, expected {protected!r}."
+    )
+
+
+def test_get_request_route_strips_root_path():
+    # Operators run the proxy mounted under a root_path (e.g. /genai). The
+    # auth gate has always compared against unprefixed paths; preserve that.
+    req = _make_request("/genai/chat/completions", b"localhost", root_path=b"/genai")
+    assert get_request_route(req) == "/chat/completions"
+
+
+def test_get_request_route_handles_root_path_with_bad_host():
+    # Combination: mounted root_path AND malicious Host. The fix must
+    # still resolve the unprefixed route correctly.
+    req = _make_request(
+        "/genai/get/internal_user_settings",
+        b"localhost/?x=1",
+        root_path=b"/genai",
+    )
+    assert get_request_route(req) == "/get/internal_user_settings"
+
+
+def test_slash_is_still_a_public_route():
+    # Sanity: ``/`` IS in public_routes — proving the bypass shape
+    # (corrupted path = "/") would have skipped auth before this fix.
+    assert "/" in LiteLLMRoutes.public_routes.value
+
+
+@pytest.fixture(scope="module")
+def proxy_client():
+    """One TestClient per module — TestClient construction triggers
+    FastAPI route-tree build + lifespan startup, both expensive. Env
+    vars are restored after the fixture exits so they don't bleed into
+    other test modules in the same worker."""
+    overrides = {
+        "DATABASE_URL": "",
+        "DISABLE_SCHEMA_UPDATE": "True",
+        "LITELLM_MASTER_KEY": "sk-1234",
+    }
+    saved = {k: os.environ.get(k) for k in overrides}
+    for k, v in overrides.items():
+        os.environ[k] = v
+    try:
+        import litellm.proxy.proxy_server as ps
+
+        with patch.object(ps, "master_key", "sk-1234"):
+            yield TestClient(ps.app)
+    finally:
+        for k, v in saved.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+
+
+@pytest.mark.parametrize("host_header", _BYPASS_HOST_HEADERS)
+def test_e2e_protected_admin_route_remains_401(proxy_client, host_header):
+    """End-to-end: full FastAPI app sees the malicious Host but the
+    auth gate still refuses the request because get_request_route now
+    returns the real path."""
+    r = proxy_client.get(
+        "/get/internal_user_settings",
+        headers={"Host": host_header.decode("latin-1")},
+    )
+    assert r.status_code == 401, (
+        f"Malformed Host={host_header!r} bypassed auth: returned "
+        f"{r.status_code} (body: {r.text[:200]})"
+    )
+
+
+def test_websocket_synthetic_request_carries_scope_path():
+    """``user_api_key_auth_websocket`` builds a synthetic HTTP Request
+    from the WebSocket scope. It must propagate ``path`` (and the
+    related routing fields) so the auth helpers downstream see the
+    real route — otherwise an empty path would be classified as the
+    public root and skip auth."""
+    from starlette.datastructures import URL
+    from unittest.mock import MagicMock
+    from fastapi import WebSocket
+
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket
+
+    captured: dict = {}
+
+    async def fake_user_api_key_auth(request, api_key):
+        captured["scope"] = dict(request.scope)
+        captured["route"] = get_request_route(request)
+        return None  # short-circuit downstream
+
+    ws = MagicMock(spec=WebSocket)
+    ws.scope = {
+        "type": "websocket",
+        "path": "/v1/realtime",
+        "raw_path": b"/v1/realtime",
+        "root_path": "",
+        "query_string": b"model=gpt-4o",
+        "headers": [(b"authorization", b"Bearer sk-x")],
+    }
+    ws.url = URL("ws://localhost/v1/realtime")
+    ws.query_params = {"model": "gpt-4o"}
+    ws.headers = {"authorization": "Bearer sk-x"}
+
+    with patch(
+        "litellm.proxy.auth.user_api_key_auth.user_api_key_auth",
+        new=fake_user_api_key_auth,
+    ):
+        import asyncio
+
+        asyncio.get_event_loop().run_until_complete(user_api_key_auth_websocket(ws))
+
+    assert captured["scope"]["path"] == "/v1/realtime"
+    assert captured["route"] == "/v1/realtime"