Fix route check blocking team-admin access, N+1 queries, and silent exception swallowing

- Let /v2/user/info pass through route-level check (endpoint handles its
  own access control including team-admin logic), matching /key/info pattern
- Batch-fetch teams with find_many instead of per-team get_team_object calls
- Log exceptions in team admin check instead of silently swallowing them

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang 2026-03-05 17:07:29 -08:00
parent 4ff02a4c65
commit 771c13d56b
4 changed files with 38 additions and 66 deletions

View file

@ -170,7 +170,10 @@ class RouteChecks:
if route == "/key/info":
# handled by function itself
pass
elif route in ("/user/info", "/v2/user/info"):
elif route == "/v2/user/info":
# handled by endpoint itself (supports team-admin access)
pass
elif route == "/user/info":
# check if user can access this route
query_params = request.query_params
user_id = query_params.get("user_id")

View file

@ -30,7 +30,6 @@ from litellm.proxy.management_endpoints.common_daily_activity import (
get_daily_activity,
get_daily_activity_aggregated,
)
from litellm.proxy.auth.auth_checks import get_team_object
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_user_has_admin_view,
@ -819,23 +818,28 @@ async def _is_team_admin_for_user(
) -> bool:
"""
Check if the caller is a team admin for any team that the target user belongs to.
"""
from litellm.proxy.proxy_server import user_api_key_cache
for team_id in target_user_teams:
try:
team_obj = await get_team_object(
team_id=team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
check_db_only=True,
)
if _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=team_obj
):
return True
except Exception:
continue
Batch-fetches all teams in a single query to avoid N+1 DB calls.
"""
if not target_user_teams:
return False
try:
team_rows = await prisma_client.db.litellm_teamtable.find_many(
where={"team_id": {"in": target_user_teams}}
)
except Exception:
verbose_proxy_logger.exception(
"_is_team_admin_for_user: failed to fetch teams for user"
)
return False
for row in team_rows:
team_obj = LiteLLM_TeamTable(**row.model_dump())
if _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=team_obj
):
return True
return False

View file

@ -1192,9 +1192,11 @@ def test_non_admin_non_team_admin_cannot_access_config_update_but_can_attempt_re
assert "Only proxy admin can be used to generate" in str(exc_info.value)
def test_v2_user_info_non_admin_blocked_for_other_user():
def test_v2_user_info_passes_through_route_check():
"""
Test that /v2/user/info blocks non-admin users from querying another user's info.
Test that /v2/user/info passes through the route-level check for all users,
even when querying another user. The endpoint handles its own access control
(including team-admin logic).
"""
user_obj = LiteLLM_UserTable(
user_id="user-A",
@ -1209,38 +1211,7 @@ def test_v2_user_info_non_admin_blocked_for_other_user():
request = MagicMock(spec=Request)
request.query_params = {"user_id": "user-B"}
with pytest.raises(HTTPException) as exc_info:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route="/v2/user/info",
request=request,
valid_token=valid_token,
request_data={},
)
assert exc_info.value.status_code == 403
assert "key not allowed to access this user's info" in str(exc_info.value.detail)
def test_v2_user_info_non_admin_allowed_for_own_user():
"""
Test that /v2/user/info allows non-admin users to query their own info.
"""
user_obj = LiteLLM_UserTable(
user_id="user-A",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
valid_token = UserAPIKeyAuth(
user_id="user-A",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
request = MagicMock(spec=Request)
request.query_params = {"user_id": "user-A"}
# Should not raise
# Should not raise — endpoint handles access control itself
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,

View file

@ -1577,7 +1577,6 @@ async def test_user_info_v2_team_admin_can_query_team_member(mocker):
)
mock_prisma_client.get_data = mocker.AsyncMock(return_value=mock_target_user)
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
# Caller (user-A) is admin of team-1
mock_team = LiteLLM_TeamTable(
@ -1587,13 +1586,11 @@ async def test_user_info_v2_team_admin_can_query_team_member(mocker):
Member(user_id="user-B", role="user"),
],
)
mock_get_team_object = mocker.AsyncMock(return_value=mock_team)
mocker.patch(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_team_object",
mock_get_team_object,
mock_prisma_client.db.litellm_teamtable.find_many = mocker.AsyncMock(
return_value=[mock_team]
)
mocker.patch("litellm.proxy.proxy_server.user_api_key_cache", mocker.MagicMock())
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mock_request = mocker.MagicMock(spec=Request)
mock_user_api_key_dict = UserAPIKeyAuth(
@ -1635,7 +1632,6 @@ async def test_user_info_v2_team_member_cannot_query_other_team_member(mocker):
)
mock_prisma_client.get_data = mocker.AsyncMock(return_value=mock_target_user)
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
# Caller (user-A) is a regular member, NOT admin
mock_team = LiteLLM_TeamTable(
@ -1645,13 +1641,11 @@ async def test_user_info_v2_team_member_cannot_query_other_team_member(mocker):
Member(user_id="user-B", role="user"),
],
)
mock_get_team_object = mocker.AsyncMock(return_value=mock_team)
mocker.patch(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_team_object",
mock_get_team_object,
mock_prisma_client.db.litellm_teamtable.find_many = mocker.AsyncMock(
return_value=[mock_team]
)
mocker.patch("litellm.proxy.proxy_server.user_api_key_cache", mocker.MagicMock())
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mock_request = mocker.MagicMock(spec=Request)
mock_user_api_key_dict = UserAPIKeyAuth(