mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
parent
e6ac4cb230
commit
f4d29cc268
2 changed files with 180 additions and 18 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
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"
|
||||
Loading…
Add table
Reference in a new issue