From f4d29cc2682baab9da24c38d4496c16dd2a4eea0 Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Wed, 4 Mar 2026 21:36:08 -0800 Subject: [PATCH] Allow team admins to query team members via /v2/user/info Team admins can now query info for users in teams they administer. Access control order: proxy admin > self > team admin > deny. Co-Authored-By: Claude Opus 4.6 --- .../internal_user_endpoints.py | 65 +++++++-- .../test_internal_user_endpoints.py | 133 +++++++++++++++++- 2 files changed, 180 insertions(+), 18 deletions(-) diff --git a/litellm/proxy/management_endpoints/internal_user_endpoints.py b/litellm/proxy/management_endpoints/internal_user_endpoints.py index 014d0b0c910..679867935b7 100644 --- a/litellm/proxy/management_endpoints/internal_user_endpoints.py +++ b/litellm/proxy/management_endpoints/internal_user_endpoints.py @@ -30,7 +30,11 @@ from litellm.proxy.management_endpoints.common_daily_activity import ( get_daily_activity, get_daily_activity_aggregated, ) -from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view +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, +) from litellm.proxy.management_endpoints.key_management_endpoints import ( generate_key_helper_fn, prepare_metadata_fields, @@ -766,19 +770,6 @@ async def user_info_v2( detail="user_id is required", ) - # Non-admin users can only query their own info - if ( - user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN - and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY - and user_id != user_api_key_dict.user_id - ): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Not allowed to access other user's info. Your user_id={}, requested user_id={}".format( - user_api_key_dict.user_id, user_id - ), - ) - user_info = await prisma_client.get_data(user_id=user_id) if user_info is None: @@ -787,6 +778,25 @@ async def user_info_v2( detail=f"User {user_id} not found", ) + # Access control: non-admin users can only query their own info + # or info of users in teams they admin + if ( + user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN + and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY + and user_id != user_api_key_dict.user_id + ): + if not await _is_team_admin_for_user( + user_api_key_dict=user_api_key_dict, + target_user_teams=user_info.teams, + prisma_client=prisma_client, + ): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Not allowed to access other user's info. Your user_id={}, requested user_id={}".format( + user_api_key_dict.user_id, user_id + ), + ) + return UserInfoV2Response( user_id=user_id, user_info=user_info.model_dump(), @@ -800,6 +810,33 @@ async def user_info_v2( raise handle_exception_on_proxy(e) +async def _is_team_admin_for_user( + user_api_key_dict: UserAPIKeyAuth, + target_user_teams: list, + prisma_client: "PrismaClient", +) -> 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 + return False + + async def _get_user_info_for_proxy_admin(user_api_key_dict: UserAPIKeyAuth): """ Admin UI Endpoint - Returns All Teams and Keys when Proxy Admin is querying diff --git a/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py index e45df19bc45..07d7da910b5 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py @@ -1514,17 +1514,24 @@ async def test_user_info_v1_has_deprecation_header(mocker): @pytest.mark.asyncio async def test_user_info_v2_non_admin_cannot_query_other_user(mocker): """ - Test that a non-admin user gets 403 when querying another user's info - via /v2/user/info endpoint handler (defense-in-depth). + Test that a non-admin, non-team-admin user gets 403 when querying + another user's info via /v2/user/info. """ from fastapi import Request - from litellm.proxy._types import ProxyException, UserAPIKeyAuth + from litellm.proxy._types import LiteLLM_UserTable, ProxyException, UserAPIKeyAuth from litellm.proxy.management_endpoints.internal_user_endpoints import ( user_info_v2, ) mock_prisma_client = mocker.MagicMock() + + mock_target_user = LiteLLM_UserTable( + user_id="user-B", + teams=[], + ) + + mock_prisma_client.get_data = mocker.AsyncMock(return_value=mock_target_user) mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_request = mocker.MagicMock(spec=Request) @@ -1540,4 +1547,122 @@ async def test_user_info_v2_non_admin_cannot_query_other_user(mocker): ) assert exc_info.value.code == "403" - assert "Not allowed" in str(exc_info.value.message) \ No newline at end of file + assert "Not allowed" in str(exc_info.value.message) + + +@pytest.mark.asyncio +async def test_user_info_v2_team_admin_can_query_team_member(mocker): + """ + Test that a team admin can query info for a user in their team. + """ + from fastapi import Request + + from litellm.proxy._types import ( + LiteLLM_TeamTable, + LiteLLM_UserTable, + Member, + UserAPIKeyAuth, + ) + from litellm.proxy.management_endpoints.internal_user_endpoints import ( + user_info_v2, + ) + + mock_prisma_client = mocker.MagicMock() + + # Target user is in team-1 + mock_target_user = LiteLLM_UserTable( + user_id="user-B", + user_email="userB@example.com", + teams=["team-1"], + ) + + 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( + team_id="team-1", + members_with_roles=[ + Member(user_id="user-A", role="admin"), + 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, + ) + mocker.patch("litellm.proxy.proxy_server.user_api_key_cache", mocker.MagicMock()) + + mock_request = mocker.MagicMock(spec=Request) + mock_user_api_key_dict = UserAPIKeyAuth( + user_id="user-A", user_role="internal_user" + ) + + result = await user_info_v2( + user_id="user-B", + user_api_key_dict=mock_user_api_key_dict, + request=mock_request, + ) + + assert result.user_id == "user-B" + + +@pytest.mark.asyncio +async def test_user_info_v2_team_member_cannot_query_other_team_member(mocker): + """ + Test that a non-admin team member cannot query another member's info. + """ + from fastapi import Request + + from litellm.proxy._types import ( + LiteLLM_TeamTable, + LiteLLM_UserTable, + Member, + ProxyException, + UserAPIKeyAuth, + ) + from litellm.proxy.management_endpoints.internal_user_endpoints import ( + user_info_v2, + ) + + mock_prisma_client = mocker.MagicMock() + + mock_target_user = LiteLLM_UserTable( + user_id="user-B", + teams=["team-1"], + ) + + 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( + team_id="team-1", + members_with_roles=[ + Member(user_id="user-A", role="user"), + 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, + ) + mocker.patch("litellm.proxy.proxy_server.user_api_key_cache", mocker.MagicMock()) + + mock_request = mocker.MagicMock(spec=Request) + mock_user_api_key_dict = UserAPIKeyAuth( + user_id="user-A", user_role="internal_user" + ) + + with pytest.raises(ProxyException) as exc_info: + await user_info_v2( + user_id="user-B", + user_api_key_dict=mock_user_api_key_dict, + request=mock_request, + ) + + assert exc_info.value.code == "403" \ No newline at end of file