From fe63650ebd0945933f080d1f79fd73db45b341dd Mon Sep 17 00:00:00 2001 From: ryan-crabbe-berri Date: Mon, 18 May 2026 08:56:25 -0700 Subject: [PATCH] fix(proxy): gate team allowed_passthrough_routes to proxy admins (#28097) * fix(proxy): gate team allowed_passthrough_routes to proxy admins allowed_passthrough_routes short-circuits the role-based route gate, so the keys endpoints already restrict it to proxy admins. The team writers (/team/new, /team/update) had no equivalent check, letting an org admin (a non-proxy-admin who clears the route gate and _verify_team_access) self-grant pass-through routes on their team. Lift the keys check into a shared helper and apply it to both team endpoints. Resolves LIT-3019 * docs(proxy): note view-only admins are intentionally excluded from passthrough gate Clarifies the proxy-admin guard per review feedback; no behavior change. Refs LIT-3019 --- .../management_endpoints/common_utils.py | 32 ++++++ .../key_management_endpoints.py | 31 +----- .../management_endpoints/team_endpoints.py | 9 ++ .../test_team_endpoints.py | 100 ++++++++++++++++++ 4 files changed, 142 insertions(+), 30 deletions(-) diff --git a/litellm/proxy/management_endpoints/common_utils.py b/litellm/proxy/management_endpoints/common_utils.py index a43d15a580..dc27e87726 100644 --- a/litellm/proxy/management_endpoints/common_utils.py +++ b/litellm/proxy/management_endpoints/common_utils.py @@ -1,6 +1,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union from fastapi import HTTPException, status +from pydantic import BaseModel from litellm._logging import verbose_proxy_logger from litellm.caching import DualCache @@ -53,6 +54,37 @@ def require_caller_user_id_for_non_admin( return user_api_key_dict.user_id +def _check_passthrough_routes_caller_permission( + data: BaseModel, + user_api_key_dict: UserAPIKeyAuth, + *, + entity: str = "key", +) -> None: + """ + Only proxy admins may set `allowed_passthrough_routes` (top-level or under + `metadata`) — it short-circuits the role-based route gate, so keys and teams + must be gated identically. + """ + # view-only admins excluded by design; blocked upstream from writes anyway + if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: + return + if getattr(data, "allowed_passthrough_routes", None): + raise HTTPException( + status_code=403, + detail={ + "error": f"Only proxy admins can set `allowed_passthrough_routes` on a {entity}." + }, + ) + metadata = getattr(data, "metadata", None) + if isinstance(metadata, dict) and metadata.get("allowed_passthrough_routes"): + raise HTTPException( + status_code=403, + detail={ + "error": f"Only proxy admins can set `metadata.allowed_passthrough_routes` on a {entity}." + }, + ) + + def _is_user_team_admin( user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable ) -> bool: diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 7ff706eb91..2ab147043d 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -55,6 +55,7 @@ from litellm.proxy.common_utils.rbac_utils import check_org_admin_can_generate_k from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time from litellm.proxy.hooks.key_management_event_hooks import KeyManagementEventHooks from litellm.proxy.management_endpoints.common_utils import ( + _check_passthrough_routes_caller_permission, _is_user_org_admin_for_team, _is_user_team_admin, _set_object_metadata_field, @@ -548,36 +549,6 @@ def _check_allowed_routes_caller_permission( ) -def _check_passthrough_routes_caller_permission( - data: BaseModel, - user_api_key_dict: UserAPIKeyAuth, -) -> None: - """ - Only proxy admins may set `allowed_passthrough_routes` on a key, either at - the top level of the request or nested under `metadata`. - - The route gate evaluates passthrough access ahead of the standard role - gate, so the field is restricted to admins to keep that ordering safe. - """ - if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: - return - if getattr(data, "allowed_passthrough_routes", None): - raise HTTPException( - status_code=403, - detail={ - "error": "Only proxy admins can set `allowed_passthrough_routes` on a key." - }, - ) - metadata = getattr(data, "metadata", None) - if isinstance(metadata, dict) and metadata.get("allowed_passthrough_routes"): - raise HTTPException( - status_code=403, - detail={ - "error": "Only proxy admins can set `metadata.allowed_passthrough_routes` on a key." - }, - ) - - async def validate_team_id_used_in_service_account_request( team_id: Optional[str], prisma_client: Optional[PrismaClient], diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py index 35e3d196e9..86c4d6dcd9 100644 --- a/litellm/proxy/management_endpoints/team_endpoints.py +++ b/litellm/proxy/management_endpoints/team_endpoints.py @@ -73,6 +73,7 @@ from litellm.proxy.auth.auth_checks import ( ) from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.management_endpoints.common_utils import ( + _check_passthrough_routes_caller_permission, _is_user_org_admin_for_team, _is_user_team_admin, _set_object_metadata_field, @@ -1049,6 +1050,10 @@ async def new_team( # noqa: PLR0915 Member(role="admin", user_id=user_api_key_dict.user_id) ) + _check_passthrough_routes_caller_permission( + data, user_api_key_dict, entity="team" + ) + ## ADD TO MODEL TABLE _model_id = None if data.model_aliases is not None and isinstance(data.model_aliases, dict): @@ -1646,6 +1651,10 @@ async def update_team( # noqa: PLR0915 user_api_key_dict=user_api_key_dict, ) + _check_passthrough_routes_caller_permission( + data, user_api_key_dict, entity="team" + ) + if data.soft_budget is not None: max_budget_to_check = ( data.max_budget diff --git a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py index 5c7bbc46c9..b450a26290 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py @@ -7943,3 +7943,103 @@ async def test_team_member_me_returns_404_for_unknown_team(mock_db_client): user_api_key_dict=caller_auth, ) assert exc_info.value.status_code == 404 + + +def _non_admin_auth(): + return UserAPIKeyAuth( + user_id="u-team-admin", user_role=LitellmUserRoles.INTERNAL_USER + ) + + +def test_check_passthrough_routes_caller_permission_team(): + from litellm.proxy._types import NewTeamRequest + from litellm.proxy.management_endpoints.common_utils import ( + _check_passthrough_routes_caller_permission, + ) + + admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + non_admin = _non_admin_auth() + + _check_passthrough_routes_caller_permission( + NewTeamRequest(allowed_passthrough_routes=["/foo/*"]), admin, entity="team" + ) + + _check_passthrough_routes_caller_permission( + NewTeamRequest(), non_admin, entity="team" + ) + _check_passthrough_routes_caller_permission( + NewTeamRequest(allowed_passthrough_routes=[]), non_admin, entity="team" + ) + + with pytest.raises(HTTPException) as exc: + _check_passthrough_routes_caller_permission( + NewTeamRequest(allowed_passthrough_routes=["/admin/*"]), + non_admin, + entity="team", + ) + assert exc.value.status_code == 403 + assert "allowed_passthrough_routes" in str(exc.value.detail) + assert "team" in str(exc.value.detail) + + with pytest.raises(HTTPException) as exc: + _check_passthrough_routes_caller_permission( + NewTeamRequest(metadata={"allowed_passthrough_routes": ["/admin/*"]}), + non_admin, + entity="team", + ) + assert exc.value.status_code == 403 + assert "metadata.allowed_passthrough_routes" in str(exc.value.detail) + + +@pytest.mark.asyncio +async def test_new_team_blocks_non_admin_passthrough_routes(mock_db_client): + """A non-proxy-admin cannot self-grant pass-through routes via /team/new.""" + mock_db_client.db.litellm_teamtable.count = AsyncMock(return_value=0) + from fastapi import Request + + from litellm.proxy._types import NewTeamRequest, ProxyException + from litellm.proxy.management_endpoints.team_endpoints import new_team + + with patch( + "litellm.proxy.management_endpoints.team_endpoints._check_user_team_limits", + AsyncMock(return_value=None), + ): + with pytest.raises(ProxyException) as exc: + await new_team( + data=NewTeamRequest( + team_alias="t", allowed_passthrough_routes=["/admin/*"] + ), + http_request=MagicMock(spec=Request), + user_api_key_dict=_non_admin_auth(), + ) + assert str(exc.value.code) == "403" + assert "allowed_passthrough_routes" in str(exc.value.message) + + +@pytest.mark.asyncio +async def test_update_team_blocks_non_admin_passthrough_routes(mock_db_client): + """Even a team manager (non-proxy-admin) cannot set pass-through routes via + /team/update — the gate runs after _verify_team_access.""" + from fastapi import Request + + from litellm.proxy._types import ProxyException, UpdateTeamRequest + from litellm.proxy.management_endpoints.team_endpoints import update_team + + existing = MagicMock() + existing.model_dump.return_value = {"team_id": "t1"} + mock_db_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=existing) + + with patch( + "litellm.proxy.management_endpoints.team_endpoints._verify_team_access", + AsyncMock(return_value=None), + ): + with pytest.raises(ProxyException) as exc: + await update_team( + data=UpdateTeamRequest( + team_id="t1", allowed_passthrough_routes=["/admin/*"] + ), + http_request=MagicMock(spec=Request), + user_api_key_dict=_non_admin_auth(), + ) + assert str(exc.value.code) == "403" + assert "allowed_passthrough_routes" in str(exc.value.message)