fix(auth): let jwt team_allowed_routes paths grant auth=true passthrough

Explicit paths and trailing-wildcard prefixes in litellm_jwtauth.team_allowed_routes
passed the JWT route check but were then denied by the auth-enforced passthrough
gates, which only read allowed_passthrough_routes from key or team metadata. Both
gates now also accept an explicit team_allowed_routes entry for tokens built by JWT
auth. Named route groups still never grant, and virtual keys, including JWT-mapped
ones, stay key-scoped
This commit is contained in:
ryan-crabbe-berri 2026-09-21 15:11:48 -07:00
parent 7b8bc54237
commit a8003102b2
4 changed files with 330 additions and 4 deletions

View file

@ -14,7 +14,7 @@ import hashlib
import os
import re
import time
from collections.abc import Awaitable, Callable, Mapping, Sequence
from collections.abc import Awaitable, Callable, Collection, Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Final, Literal, NoReturn, Protocol, TypeVar, cast
@ -1602,6 +1602,7 @@ class JWTAuthManager:
team_object: LiteLLM_TeamTable | None,
route: str,
request_method: str | None = None,
team_allowed_routes: Collection[str] = (),
) -> bool:
normalized_request_method: Final = request_method.upper() if isinstance(request_method, str) else None
if not RouteChecks.is_auth_enforced_pass_through_route(
@ -1610,8 +1611,11 @@ class JWTAuthManager:
):
return True
if RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=team_allowed_routes):
return True
# JWT team selection is team-scoped; key metadata is not available here,
# so passthrough access is granted only by the selected team's metadata.
# so beyond the JWT config grant above, only the selected team's metadata grants access.
return RouteChecks.check_passthrough_route_access(
route=route,
user_api_key_dict=UserAPIKeyAuth(team_metadata=(team_object.metadata or {}) if team_object else {}),
@ -1689,6 +1693,7 @@ class JWTAuthManager:
team_object=team_object,
route=route,
request_method=request_method,
team_allowed_routes=jwt_handler.litellm_jwtauth.team_allowed_routes or (),
):
is_allowed = False
denied_auth_enforced_pass_through_route = True
@ -2584,6 +2589,7 @@ class JWTAuthManager:
team_object=team_object,
route=route,
request_method=request_method,
team_allowed_routes=handler.litellm_jwtauth.team_allowed_routes or (),
):
JWTAuthManager._raise_team_passthrough_route_denial(route=route)
@ -2653,6 +2659,7 @@ class JWTAuthManager:
team_object=team_object,
route=route,
request_method=request_method,
team_allowed_routes=handler.litellm_jwtauth.team_allowed_routes or (),
):
JWTAuthManager._raise_team_passthrough_route_denial(route=route)
elif team_id is None:

View file

@ -268,7 +268,11 @@ class RouteChecks:
route=route,
method=RouteChecks._get_request_method(request=request),
):
RouteChecks._require_auth_pass_through_access(route=route, valid_token=valid_token)
RouteChecks._require_auth_pass_through_access(
route=route,
valid_token=valid_token,
jwt_team_allowed_routes=RouteChecks._jwt_team_allowed_routes(valid_token=valid_token),
)
elif RouteChecks.is_llm_api_route(route=route):
pass
elif RouteChecks.is_info_route(route=route):
@ -679,16 +683,41 @@ class RouteChecks:
),
)
@staticmethod
def jwt_team_routes_grant_pass_through(route: str, team_allowed_routes: Collection[str]) -> bool:
"""
Explicit paths and trailing-wildcard prefixes grant auth=true pass-through. Entries are only ever compared
as paths, so a named route group like ``openai_routes`` never grants.
"""
return any(
RouteChecks.route_matches_wildcard_pattern(route=route, pattern=allowed_route)
for allowed_route in team_allowed_routes
)
@staticmethod
def _jwt_team_allowed_routes(valid_token: UserAPIKeyAuth) -> Collection[str]:
"""``team_allowed_routes`` for tokens built by JWT auth; JWT-mapped virtual keys stay key-scoped."""
if valid_token.jwt_claims is None or valid_token.token is not None:
return ()
from litellm.proxy.proxy_server import jwt_handler
return jwt_handler.litellm_jwtauth.team_allowed_routes or ()
@staticmethod
def _require_auth_pass_through_access(
route: str,
valid_token: UserAPIKeyAuth,
jwt_team_allowed_routes: Collection[str] = (),
) -> None:
"""
Require an explicit ``allowed_passthrough_routes`` match for auth=true pass-through.
Require an explicit grant for auth=true pass-through: ``allowed_passthrough_routes`` on the
key or team, or an explicit JWT ``team_allowed_routes`` entry.
"""
if RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token):
return
if RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=jwt_team_allowed_routes):
return
raise RouteChecks._auth_pass_through_denied_exception(route=route)
@staticmethod

View file

