refactor(auth): read WebSocket keys with the shared Authorization parser

WebSocket auth had its own Bearer-only parser, so a Deepgram SDK sending Token, or bearer in its access_token mode, was refused even though HTTP auth accepted bearer. The socket path now uses get_bearer_token, which also learns Token, and the per-route scheme list is gone

Every route now accepts Token, and the passthrough leak check strips a caller key echoed under it because it reuses the same parser

litellm/proxy/auth/user_api_key_auth.py

@@ -1122,6 +1122,8 @@ def _should_route_jwt_to_oauth2_override(token: str, jwt_handler: JWTHandler) ->
 def get_bearer_token(
     api_key: str,
 ) -> str:
+    if api_key.startswith("Token "):
+        return api_key[len("Token ") :]
     if api_key.startswith("Bearer "):  # ensure Bearer token passed in
         api_key = api_key.replace("Bearer ", "")  # extract the token
     elif api_key.startswith("Basic "):
@@ -1202,14 +1204,8 @@ def _has_required_baggage_tag_mapping() -> bool:
 
 
 async def user_api_key_auth_websocket_for_model(
-    websocket: WebSocket,
-    model: str | None,
-    *,
-    close_on_auth_error: bool = True,
-    accepted_authorization_schemes: tuple[str, ...] = ("Bearer",),
+    websocket: WebSocket, model: str | None, *, close_on_auth_error: bool = True
 ) -> UserAPIKeyAuth:
-    """``accepted_authorization_schemes`` are the ``Authorization`` schemes that may carry the LiteLLM key. A route
-    adds one only when a provider's own SDK sends it to the same path"""
     ws_scope: Final = websocket.scope or {}
     scope_headers: Final = list(ws_scope.get("headers") or [])
     # ``get_request_route`` falls back to ``request.url.path`` when
@@ -1252,16 +1248,11 @@ async def user_api_key_auth_websocket_for_model(
             await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
             raise HTTPException(status_code=403, detail="No API key provided")
     else:
-        scheme_prefix: Final = next(
-            (f"{scheme} " for scheme in accepted_authorization_schemes if authorization.startswith(f"{scheme} ")),
-            None,
-        )
-        if scheme_prefix is None:
+        api_key = get_bearer_token(authorization).strip()
+        if not api_key:
             await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
             raise HTTPException(status_code=403, detail="Invalid Authorization header format")
 
-        api_key = authorization[len(scheme_prefix) :].strip()
-
     # Call user_api_key_auth with the extracted API key
     # Note: You'll need to modify this to work with WebSocket context if needed
     try:

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -3450,14 +3450,9 @@ _DEEPGRAM_WS_UNPRICED_REASON: Final = (
 )
 
 
-_DEEPGRAM_WS_AUTHORIZATION_SCHEMES: Final = ("Bearer", "Token")
-
-
 async def deepgram_listen_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyAuth:
     return await user_api_key_auth_websocket_for_model(
-        websocket,
-        model=deepgram_listen_requested_model(websocket.url.query),
-        accepted_authorization_schemes=_DEEPGRAM_WS_AUTHORIZATION_SCHEMES,
+        websocket, model=deepgram_listen_requested_model(websocket.url.query)
     )
 
 

tests/unit/proxy/auth/test_user_api_key_auth.py

@@ -2303,21 +2303,18 @@ def _websocket_authorized_by(sent: list[Message], authorization: str) -> WebSock
     )
 
 
-DEEPGRAM_SDK_SCHEMES: Final = ("Bearer", "Token")
-
-
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    ("authorization", "accepted_schemes"),
+    "authorization",
     [
-        pytest.param("Bearer {key}", ("Bearer",), id="bearer by default"),
-        pytest.param("Bearer {key}", DEEPGRAM_SDK_SCHEMES, id="bearer where token is also accepted"),
-        pytest.param("Token {key}", DEEPGRAM_SDK_SCHEMES, id="token where the route accepts it"),
-        pytest.param("Token {key} ", DEEPGRAM_SDK_SCHEMES, id="token with trailing space"),
+        pytest.param("Bearer {key}", id="bearer"),
+        pytest.param("bearer {key}", id="deepgram sdk access_token mode"),
+        pytest.param("Token {key}", id="deepgram sdk api_key mode"),
+        pytest.param("Token {key} ", id="token with trailing space"),
     ],
 )
