diff --git a/litellm/proxy/management_endpoints/common_utils.py b/litellm/proxy/management_endpoints/common_utils.py index bc2da33672f..0b5a19e93ba 100644 --- a/litellm/proxy/management_endpoints/common_utils.py +++ b/litellm/proxy/management_endpoints/common_utils.py @@ -176,6 +176,27 @@ def _team_member_has_permission( return False +def _user_has_full_team_view( + user_api_key_dict: UserAPIKeyAuth, + team_obj: LiteLLM_TeamTable, + permission: str, + is_confirmed_member: bool, +) -> bool: + """Whether a non-admin caller may see a team's full activity vs only their own keys. + + Membership comes from ``is_confirmed_member`` (the caller's own ``teams`` + list), not from ``team.members_with_roles``: SSO/JWT sync records a user in + ``LiteLLM_UserTable.teams`` but only materializes ``members_with_roles`` for + the request's active team, so a member of a permission-granting team would + otherwise be invisible here and silently scoped to their own keys. + """ + if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj): + return True + if not is_confirmed_member: + return False + return permission in (team_obj.team_member_permissions or []) + + async def _user_has_admin_privileges( user_api_key_dict: UserAPIKeyAuth, prisma_client: Optional["PrismaClient"] = None, diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py index f533774ce18..b889005e170 100644 --- a/litellm/proxy/management_endpoints/team_endpoints.py +++ b/litellm/proxy/management_endpoints/team_endpoints.py @@ -14,7 +14,7 @@ import json import math import traceback from datetime import datetime, timezone -from typing import Any, Dict, List, Optional, Tuple, Union, cast +from typing import Any, Dict, List, Optional, Set, Tuple, Union, cast import fastapi from fastapi import APIRouter, Depends, Header, HTTPException, Request, status @@ -79,10 +79,10 @@ from litellm.proxy.management_endpoints.common_utils import ( _is_user_org_admin_for_team, _is_user_team_admin, _set_object_metadata_field, - _team_member_has_permission, _update_metadata_fields, _upsert_budget_and_membership, _user_has_admin_view, + _user_has_full_team_view, ) from litellm.proxy.management_endpoints.organization_endpoints import ( add_member_to_organization, @@ -5349,14 +5349,13 @@ async def get_team_daily_activity( # Convert comma-separated tags string to list if provided team_ids_list = team_ids.split(",") if team_ids else None - exclude_team_ids_list: Optional[List[str]] = None + exclude_team_ids_list: Optional[List[str]] = ( + exclude_team_ids.split(",") if exclude_team_ids else None + ) - if exclude_team_ids: - exclude_team_ids_list = ( - exclude_team_ids.split(",") if exclude_team_ids else None - ) - - if not _user_has_admin_view(user_api_key_dict): + is_admin_view = _user_has_admin_view(user_api_key_dict) + member_team_ids: Set[str] = set() + if not is_admin_view: user_info = await get_user_object( user_id=user_api_key_dict.user_id, prisma_client=prisma_client, @@ -5374,20 +5373,22 @@ async def get_team_daily_activity( }, ) + member_team_ids = set(user_info.teams or []) if team_ids_list is None: - team_ids_list = user_info.teams + team_ids_list = list(member_team_ids) else: - # check if all team_ids are in user_info.teams - for team_id in team_ids_list: - if team_id not in user_info.teams: - raise HTTPException( - status_code=404, - detail={ - "error": "User does not belong to Team= {}. Call `/user/info` to see user's teams".format( - team_id - ) - }, - ) + unauthorized_team = next( + (t for t in team_ids_list if t not in member_team_ids), None + ) + if unauthorized_team is not None: + raise HTTPException( + status_code=404, + detail={ + "error": "User does not belong to Team= {}. Call `/user/info` to see user's teams".format( + unauthorized_team + ) + }, + ) ## Fetch team aliases and check team admin status where_condition = {} @@ -5400,43 +5401,30 @@ async def get_team_daily_activity( t.team_id: {"team_alias": t.team_alias} for t in team_aliases } - # Check if user is team admin or has /team/daily/activity permission - # If not, filter by user's API keys. - # - # Earlier this loop used `any-team admin -> set has_full_team_view=True - # for the entire request`, so an admin of one team that requested - # data for several teams would see API-key-level breakdowns for all - # of them. Require full view on EVERY requested team — if the caller - # only has admin/permission for a strict subset, fall back to - # filtering the entire response by their own API keys (they can re- - # request the admin-only teams separately to get the wider view). + # A non-admin caller sees a team's full activity only if they admin every + # requested team or that team granted the `/team/daily/activity` member + # permission; otherwise the whole response is scoped to their own keys. + # Membership comes from `member_team_ids` (the caller's own `teams` list, + # the same source the 404 gate above trusts), not `members_with_roles` — + # SSO/JWT sync leaves that unpopulated for teams other than the request's + # active one, which otherwise drops a permission-granting team's member to + # their own keys and shows zero team usage. user_api_keys: Optional[List[str]] = None - if not _user_has_admin_view(user_api_key_dict) and team_ids_list and team_aliases: - has_full_team_view = True - for team_alias in team_aliases: - team_obj = LiteLLM_TeamTable(**team_alias.model_dump()) - is_admin = _is_user_team_admin( - user_api_key_dict=user_api_key_dict, team_obj=team_obj - ) - has_perm = _team_member_has_permission( + if not is_admin_view and team_ids_list and team_aliases: + has_full_team_view = all( + _user_has_full_team_view( user_api_key_dict=user_api_key_dict, - team_obj=team_obj, + team_obj=LiteLLM_TeamTable(**team_alias.model_dump()), permission="/team/daily/activity", + is_confirmed_member=team_alias.team_id in member_team_ids, ) - if not (is_admin or has_perm): - has_full_team_view = False - break - - # If user does not have full team view, filter by their API keys + for team_alias in team_aliases + ) if not has_full_team_view: - # Get all API keys for this user user_keys = await VerificationTokenRepository( prisma_client ).table.find_many(where={"user_id": user_api_key_dict.user_id}) - user_api_keys = [key.token for key in user_keys if key.token] - # If user has no API keys, return empty result - if not user_api_keys: - user_api_keys = [""] # Use empty string to ensure no matches + user_api_keys = [key.token for key in user_keys if key.token] or [""] # If api_key parameter is provided, use it; otherwise use user_api_keys if set final_api_key_filter: Optional[Union[str, List[str]]] = api_key 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 ecce92e7324..71af1e32083 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py @@ -9434,3 +9434,134 @@ async def test_team_info_forwards_key_limit_to_get_data(): ) assert mock_prisma.get_data.await_args.kwargs["limit"] == 7 + + +def _team_with_permission(team_id, members, permissions): + return LiteLLM_TeamTable( + team_id=team_id, + members_with_roles=members, + team_member_permissions=permissions, + ) + + +def test_full_team_view_sso_member_with_permission_not_in_members_with_roles(): + """Regression: an SSO/JWT member is recorded in LiteLLM_UserTable.teams but + not in the team's members_with_roles. When that team grants the + /team/daily/activity permission, the caller must get full team view. + + The old members_with_roles-only check (_team_member_has_permission) returns + False for exactly this input, which is what silently scoped the caller to + their own keys and returned zero team usage. + """ + from litellm.proxy.management_endpoints.common_utils import ( + _team_member_has_permission, + _user_has_full_team_view, + ) + + caller = UserAPIKeyAuth( + user_id="sso-user", user_role=LitellmUserRoles.INTERNAL_USER + ) + team = _team_with_permission( + team_id="team-clinical", + members=[Member(user_id="someone-else", role="admin")], + permissions=["/team/daily/activity"], + ) + + assert ( + _team_member_has_permission( + user_api_key_dict=caller, + team_obj=team, + permission="/team/daily/activity", + ) + is False + ) + assert ( + _user_has_full_team_view( + user_api_key_dict=caller, + team_obj=team, + permission="/team/daily/activity", + is_confirmed_member=True, + ) + is True + ) + + +def test_full_team_view_member_without_permission_stays_scoped(): + """A confirmed member of a team that did NOT grant the permission and who is + not an admin gets no full view (still scoped to their own keys).""" + from litellm.proxy.management_endpoints.common_utils import ( + _user_has_full_team_view, + ) + + caller = UserAPIKeyAuth( + user_id="sso-user", user_role=LitellmUserRoles.INTERNAL_USER + ) + team = _team_with_permission( + team_id="team-clinical", + members=[Member(user_id="someone-else", role="admin")], + permissions=["/key/info"], + ) + + assert ( + _user_has_full_team_view( + user_api_key_dict=caller, + team_obj=team, + permission="/team/daily/activity", + is_confirmed_member=True, + ) + is False + ) + + +def test_full_team_view_team_admin_always_sees_team(): + """A team admin (present in members_with_roles with role admin) gets full + view regardless of the member-permission list.""" + from litellm.proxy.management_endpoints.common_utils import ( + _user_has_full_team_view, + ) + + caller = UserAPIKeyAuth( + user_id="team-admin", user_role=LitellmUserRoles.INTERNAL_USER + ) + team = _team_with_permission( + team_id="team-clinical", + members=[Member(user_id="team-admin", role="admin")], + permissions=[], + ) + + assert ( + _user_has_full_team_view( + user_api_key_dict=caller, + team_obj=team, + permission="/team/daily/activity", + is_confirmed_member=True, + ) + is True + ) + + +def test_full_team_view_denied_when_not_a_confirmed_member(): + """Defensive: even if a team granted the permission, a caller who is not a + confirmed member (not in their own teams list) and not an admin is denied.""" + from litellm.proxy.management_endpoints.common_utils import ( + _user_has_full_team_view, + ) + + caller = UserAPIKeyAuth( + user_id="outsider", user_role=LitellmUserRoles.INTERNAL_USER + ) + team = _team_with_permission( + team_id="team-clinical", + members=[Member(user_id="someone-else", role="admin")], + permissions=["/team/daily/activity"], + ) + + assert ( + _user_has_full_team_view( + user_api_key_dict=caller, + team_obj=team, + permission="/team/daily/activity", + is_confirmed_member=False, + ) + is False + )