@ -291,6 +291,106 @@ async def test_find_team_with_model_access_uses_request_method_for_passthrough_a
assert "allowed_passthrough_routes" in exc_info.value.detail
_AUTH_ENFORCED_MODEL_HOST_ROUTES: Final = {
"test-uuid-1:subpath:/model-host/v1/extractor:GET,POST": {
"endpoint_id": "test-uuid-1",
"path": "/model-host/v1/extractor",
"type": "subpath",
"auth": True,
},
}
@pytest.mark.asyncio
async def test_find_team_with_model_access_team_allowed_routes_wildcard_grants_auth_passthrough():
jwt_handler = JWTHandler()
jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth(team_allowed_routes=["openai_routes", "/model-host/*"])
team_without_passthrough_allowlist = LiteLLM_TeamTable(team_id="team-a", models=["all-proxy-models"], metadata={})
with (
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team_without_passthrough_allowlist,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
_AUTH_ENFORCED_MODEL_HOST_ROUTES,
),
patch("litellm.proxy.utils.get_server_root_path", return_value="/"),
):
team_id, team_obj = await JWTAuthManager.find_team_with_model_access(
team_ids={"team-a"},
requested_model=None,
route="/model-host/v1/extractor/predict",
jwt_handler=jwt_handler,
prisma_client=None,
user_api_key_cache=MagicMock(),
parent_otel_span=None,
proxy_logging_obj=MagicMock(),
request_method="POST",
)
assert team_id == "team-a"
assert team_obj == team_without_passthrough_allowlist
@pytest.mark.asyncio
async def test_auth_builder_header_team_allows_auth_passthrough_for_team_allowed_routes_wildcard():
from litellm.proxy.utils import ProxyLogging
jwt_handler = JWTHandler()
user_api_key_cache = DualCache()
jwt_handler.update_environment(
prisma_client=None,
user_api_key_cache=user_api_key_cache,
litellm_jwtauth=LiteLLM_JWTAuth(
team_ids_jwt_field="groups",
user_id_jwt_field="sub",
team_allowed_routes=["openai_routes", "/model-host/*"],
),
)
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock) as mock_auth_jwt,
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(JWTAuthManager, "check_admin_access", new_callable=AsyncMock, return_value=None),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=LiteLLM_TeamTable(team_id="team-2", metadata={}),
),
patch.object(
JWTAuthManager,
"get_objects",
new_callable=AsyncMock,
return_value=(None, None, None, None, "user-1"),
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
_AUTH_ENFORCED_MODEL_HOST_ROUTES,
),
patch("litellm.proxy.utils.get_server_root_path", return_value="/"),
):
mock_auth_jwt.return_value = {"sub": "user-1", "scope": "", "groups": ["team-1", "team-2"]}
result = await JWTAuthManager.auth_builder(
api_key="jwt-token",
jwt_handler=jwt_handler,
request_data={},
general_settings={},
route="/model-host/v1/extractor/predict",
prisma_client=None,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
request_headers={"x-litellm-team-id": "team-2"},
request_method="POST",
)
assert result["team_id"] == "team-2"
@pytest.mark.asyncio
async def test_auth_builder_proxy_admin_user_role():
"""Test that is_proxy_admin is True when user_object.user_role is PROXY_ADMIN"""
@ -6463,6 +6563,90 @@ async def test_auth_builder_db_fallback_enforces_passthrough_route_access():
assert "passthrough route" in exc_info.value.detail
async def _auth_builder_via_db_team_fallback(team_allowed_routes: list[str]):
user_id = "u_passthrough"
user_object = LiteLLM_UserTable(
user_id=user_id,
user_role=LitellmUserRoles.INTERNAL_USER,
teams=["team_no_passthrough"],
)
jwt_handler = JWTHandler()
jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth(fallback_to_db_teams=True, team_allowed_routes=team_allowed_routes)
async def fake_get_team(team_id, **kwargs):
return LiteLLM_TeamTable(team_id=team_id, metadata={})
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock, return_value={"sub": user_id, "scope": ""}),
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(jwt_handler, "get_rbac_role", return_value=None),
patch.object(jwt_handler, "get_scopes", return_value=[]),
patch.object(jwt_handler, "get_object_id", return_value=None),
patch.object(
JWTAuthManager,
"get_user_info",
new_callable=AsyncMock,
return_value=(user_id, "u@example.com", True),
),
patch.object(jwt_handler, "get_org_id", return_value=None),
patch.object(jwt_handler, "get_end_user_id", return_value=None),
patch.object(JWTAuthManager, "check_admin_access", new_callable=AsyncMock, return_value=None),
patch.object(
JWTAuthManager,
"get_objects",
new_callable=AsyncMock,
return_value=(user_object, None, None, None, user_id),
),
patch.object(JWTAuthManager, "map_user_to_teams", new_callable=AsyncMock),
patch.object(JWTAuthManager, "validate_object_id", return_value=True),
patch.object(JWTAuthManager, "sync_user_role_and_teams", new_callable=AsyncMock),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
side_effect=fake_get_team,
),
patch(
"litellm.proxy.auth.handle_jwt.get_team_membership",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
_AUTH_ENFORCED_MODEL_HOST_ROUTES,
),
patch("litellm.proxy.utils.get_server_root_path", return_value="/"),
):
return await JWTAuthManager.auth_builder(
api_key="test_jwt_token",
jwt_handler=jwt_handler,
request_data={},
general_settings={"enforce_rbac": False},
route="/model-host/v1/extractor/predict",
prisma_client=None,
user_api_key_cache=None,
parent_otel_span=None,
proxy_logging_obj=None,
request_headers=None,
request_method="POST",
)
@pytest.mark.asyncio
async def test_auth_builder_db_fallback_team_allowed_routes_wildcard_grants_auth_passthrough():
result = await _auth_builder_via_db_team_fallback(team_allowed_routes=["openai_routes", "/model-host/*"])
assert result["team_id"] == "team_no_passthrough"
@pytest.mark.asyncio
async def test_auth_builder_db_fallback_route_groups_alone_do_not_grant_auth_passthrough():
with pytest.raises(HTTPException) as exc_info:
await _auth_builder_via_db_team_fallback(team_allowed_routes=["openai_routes", "mapped_pass_through_routes"])
assert exc_info.value.status_code == 403, exc_info.value.detail
assert "allowed_passthrough_routes" in exc_info.value.detail
@pytest.mark.asyncio
async def test_sync_user_role_and_teams_singular_claim_reconciles_memberships():
"""When fallback_to_db_teams is on but the JWT carries a singular team claim

View file

@ -1263,6 +1263,112 @@ def test_non_proxy_admin_allows_auth_pass_through_with_team_allowlist():
)
@pytest.mark.parametrize(
"route, team_allowed_routes, expected",
[
("/model-host/v1/extractor/predict", ["/model-host/*"], True),
("/model-host", ["/model-host/*"], False),
("/model-host/v1/extractor", ["/model-host/v1/extractor"], True),
("/model-host/v1/extractor/predict", ["/model-host/v1/extractor"], False),
("/other/v1/extractor", ["/model-host/*"], False),
("/model-host/v1/extractor", ["openai_routes", "llm_api_routes", "mapped_pass_through_routes"], False),
("/model-host/v1/extractor", [], False),
],
)
def test_jwt_team_routes_grant_pass_through_only_for_explicit_paths(route, team_allowed_routes, expected):
assert (
RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=team_allowed_routes)
is expected
)
_AUTH_ENFORCED_MODEL_HOST_ROUTES: Final = {
"test-uuid-1:subpath:/model-host/v1/extractor:GET,POST": {
"endpoint_id": "test-uuid-1",
"path": "/model-host/v1/extractor",
"type": "subpath",
"auth": True,
},
}
def _jwt_handler_with_team_allowed_routes(team_allowed_routes: list[str]):
from litellm.proxy._types import LiteLLM_JWTAuth
from litellm.proxy.auth.handle_jwt import JWTHandler
jwt_handler: Final = JWTHandler()
jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth(team_allowed_routes=team_allowed_routes)
return jwt_handler
def _check_model_host_route_as(valid_token: UserAPIKeyAuth, team_allowed_routes: list[str]) -> None:
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
_AUTH_ENFORCED_MODEL_HOST_ROUTES,
),
patch("litellm.proxy.utils.get_server_root_path", return_value="/"),
patch(
"litellm.proxy.proxy_server.jwt_handler",
_jwt_handler_with_team_allowed_routes(team_allowed_routes),
),
):
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=None,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route="/model-host/v1/extractor/predict",
request=MagicMock(spec=Request),
valid_token=valid_token,
request_data={},
)
def test_non_proxy_admin_allows_auth_pass_through_for_jwt_team_allowed_routes_wildcard():
jwt_token: Final = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
team_id="team-a",
jwt_claims={"sub": "test_user"},
)
_check_model_host_route_as(jwt_token, team_allowed_routes=["openai_routes", "/model-host/*"])
def test_non_proxy_admin_denies_auth_pass_through_for_jwt_when_only_route_groups_configured():
jwt_token: Final = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
team_id="team-a",
jwt_claims={"sub": "test_user"},
)
with pytest.raises(HTTPException) as exc_info:
_check_model_host_route_as(jwt_token, team_allowed_routes=["openai_routes", "mapped_pass_through_routes"])
assert exc_info.value.status_code == 403, exc_info.value.detail
assert "allowed_passthrough_routes" in exc_info.value.detail
@pytest.mark.parametrize(
"jwt_claims",
[None, {"sub": "test_user"}],
ids=["plain_virtual_key", "jwt_mapped_virtual_key"],
)
def test_non_proxy_admin_jwt_team_allowed_routes_never_grant_pass_through_to_virtual_keys(jwt_claims):
virtual_key: Final = UserAPIKeyAuth(
api_key="sk-test-key",
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
jwt_claims=jwt_claims,
)
with pytest.raises(HTTPException) as exc_info:
_check_model_host_route_as(virtual_key, team_allowed_routes=["openai_routes", "/model-host/*"])
assert exc_info.value.status_code == 403, exc_info.value.detail
assert "allowed_passthrough_routes" in exc_info.value.detail
def test_virtual_key_without_llm_api_routes_cannot_access_pass_through():
"""
Test that virtual keys without llm_api_routes permission cannot access registered pass-through endpoints.