-async def test_websocket_auth_finds_the_virtual_key_under_each_accepted_scheme(
-    monkeypatch: pytest.MonkeyPatch, authorization: str, accepted_schemes: tuple[str, ...]
+async def test_websocket_auth_finds_the_virtual_key_under_each_scheme_http_auth_accepts(
+    monkeypatch: pytest.MonkeyPatch, authorization: str
 ) -> None:
     from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket_for_model
     from litellm.proxy.proxy_server import hash_token
@@ -2327,9 +2324,7 @@ async def test_websocket_auth_finds_the_virtual_key_under_each_accepted_scheme(
     sent: Final[list[Message]] = []
 
     authorized: Final = await user_api_key_auth_websocket_for_model(
-        _websocket_authorized_by(sent, authorization.format(key=user_key)),
-        "nova-3",
-        accepted_authorization_schemes=accepted_schemes,
+        _websocket_authorized_by(sent, authorization.format(key=user_key)), "nova-3"
     )
 
     assert authorized.token == hash_token(user_key)
@@ -2338,16 +2333,15 @@ async def test_websocket_auth_finds_the_virtual_key_under_each_accepted_scheme(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    ("authorization", "accepted_schemes"),
+    "authorization",
     [
-        pytest.param("Token {key}", ("Bearer",), id="token where the route did not opt in"),
-        pytest.param("token {key}", DEEPGRAM_SDK_SCHEMES, id="scheme is case sensitive like bearer"),
-        pytest.param("Token{key}", DEEPGRAM_SDK_SCHEMES, id="no space after the scheme"),
-        pytest.param("Basic {key}", DEEPGRAM_SDK_SCHEMES, id="scheme not in the list"),
+        pytest.param("Token{key}", id="no space after the scheme"),
+        pytest.param("Digest {key}", id="unknown scheme"),
+        pytest.param("Token ", id="scheme without a key"),
     ],
 )
-async def test_websocket_auth_refuses_other_schemes_with_a_policy_close(
-    monkeypatch: pytest.MonkeyPatch, authorization: str, accepted_schemes: tuple[str, ...]
+async def test_websocket_auth_refuses_an_unparseable_authorization_with_a_policy_close(
+    monkeypatch: pytest.MonkeyPatch, authorization: str
 ) -> None:
     from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket_for_model
     from litellm.proxy.proxy_server import hash_token
@@ -2358,11 +2352,25 @@ async def test_websocket_auth_refuses_other_schemes_with_a_policy_close(
 
     with pytest.raises(HTTPException) as refusal:
         await user_api_key_auth_websocket_for_model(
-            _websocket_authorized_by(sent, authorization.format(key=user_key)),
-            "nova-3",
-            accepted_authorization_schemes=accepted_schemes,
+            _websocket_authorized_by(sent, authorization.format(key=user_key)), "nova-3"
         )
 
     assert refusal.value.status_code == 403
     assert refusal.value.detail == "Invalid Authorization header format"
     assert sent == [{"type": "websocket.close", "code": status.WS_1008_POLICY_VIOLATION, "reason": ""}]
+
+
+@pytest.mark.asyncio
+async def test_http_auth_finds_the_virtual_key_in_a_token_authorization(monkeypatch: pytest.MonkeyPatch) -> None:
+    from fastapi import Request as HTTPRequest
+
+    from litellm.proxy.proxy_server import hash_token
+
+    user_key: Final = "sk-http-key-for-nova"
+    _serve_virtual_key(monkeypatch, user_key, UserAPIKeyAuth(token=hash_token(user_key)))
+    request: Final = HTTPRequest(scope={"type": "http", "method": "POST", "path": "/chat/completions", "headers": []})
+    request._url = URL(url="/chat/completions")  # pyright: ignore[reportPrivateUsage]  # Starlette caches the URL here
+
+    authorized: Final = await user_api_key_auth(request=request, api_key=f"Token {user_key}")
+
+    assert authorized.token == hash_token(user_key)

tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py

@@ -504,17 +504,3 @@ def test_deepgram_listen_accepts_the_litellm_key_in_a_deepgram_sdk_token_header(
     assert [(call.target, call.custom_headers) for call in relay.calls] == [
         ("wss://api.deepgram.com/v1/listen?model=nova-3", MappingProxyType({"Authorization": "Token dg-provider-key"}))
     ]
-
-
-def test_a_non_deepgram_websocket_route_still_refuses_a_token_header() -> None:
-    relay: Final = _FakeRelay()
-    client: Final = TestClient(_app_with_relay(relay))
-
-    with pytest.raises(WebSocketDisconnect) as disconnect:
-        with client.websocket_connect(
-            "/openai/v1/realtime?model=gpt-4o-realtime-preview", headers={"Authorization": "Token sk-litellm"}
-        ):
-            pass
-
-    assert disconnect.value.code == 1008
-    assert relay.calls == []

tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -4648,7 +4648,7 @@ class TestVertexCredentiallessPassthroughVirtualKeyLeak:
         assert self.VKEY not in " ".join(f"{name}:{value}" for name, value in forwarded.items())
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("scheme", ["Bearer", "bearer", "Basic"])
+    @pytest.mark.parametrize("scheme", ["Bearer", "bearer", "Basic", "Token"])
     async def test_virtual_key_echoed_in_authorization_with_any_scheme_is_stripped(self, monkeypatch, scheme):
         raised, forwarded = await self._run(
             monkeypatch,