scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-02 02:11:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

refactor(proxy): answer every team access check with TeamAccess.allows (#43364)

Browse source 
* refactor(proxy): route every team-admin decision through auth/team_access.py

Move the six team-admin helpers out of common_utils, team_endpoints and
key_management_endpoints into litellm/proxy/auth/team_access.py under public
names, and point every management route and helper at them. The key routes
keep checking team admin before org admin, so a team admin whose user row is
gone still passes as before. Status codes and bodies are unchanged, which the
223-case team-admin matrix confirms at the merge base and at the tip

common_utils keeps `_is_user_team_admin` as an alias because the published
litellm-enterprise 0.1.71 wheel still imports it from there

* refactor(proxy): answer every team access check with TeamAccess.allows

Replace the six helpers in auth/team_access.py with one resolver in
litellm/proxy/management/teams/access.py. Each route passes the roles it
accepts (TEAM_OR_ORG_ADMIN or TEAM_ADMIN_ONLY), and /team/update and
/team/info rank roles through strongest_role so org admin still outranks
team admin there

The org lookup moves behind an OrgRoles protocol, implemented by
PrismaOrgRoles in management/users/service.py, and get_team_access in
management/teams/dependencies.py is the only place that reads proxy_server
globals. _check_key_admin_access keeps its name and body from main

Routes that checked org admin first now read the roster first, so a team
admin whose org lookup errors now passes on /team/delete, /team/block,
/team/unblock, member reset_spend and reset_budget, and the team callback
routes. No allowed caller is denied
This commit is contained in:
ryan-crabbe-berri 2026-09-30 15:27:33 -07:00 • committed by GitHub
parent 52b9fa2ba1
commit 632b69b5c8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
38 changed files with 504 additions and 639 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py
View file

@ -22,10 +22,8 @@ from litellm._uuid import uuid
from litellm.proxy._types import *
from litellm.proxy.auth.auth_checks import delete_cached_project_object
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin, # pyright: ignore[reportPrivateUsage] # shared owner of team-admin membership
_set_object_metadata_field,
)
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_endpoints.common_utils import _set_object_metadata_field
from litellm.proxy.management_endpoints.team_admin_field_permissions import team_admin_may_manage_projects
from litellm.proxy.management_helpers.utils import (
management_endpoint_wrapper,
@ -117,7 +115,7 @@ async def _check_user_permission_for_project(
return False
team: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
return _is_user_team_admin(user_api_key_dict, team) or user_api_key_dict.user_id in (team.admins or [])
return is_team_admin(user_api_key_dict, team) or user_api_key_dict.user_id in (team.admins or [])
async def _validate_team_exists(

2
litellm/proxy/_types.py
View file

@ -901,7 +901,7 @@ class LiteLLMRoutes(enum.Enum):
"/team/spend/by_user",
"/team/{team_id}/members/me",
# POST/GET the team's logging callbacks, and DELETE one of them. Every
# handler calls _verify_team_access, which admits only a proxy admin, an
# handler asks TeamAccess.allows for TEAM_OR_ORG_ADMIN: a proxy admin, an
# org admin for the team, or an admin of this team.
#
# team_id is a free-form string, so it spells these with the same path

0
litellm/proxy/management/__init__.py Normal file
View file

0
litellm/proxy/management/teams/__init__.py Normal file
View file

55
litellm/proxy/management/teams/access.py Normal file
View file

@ -0,0 +1,55 @@
"""Who may act on a team: every management route asks ``TeamAccess.allows`` with the roles it accepts."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Final, Literal, NoReturn, Protocol, TypeAlias
from fastapi import HTTPException, status
from litellm.proxy._types import LiteLLM_TeamTable, LitellmUserRoles, UserAPIKeyAuth
TeamRole: TypeAlias = Literal["proxy_admin", "org_admin", "team_admin"]
TEAM_ADMIN_ONLY: Final[frozenset[TeamRole]] = frozenset({"proxy_admin", "team_admin"})
TEAM_OR_ORG_ADMIN: Final[frozenset[TeamRole]] = frozenset({"proxy_admin", "team_admin", "org_admin"})
class OrgRoles(Protocol):
async def is_org_admin(self, user_id: str, organization_id: str) -> bool: ...
@dataclass(frozen=True, slots=True)
class TeamAccess:
org_roles: OrgRoles
async def allows(self, caller: UserAPIKeyAuth, team: LiteLLM_TeamTable, allow: frozenset[TeamRole]) -> bool:
"""Team admin is checked before org admin, so only callers off the roster pay for the org lookup."""
if "proxy_admin" in allow and caller.user_role == LitellmUserRoles.PROXY_ADMIN:
return True
if "team_admin" in allow and is_team_admin(caller, team):
return True
return "org_admin" in allow and await self._is_org_admin(caller, team)
async def strongest_role(self, caller: UserAPIKeyAuth, team: LiteLLM_TeamTable) -> TeamRole | None:
"""Org admin outranks team admin so a caller holding both keeps unrestricted edits."""
if caller.user_role == LitellmUserRoles.PROXY_ADMIN:
return "proxy_admin"
if await self._is_org_admin(caller, team):
return "org_admin"
return "team_admin" if is_team_admin(caller, team) else None
async def _is_org_admin(self, caller: UserAPIKeyAuth, team: LiteLLM_TeamTable) -> bool:
if not caller.user_id or not team.organization_id:
return False
return await self.org_roles.is_org_admin(caller.user_id, team.organization_id)
def is_team_admin(user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable) -> bool:
return any(
member.user_id is not None and member.user_id == user_api_key_dict.user_id and member.role == "admin"
for member in team_obj.members_with_roles
)
def team_access_denied() -> NoReturn:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You do not have access to this team")

10
litellm/proxy/management/teams/dependencies.py Normal file
View file

@ -0,0 +1,10 @@
from __future__ import annotations
from litellm.proxy.management.teams.access import TeamAccess
from litellm.proxy.management.users.service import PrismaOrgRoles
def get_team_access() -> TeamAccess:
from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache
return TeamAccess(org_roles=PrismaOrgRoles(prisma_client, user_api_key_cache, proxy_logging_obj))

0
litellm/proxy/management/users/__init__.py Normal file
View file

36
litellm/proxy/management/users/service.py Normal file
View file

@ -0,0 +1,36 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Final
from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles
if TYPE_CHECKING:
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.utils import PrismaClient, ProxyLogging
def holds_org_admin(user: LiteLLM_UserTable | None, organization_id: str) -> bool:
return user is not None and any(
membership.organization_id == organization_id and membership.user_role == LitellmUserRoles.ORG_ADMIN.value
for membership in user.organization_memberships or []
)
@dataclass(frozen=True, slots=True)
class PrismaOrgRoles:
prisma_client: PrismaClient | None
user_api_key_cache: UserApiKeyCache
proxy_logging_obj: ProxyLogging
async def is_org_admin(self, user_id: str, organization_id: str) -> bool:
from litellm.proxy.auth.auth_checks import get_user_object
user: Final = await get_user_object(
user_id=user_id,
prisma_client=self.prisma_client,
user_api_key_cache=self.user_api_key_cache,
user_id_upsert=False,
proxy_logging_obj=self.proxy_logging_obj,
)
return holds_org_admin(user, organization_id)

6
litellm/proxy/management_endpoints/auto_router_endpoints.py
View file

@ -41,9 +41,7 @@ from litellm.proxy.litellm_pre_call_utils import (
LiteLLMProxyRequestSetup,
refresh_proxy_server_request_body_snapshot,
)
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin, # pyright: ignore[reportPrivateUsage] # shared owner of team-admin membership
)
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_helpers.auto_router_permissions import (
authorize_member_auto_router_dependencies,
authorize_member_auto_router_team,
@ -249,7 +247,7 @@ async def _authorize_router_dry_run(user_api_key_dict: UserAPIKeyAuth, team_id:
)
team: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team):
if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team):
ModelManagementAuthChecks.can_user_make_team_model_call(
team_id=team_id,
user_api_key_dict=user_api_key_dict,

51
litellm/proxy/management_endpoints/common_utils.py
View file

@ -61,6 +61,7 @@ from litellm.proxy._types import ( # noqa: F401 re-exported
user_api_key_has_admin_view as _user_has_admin_view,
)
from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.utils import _premium_user_check
from litellm.repositories.team_repository import TeamRepository
from litellm.types.utils import BudgetConfig
@ -69,6 +70,9 @@ if TYPE_CHECKING:
from litellm.proxy._types import NewProjectRequest, UpdateProjectRequest
from litellm.proxy.utils import PrismaClient, ProxyLogging
# TODO: drop once the litellm-enterprise pin moves past 0.1.71, which imports this name
_is_user_team_admin: Final = is_team_admin
def validate_team_model_max_budget(
model_max_budget: Mapping[str, BudgetConfig] | None,
@ -201,49 +205,6 @@ def _check_disable_global_guardrails_caller_permission(
)
def _is_user_team_admin(user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable) -> bool:
for member in team_obj.members_with_roles:
if (member.user_id is not None and member.user_id == user_api_key_dict.user_id) and member.role == "admin":
return True
return False
async def _is_user_org_admin_for_team(user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable) -> bool:
"""
Check if user is an org admin for the team's organization.
Returns True if:
- The team belongs to an organization, AND
- The user has org_admin role in that organization
"""
if not team_obj.organization_id or not user_api_key_dict.user_id:
return False
from litellm.proxy.auth.auth_checks import get_user_object
from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
caller_user: Final = await get_user_object(
user_id=user_api_key_dict.user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
proxy_logging_obj=proxy_logging_obj,
)
if caller_user is None:
return False
for m in caller_user.organization_memberships or []:
if m.organization_id == team_obj.organization_id and m.user_role == LitellmUserRoles.ORG_ADMIN.value:
return True
return False
def _team_member_has_permission(
user_api_key_dict: UserAPIKeyAuth,
team_obj: LiteLLM_TeamTable,
@ -315,7 +276,7 @@ async def _user_has_admin_privileges(
for team in teams:
team_obj = LiteLLM_TeamTable.model_validate(team.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return True
except Exception as e:
@ -384,7 +345,7 @@ async def _team_admin_can_invite_user(
admin_team_ids: Final = [
team.team_id
for team in teams
if _is_user_team_admin(
if is_team_admin(
user_api_key_dict=user_api_key_dict,
team_obj=LiteLLM_TeamTable.model_validate(team.model_dump()),
)

8
litellm/proxy/management_endpoints/internal_user_endpoints.py
View file

@ -51,13 +51,13 @@ from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
from litellm.proxy.hooks.key_management_event_hooks import KeyManagementEventHooks
from litellm.proxy.hooks.model_max_budget_limiter import build_model_max_budget_usage
from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_endpoints.common_daily_activity import (
DailySpendRecord,
get_daily_activity,
get_daily_activity_aggregated,
)
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_user_has_admin_view,
require_caller_user_id_for_non_admin,
validate_budget_duration,
@ -1052,7 +1052,7 @@ async def _check_user_info_v2_access(
teams: Final = await _team_table(prisma_client).find_many(where={"team_id": {"in": caller_user.teams}})
for team in teams:
team_obj = LiteLLM_TeamTable.model_validate(team.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
# Check if target user is in this team
if team.team_id in (target_user.teams or []):
return target_user
@ -2714,8 +2714,6 @@ async def _resolve_team_org_filter(
proxy_logging_obj: "ProxyLogging | None",
) -> list[str]:
"""Look up the team and return its org as a filter list, or raise 403."""
from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin
try:
team_obj: Final = await get_team_object(
team_id=team_id,
@ -2729,7 +2727,7 @@ async def _resolve_team_org_filter(
detail={"error": f"scope_user_search_to_org is enabled but team '{team_id}' was not found."},
)
if not _is_user_team_admin(user_api_key_dict, team_obj):
if not is_team_admin(user_api_key_dict, team_obj):
raise HTTPException(
status_code=403,
detail={"error": "scope_user_search_to_org is enabled. You must be an admin of this team to search users."},

39
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -85,11 +85,11 @@ from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.hooks.key_management_event_hooks import KeyManagementEventHooks
from litellm.proxy.hooks.model_max_budget_limiter import build_model_max_budget_usage
from litellm.proxy.management.teams.access import TEAM_ADMIN_ONLY, TEAM_OR_ORG_ADMIN, is_team_admin
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.common_utils import (
_check_disable_global_guardrails_caller_permission,
_check_passthrough_routes_caller_permission,
_is_user_org_admin_for_team,
_is_user_team_admin,
_set_object_metadata_field,
_team_member_has_permission,
_user_has_admin_view,
@ -3053,7 +3053,7 @@ async def _acting_as_team_admin_for_key_update(
user_api_key_cache=user_api_key_cache,
check_db_only=True,
)
if not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_for_grant):
if not is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_for_grant):
return False
team_admin_key_request_or_raise(
team_admin_key_edit_verdict(
@ -4056,17 +4056,11 @@ async def validate_key_team_change(
)
# Check if the person initiating the change is a Proxy Admin or Team Admin
if (
change_initiated_by.user_role == LitellmUserRoles.PROXY_ADMIN.value
or _is_user_team_admin(
user_api_key_dict=change_initiated_by,
team_obj=team,
)
or TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint(
team_member_role=None if member_object is None else member_object.role,
team_table=team_table,
route=KeyManagementRoutes.KEY_UPDATE.value,
)
initiator_is_admin: Final = await get_team_access().allows(change_initiated_by, team, TEAM_ADMIN_ONLY)
if initiator_is_admin or TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint(
team_member_role=None if member_object is None else member_object.role,
team_table=team_table,
route=KeyManagementRoutes.KEY_UPDATE.value,
):
return
else:
@ -4952,7 +4946,7 @@ async def can_modify_verification_token(
return False
# Check if user is team admin
if _is_user_team_admin(
if is_team_admin(
user_api_key_dict=user_api_key_dict,
team_obj=team_table,
):
@ -6013,7 +6007,7 @@ async def _check_proxy_or_team_admin_for_key(
check_db_only=True,
)
if team_table is not None:
if _is_user_team_admin(
if is_team_admin(
user_api_key_dict=user_api_key_dict,
team_obj=team_table,
):
@ -6409,9 +6403,7 @@ def _get_admin_team_ids_from_objects(
team_objects: list[LiteLLM_TeamTable],
) -> list[str]:
"""Filter team objects to those where the user is an admin."""
return [
team.team_id for team in team_objects if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)
]
return [team.team_id for team in team_objects if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)]
def _get_team_ids_with_key_list_permission_from_objects(
@ -6425,7 +6417,7 @@ def _get_team_ids_with_key_list_permission_from_objects(
return [
team.team_id
for team in team_objects
if not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)
if not is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)
and _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
team_obj=team,
@ -7285,11 +7277,8 @@ async def _check_key_admin_access(
user_api_key_cache=user_api_key_cache,
check_db_only=True,
)
if team_obj is not None:
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return
if team_obj is not None and await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN):
return
raise HTTPException(
status_code=403,

12
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -68,7 +68,8 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import (
)
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.db.routing_prisma_wrapper import WriterPinnedClient
from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin
from litellm.proxy.management.teams.access import TEAM_ADMIN_ONLY, is_team_admin
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.team_endpoints import (
_refresh_cached_team,
append_team_models,
@ -2004,7 +2005,7 @@ class ModelManagementAuthChecks:
)
if user_api_key_dict.user_role and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return True
elif team_obj is None or not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
elif team_obj is None or not is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
raise HTTPException(
status_code=403,
detail={
@ -2133,11 +2134,8 @@ class ModelManagementAuthChecks:
)
team_obj: Final = LiteLLM_TeamTable.model_validate(team_obj_row.model_dump())
if (
member_operation is not None
and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
):
caller_is_admin: Final = await get_team_access().allows(user_api_key_dict, team_obj, TEAM_ADMIN_ONLY)
if member_operation is not None and not caller_is_admin:
from litellm.proxy.proxy_server import llm_router
if llm_router is None or (member_operation == "update" and incoming_model_params is None):

43
litellm/proxy/management_endpoints/team_callback_endpoints.py
View file

@ -44,10 +44,9 @@ from litellm.proxy.litellm_pre_call_utils import (
_get_validated_callback_metadata,
convert_key_logging_metadata_to_callback,
)
from litellm.proxy.management_endpoints.team_endpoints import (
_refresh_cached_team,
_verify_team_access,
)
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN, team_access_denied
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.team_endpoints import _refresh_cached_team
from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
from litellm.repositories.team_repository import TeamRepository
@ -239,9 +238,9 @@ def _unknown_team_error(team_id: str, user_api_key_dict: UserAPIKeyAuth, status_
"""Report an unknown team without telling an unauthorized caller that it is unknown.
These routes are reachable by any authenticated caller so that a team admin can
get as far as _verify_team_access. A distinct "does not exist" would therefore let
get as far as the team access check. A distinct "does not exist" would therefore let
any valid key probe which team ids exist, so a caller who could not have managed
the team either way gets the same 403 body _verify_team_access raises.
the team either way gets the same 403 body team_access_denied raises.
"""
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return _callback_error(status_code, f"Team id = {team_id} does not exist.")
@ -332,10 +331,10 @@ async def add_team_callbacks(
# team may write callback credentials. Without this, any
# authenticated key holder could overwrite another team's logging
# config (and read back the credentials they wrote).
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable(**_existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
_validate_team_callback(data)
@ -501,10 +500,10 @@ async def delete_team_callback(
# IDOR guard: only proxy admins / org admins / team admins of THIS team may
# deregister its callbacks, otherwise any authenticated key holder could
# silence another team's observability integration.
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable(**_existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
team_metadata: Final = _existing_team.metadata
registered_callbacks: Final = team_metadata.get("logging")
@ -634,10 +633,10 @@ async def disable_team_logging(
# IDOR guard: only proxy admins / org admins / team admins of THIS
# team may disable its logging — otherwise any authenticated key
# holder can silence audit logging for any team.
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable(**_existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
# Update team metadata to disable logging
team_metadata = _existing_team.metadata
@ -775,10 +774,10 @@ async def get_team_callbacks(
# IDOR guard: callback metadata holds third-party API credentials
# (Langfuse / Langsmith / GCS). Only proxy admins / org admins /
# team admins of THIS team may read them.
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable(**_existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
team_callback_settings_obj: Final = _resolve_team_callbacks(_existing_team.metadata)

132
litellm/proxy/management_endpoints/team_endpoints.py
View file

@ -27,7 +27,6 @@ from typing import (
NamedTuple,
NoReturn,
Protocol,
TypeAlias,
TypeVar,
cast,
)
@ -123,14 +122,14 @@ from litellm.proxy.hooks.model_max_budget_limiter import (
build_model_max_budget_usage,
resolve_model_budget,
)
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN, TeamRole, is_team_admin, team_access_denied
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.common_daily_activity import (
get_daily_activity_aggregated,
)
from litellm.proxy.management_endpoints.common_utils import (
_check_disable_global_guardrails_caller_permission,
_check_passthrough_routes_caller_permission,
_is_user_org_admin_for_team,
_is_user_team_admin,
_set_object_metadata_field,
_team_member_has_permission,
_update_metadata_fields,
@ -478,45 +477,6 @@ async def _refresh_cached_team(
)
TeamAccessRole: TypeAlias = Literal["proxy_admin", "org_admin", "team_admin"]
def _raise_team_access_denied() -> NoReturn:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have access to this team",
)
async def _resolve_team_access(
team_obj: LiteLLM_TeamTable,
user_api_key_dict: UserAPIKeyAuth,
) -> TeamAccessRole | None:
"""Strongest role the caller holds over ``team_obj``, or None when they hold none.
Org admin outranks team admin so a caller holding both keeps unrestricted edits.
"""
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return "proxy_admin"
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return "org_admin"
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return "team_admin"
return None
async def _verify_team_access(
team_obj: LiteLLM_TeamTable,
user_api_key_dict: UserAPIKeyAuth,
) -> None:
"""Raise 403 unless the caller is a proxy admin, an org admin for the team's org, or a team admin."""
if await _resolve_team_access(team_obj=team_obj, user_api_key_dict=user_api_key_dict) is None:
_raise_team_access_denied()
_GENERAL_SETTINGS: Final = TypeAdapter(dict[str, object])
@ -526,7 +486,7 @@ def _general_settings() -> Mapping[str, object]:
return _GENERAL_SETTINGS.validate_python(general_settings)
def _caller_edit_access(role: TeamAccessRole | None, general_settings: Mapping[str, object]) -> TeamEditAccess:
def _caller_edit_access(role: TeamRole | None, general_settings: Mapping[str, object]) -> TeamEditAccess:
"""What the caller may change on /team/update, reported on /team/info so the dashboard never re-derives it."""
match role:
case "proxy_admin" | "org_admin":
@ -1160,7 +1120,7 @@ async def _check_user_team_limits(
Only used by /team/new for standalone teams (organization_id is None).
/team/update does NOT call this — an existing team's admin is already
authorized via _verify_team_access() and is not gated by their personal
authorized via the team access check and is not gated by their personal
wallet. Org-scoped teams use _check_org_team_limits() instead.
"""
# Validate team budget against user's max_budget
@ -2277,16 +2237,16 @@ async def update_team(
# Non-proxy-admins get the same 403 as an access denial so /team/update
# cannot be used to probe which team ids exist
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
_raise_team_access_denied()
team_access_denied()
raise HTTPException(
status_code=404,
detail={"error": f"Team not found, passed team_id={data.team_id}"},
)
existing_team: Final = LiteLLM_TeamTable.model_validate(existing_team_row.model_dump())
access_role: Final = await _resolve_team_access(team_obj=existing_team, user_api_key_dict=user_api_key_dict)
access_role: Final = await get_team_access().strongest_role(user_api_key_dict, existing_team)
if access_role is None:
_raise_team_access_denied()
team_access_denied()
if access_role == "team_admin":
data = team_admin_request_or_raise( # rebind-ok: resent values must not reach the derived writes below
team_admin_edit_verdict(
@ -2354,7 +2314,7 @@ async def update_team(
if data.organization_id is not None and len(data.organization_id) > 0: # allow unsetting the organization_id
# If the caller is relocating the team to a different org, they
# must also be PROXY_ADMIN or an org-admin of the DESTINATION org.
# _verify_team_access above only checked the team's CURRENT org,
# the team access check above only covered the team's CURRENT org,
# so without this gate an org-admin could hand their team to any
# other org (or capture a team from another org they once
# administered into a new destination).
@ -2833,11 +2793,7 @@ async def _validate_team_member_add_permissions(
the request matches the caller's own ``user_id`` and is being
added with ``role="user"``.
"""
if getattr(user_api_key_dict, "user_role", None) == LitellmUserRoles.PROXY_ADMIN.value:
return
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data):
return
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data):
if await get_team_access().allows(user_api_key_dict, complete_team_data, TEAM_OR_ORG_ADMIN):
return
if not _is_available_team(
@ -3649,11 +3605,7 @@ async def _team_member_delete(
## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
):
if not await get_team_access().allows(user_api_key_dict, existing_team_row, TEAM_OR_ORG_ADMIN):
raise HTTPException(
status_code=403,
detail={
@ -3853,11 +3805,7 @@ async def team_member_update(
## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
):
if not await get_team_access().allows(user_api_key_dict, existing_team_row, TEAM_OR_ORG_ADMIN):
raise HTTPException(
status_code=403,
detail={
@ -3966,7 +3914,7 @@ async def team_member_update(
def _check_not_resetting_own_spend(user_id: str, user_api_key_dict: UserAPIKeyAuth) -> None:
"""
_verify_team_access authorizes a team admin (or org admin) over their own
The team access check authorizes a team admin (or org admin) over their own
team, with no check that the target user_id differs from the caller. Left
unchecked, that admin could target their own LiteLLM_TeamMembership row and
repeatedly reset it to 0 right before it crosses their per-member cap,
@ -4045,7 +3993,8 @@ async def reset_team_member_spend_fn(
proxy_logging_obj=proxy_logging_obj,
check_db_only=True,
)
await _verify_team_access(team_obj=team_obj, user_api_key_dict=user_api_key_dict)
if not await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN):
team_access_denied()
_check_not_resetting_own_spend(user_id=user_id, user_api_key_dict=user_api_key_dict)
membership_where: Final = { # mutable-ok: prisma client requires a plain dict where= argument
@ -4140,7 +4089,8 @@ async def reset_team_member_budget_fn(
proxy_logging_obj=proxy_logging_obj,
check_db_only=True,
)
await _verify_team_access(team_obj=team_obj, user_api_key_dict=user_api_key_dict)
if not await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN):
team_access_denied()
membership_where: Final = { # mutable-ok: prisma client requires a plain dict where= argument
"user_id_team_id": {"user_id": user_id, "team_id": team_id} # mutable-ok: same prisma where= argument
@ -4418,10 +4368,8 @@ async def delete_team(
team_row_pydantic = LiteLLM_TeamTable.model_validate(team_row_base.model_dump())
# Verify caller has access to manage this team
await _verify_team_access(
team_obj=team_row_pydantic,
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(user_api_key_dict, team_row_pydantic, TEAM_OR_ORG_ADMIN):
team_access_denied()
team_rows.append(team_row_pydantic)
@ -4799,7 +4747,7 @@ async def validate_membership(user_api_key_dict: UserAPIKeyAuth, team_table: Lit
return
# Check if user is an org admin for the team's organization
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_table):
if await get_team_access().allows(user_api_key_dict, team_table, TEAM_OR_ORG_ADMIN):
return
raise HTTPException(
@ -4955,7 +4903,7 @@ async def team_info(
)
team_table: Final = LiteLLM_TeamTable.model_validate(team_info.model_dump())
await validate_membership(user_api_key_dict=user_api_key_dict, team_table=team_table)
access_role: Final = await _resolve_team_access(team_obj=team_table, user_api_key_dict=user_api_key_dict)
access_role: Final = await get_team_access().strongest_role(user_api_key_dict, team_table)
organization_models: Final[list[str] | None] = (
_parent_organization_models(team_info) if access_role is not None else None
)
@ -5228,10 +5176,10 @@ async def block_team(
)
# Verify caller has access to manage this team
await _verify_team_access(
team_obj=LiteLLM_TeamTable.model_validate(existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable.model_validate(existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
record: Final = await _team_db(prisma_client).update(
where={"team_id": data.team_id},
@ -5277,10 +5225,10 @@ async def unblock_team(
)
# Verify caller has access to manage this team
await _verify_team_access(
team_obj=LiteLLM_TeamTable.model_validate(existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
if not await get_team_access().allows(
user_api_key_dict, LiteLLM_TeamTable.model_validate(existing_team.model_dump()), TEAM_OR_ORG_ADMIN
):
team_access_denied()
record: Final = await _team_db(prisma_client).update(
where={"team_id": data.team_id},
@ -6161,11 +6109,7 @@ async def team_model_add(
team_obj: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
# Authorization check - only proxy admin, team admin, or org admin can add models
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
):
if not await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN):
raise HTTPException(
status_code=403,
detail={"error": "Only proxy admin or team admin can modify team models"},
@ -6281,11 +6225,7 @@ async def team_model_delete(
team_obj: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
# Authorization check - only proxy admin, team admin, or org admin can remove models
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
):
if not await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN):
raise HTTPException(
status_code=403,
detail={"error": "Only proxy admin or team admin can modify team models"},
@ -6358,8 +6298,7 @@ async def team_member_permissions(
if (
hasattr(user_api_key_dict, "user_role")
and not _user_has_admin_view(user_api_key_dict)
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
and not await get_team_access().allows(user_api_key_dict, complete_team_data, TEAM_OR_ORG_ADMIN)
and not _is_available_team(
team_id=complete_team_data.team_id,
user_api_key_dict=user_api_key_dict,
@ -6422,12 +6361,7 @@ async def update_team_member_permissions(
# Available-team self-join must NOT grant write access to team-wide
# permission policies; only proxy/team/org admins can update them.
if (
hasattr(user_api_key_dict, "user_role")
and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
):
if not await get_team_access().allows(user_api_key_dict, complete_team_data, TEAM_OR_ORG_ADMIN):
raise HTTPException(
status_code=403,
detail={
@ -6665,7 +6599,7 @@ async def _resolve_team_daily_activity_scope(
has_full_team_view = True
for team_alias in team_aliases:
team_obj = LiteLLM_TeamTable.model_validate(team_alias.model_dump())
is_admin = _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
is_admin = is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
has_perm = _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
team_obj=team_obj,

11
litellm/proxy/management_helpers/bulk_team_member_budgets.py
View file

@ -17,16 +17,15 @@ from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy._types import (
LiteLLM_TeamTable,
LitellmTableNames,
LitellmUserRoles,
Member,
UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_checks import invalidate_team_member_spend_state
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.db.routing_prisma_wrapper import WriterPinnedClient
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team, # pyright: ignore[reportPrivateUsage] # same check /team/member_update uses
_is_user_team_admin, # pyright: ignore[reportPrivateUsage] # same check /team/member_update uses
_upsert_budget_and_membership, # pyright: ignore[reportPrivateUsage] # the single-member write, shared so the two surfaces cannot drift
member_budget_patch,
)
@ -180,11 +179,7 @@ async def bulk_update_team_member_budgets(
if team is None:
raise _team_not_found(team_id)
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team)
):
if not await get_team_access().allows(user_api_key_dict, team, TEAM_OR_ORG_ADMIN):
raise _forbidden(
"Call not allowed. User not proxy admin OR team admin OR org admin for this team. "
f"route='/management/v1/teams/{team_id}/members/bulk_update'"

14
litellm/proxy/management_helpers/bulk_user_creation.py
View file

@ -34,11 +34,9 @@ from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
from litellm.proxy.list_api.common import PROBLEM_TYPE_BASE, ManagementProblem
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team, # pyright: ignore[reportPrivateUsage] # same team-admin check /user/new uses
_is_user_team_admin, # pyright: ignore[reportPrivateUsage] # same team-admin check /user/new uses
validate_budget_duration,
)
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.common_utils import validate_budget_duration
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_update_internal_new_user_params, # pyright: ignore[reportPrivateUsage, reportUnknownVariableType] # /user/new defaults; result validated below
check_if_default_team_set,
@ -292,11 +290,7 @@ async def _load_teams(prisma_client: PrismaClient, team_ids: frozenset[str]) ->
async def _team_permission_error(team: LiteLLM_TeamTable, user_api_key_dict: UserAPIKeyAuth) -> str | None:
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
return None
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team):
return None
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team):
if await get_team_access().allows(user_api_key_dict, team, TEAM_OR_ORG_ADMIN):
return None
return f"Call not allowed. User not proxy admin OR team admin. team_id={team.team_id}"

12
litellm/proxy/management_helpers/bulk_user_deletion.py
View file

@ -34,10 +34,8 @@ from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.hooks.key_management_event_hooks import KeyManagementEventHooks
from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
from litellm.proxy.list_api.common import PROBLEM_TYPE_BASE, ManagementProblem
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team, # pyright: ignore[reportPrivateUsage] # same check /team/member_delete uses
_is_user_team_admin, # pyright: ignore[reportPrivateUsage] # same check /team/member_delete uses
)
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.key_management_endpoints import (
_persist_deleted_verification_tokens, # pyright: ignore[reportPrivateUsage] # same audit path /key/delete uses
)
@ -324,11 +322,7 @@ async def bulk_remove_team_members(
if team is None:
raise _team_not_found(team_id)
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team)
and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team)
):
if not await get_team_access().allows(user_api_key_dict, team, TEAM_OR_ORG_ADMIN):
raise _forbidden(
"Call not allowed. User not proxy admin OR team admin OR org admin for this team. "
f"route='/management/v1/teams/{team_id}/members/bulk_delete'"

26
litellm/proxy/memory/memory_endpoints.py
View file

@ -32,6 +32,8 @@ from litellm.proxy._types import (
user_api_key_has_admin_view,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.repositories.prisma_protocols import TableActions
from litellm.repositories.table_repositories import MemoryRepository
from litellm.repositories.team_repository import TeamRepository
@ -200,17 +202,9 @@ async def _assert_write_access(
async def _is_team_admin_for(prisma_client: "PrismaClient", user_api_key_dict: UserAPIKeyAuth, team_id: str) -> bool:
"""
True if the caller is a team admin of `team_id`, or an org admin for the
team's organization. Mirrors the auth pattern used by team-management
endpoints (`_is_user_team_admin` + `_is_user_org_admin_for_team`).
Imported lazily to avoid a circular import with proxy_server during the
memory router's module load.
team's organization, asked through the same ``TeamAccess.allows`` the
team-management endpoints use.
"""
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
_is_user_team_admin,
)
try:
team_obj: Final = await TeamRepository(prisma_client).find_by_id(team_id, id_field="team_id")
except Exception as e:
@ -219,19 +213,11 @@ async def _is_team_admin_for(prisma_client: "PrismaClient", user_api_key_dict: U
if team_obj is None:
return False
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return True
# Org-admin path is best-effort: it pulls from the user cache via
# `get_user_object` which depends on the proxy_server module being
# initialized. In tests / non-proxy contexts that import path may fail —
# treat any error as "not an org admin" rather than crashing the request.
try:
if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return True
return await get_team_access().allows(user_api_key_dict, team_obj, TEAM_OR_ORG_ADMIN)
except Exception as e:
verbose_proxy_logger.debug("Org-admin check skipped during write-auth (team_id=%s): %s", team_id, e)
return False
return False
def _is_unique_violation(exc: Exception) -> bool:

16
litellm/proxy/spend_tracking/spend_management_endpoints.py
View file

@ -4848,10 +4848,8 @@ async def _can_team_member_view_log(
Returns True if the team exists and the user is either a team admin or
a team member with the ``/spend/logs`` permission.
"""
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_team_member_has_permission,
)
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_endpoints.common_utils import _team_member_has_permission
if team_id is None:
return False
@ -4859,7 +4857,7 @@ async def _can_team_member_view_log(
if team_row is None:
return False
team_obj: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return True
return _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
@ -5078,10 +5076,8 @@ async def _get_permitted_team_ids_for_spend_logs(
"""
# Imported here to avoid circular import: proxy_server imports this module.
from litellm.proxy.auth.auth_checks import get_user_object
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_team_member_has_permission,
)
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_endpoints.common_utils import _team_member_has_permission
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
user_obj: Final = await get_user_object(
@ -5099,7 +5095,7 @@ async def _get_permitted_team_ids_for_spend_logs(
permitted: Final[list[str]] = []
for team_row in team_rows:
team_obj = LiteLLM_TeamTable.model_validate(team_row.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj) or _team_member_has_permission(
if is_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj) or _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
team_obj=team_obj,
permission=KeyManagementRoutes.SPEND_LOGS.value,

2
tests/proxy_behavior/management/test_team_block_unblock.py
View file

@ -6,7 +6,7 @@ from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/block + /team/unblock. The handler gate is _verify_team_access
# POST /team/block + /team/unblock. The handler gate is TeamAccess.allows
# (proxy admin / team admin / org admin), but the management-route gate fronts
# it: the request carries the team's organization_id so an org admin of that
# org clears the gate's org-scoped branch. A team admin is an INTERNAL_USER

2
tests/proxy_behavior/management/test_team_delete.py
View file

@ -6,7 +6,7 @@ from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/delete runs per-team _verify_team_access. The request carries the
# POST /team/delete asks TeamAccess.allows per team. The request carries the
# team's organization_id so an org admin of that org clears the management-
# route gate; a team admin is an INTERNAL_USER on a non-internal_user route,
# so a team admin never reaches the handler. Only PROXY_ADMIN and an org admin

2
tests/proxy_behavior/management/test_team_info.py
View file

@ -70,7 +70,7 @@ async def test_team_info_authz_matrix(
assert body["team_info"]["team_id"] == target_team_id
# Phase 4 F6 — explicit pin on the `_verify_team_access` 403 message string.
# Phase 4 F6 — explicit pin on the `team_access_denied` 403 message string.
# alpha/org_b_admin already covers the branch in the matrix; this guard
# turns a silent rename of the exception detail into a CI red, which is the
# behavior tripwire that the matrix's status-only assertion cannot catch.

2
tests/proxy_behavior/management/test_team_member_reset_spend.py
View file

@ -12,7 +12,7 @@ _RESET_TO = 2.0
# POST /team/{team_id}/member/{user_id}/reset_spend. The handler gate is
# _verify_team_access (proxy admin / team admin of this team / org admin of
# TeamAccess.allows (proxy admin / team admin of this team / org admin of
# the team's org) — the same gate /team/member_update uses, so this mirrors
# that file's matrix exactly.
_MATRIX = [

4
tests/proxy_behavior/management/test_team_update.py
View file

@ -12,7 +12,7 @@ pytestmark = pytest.mark.asyncio(loop_scope="session")
# The route is self-managed (LIT-5722), so every authenticated caller reaches
# update_team and denials are the handler's 403, never the route gate's 401.
# Only PROXY_ADMIN and an ORG_ADMIN of the team's org pass: a team admin is
# admitted by _resolve_team_access but then refused because no team field is
# admitted by TeamAccess.strongest_role but then refused because no team field is
# enabled for team admins (team_admin_editable_team_fields defaults to empty).
MARKER_ALIAS = "behavior-pin-update-marker-alias"
@ -191,7 +191,7 @@ async def test_team_update_org_relocation_gate(
assert row.organization_id == world.org_a_id, "denied but team relocated"
# Phase 4 F6 — explicit pin on the `_verify_team_access` 403 detail string
# Phase 4 F6 — explicit pin on the `team_access_denied` 403 detail string
# when an org_admin clears the destination route gate but fails the source
# team's org-membership check. The relocation matrix above covers the
# status; this guard turns a silent rename of the helper's exception detail

6
tests/test_litellm/proxy/auth/test_route_checks.py
View file

@ -3043,7 +3043,7 @@ def test_team_update_gate_admits_internal_user_without_org_context(): # test-qu
def test_team_update_gate_defers_cross_org_admin_to_the_handler(): # test-quality-ok: the gate's only success signal is not raising; the handler's 403 it defers to is pinned in test_team_endpoints
"""An org admin of a DIFFERENT org clears the coarse gate like any internal user;
update_team's _resolve_team_access finds no role on the team and 403s (pinned in
update_team's TeamAccess.strongest_role finds no role on the team and 403s (pinned in
test_team_endpoints), so there is still no cross-org escalation."""
user_obj = _make_org_admin_user("org-1")
valid_token = UserAPIKeyAuth(user_id="org-admin-user", user_role=LitellmUserRoles.INTERNAL_USER.value)
@ -4019,8 +4019,8 @@ def test_team_callback_routes_reach_their_handler_for_non_admins(route, role):
"""A team admin manages their own team's logging callbacks, so the route gate
must let a non-proxy-admin through to the handler.
The handler is what authorizes: every team callback endpoint calls
_verify_team_access, which admits only a proxy admin, an org admin for the
The handler is what authorizes: every team callback endpoint asks
TeamAccess.allows, which admits only a proxy admin, an org admin for the
team, or an admin of that team, and 403s everyone else. Before this, the gate
rejected the team admin with a 401 naming proxy admin, so the handler's own
check was unreachable for them.

2
tests/test_litellm/proxy/management_endpoints/test_activity_tenant_scoping.py
View file

@ -27,7 +27,7 @@ from litellm.proxy.agent_endpoints.auth.agent_permission_handler import (
def _make_team(team_id: str, admin_user_ids: list):
"""Build a Prisma-compatible team row. `admin_user_ids` are inserted as
`members_with_roles[*].role == "admin"` because that's what
`_is_user_team_admin` checks."""
`is_team_admin` checks."""
members_with_roles = [{"user_id": uid, "role": "admin"} for uid in admin_user_ids]
row = MagicMock()
row.team_id = team_id

92
tests/test_litellm/proxy/management_endpoints/test_common_utils.py
View file

@ -25,7 +25,6 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_org_admin_can_invite_user,
_set_object_metadata_field,
_team_admin_can_invite_user,
@ -246,53 +245,12 @@ class TestUserHasAdminView:
assert _user_has_admin_view(auth_user) is False
class TestIsUserTeamAdmin:
"""Tests for _is_user_team_admin function."""
def test_published_enterprise_import_of_team_admin_check_still_answers():
from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin
@pytest.mark.parametrize(
"members_with_roles,user_id,expected",
[
(
[Member(user_id="u1", role="admin")],
"u1",
True,
),
(
[Member(user_id="u1", role="user")],
"u1",
False,
),
(
[
Member(user_id="u2", role="admin"),
Member(user_id="u1", role="admin"),
],
"u1",
True,
),
([], "u1", False),
],
)
def test_is_user_team_admin_parametrized(
self, members_with_roles, user_id, expected
):
"""Parametrized test: user is team admin only when in members_with_roles with admin role."""
mock_auth = MagicMock()
mock_auth.user_id = user_id
team = LiteLLM_TeamTable(
team_id="team-1",
members_with_roles=members_with_roles,
)
assert _is_user_team_admin(mock_auth, team) == expected
def test_is_user_team_admin_user_not_in_team(self):
"""Test returns False when user is not in team members."""
auth = UserAPIKeyAuth(user_id="u99", api_key="sk-x", user_role=None)
team = LiteLLM_TeamTable(
team_id="team-1",
members_with_roles=[Member(user_id="u1", role="admin")],
)
assert _is_user_team_admin(auth, team) is False
team = LiteLLM_TeamTable(team_id="t1", members_with_roles=[Member(user_id="admin", role="admin")])
assert _is_user_team_admin(UserAPIKeyAuth(user_id="admin"), team) is True
assert _is_user_team_admin(UserAPIKeyAuth(user_id="outsider"), team) is False
class TestOrgAdminCanInviteUser:
@ -903,46 +861,6 @@ class TestCheckDisableGlobalGuardrailsCallerPermission:
)
class TestIsUserOrgAdminForTeam:
"""The caller must be looked up with its exact identity; a nulled or omitted
lookup argument would silently mis-resolve org-admin status."""
@pytest.mark.asyncio
async def test_get_user_object_called_with_caller_identity(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = LiteLLM_TeamTable(
team_id="t1", organization_id="org1", members_with_roles=[]
)
key = UserAPIKeyAuth(
user_id="u1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER
)
fake_prisma, fake_cache, fake_logging = MagicMock(), MagicMock(), MagicMock()
mock_get_user = AsyncMock(return_value=None)
with patch(
"litellm.proxy.proxy_server.prisma_client", fake_prisma
), patch(
"litellm.proxy.proxy_server.user_api_key_cache", fake_cache
), patch(
"litellm.proxy.proxy_server.proxy_logging_obj", fake_logging
), patch(
"litellm.proxy.auth.auth_checks.get_user_object", mock_get_user
):
result = await _is_user_org_admin_for_team(key, team)
assert result is False
mock_get_user.assert_awaited_once_with(
user_id="u1",
prisma_client=fake_prisma,
user_api_key_cache=fake_cache,
user_id_upsert=False,
proxy_logging_obj=fake_logging,
)
class TestTeamMemberHasPermission:
def test_requires_caller_to_be_a_team_member(self):
from litellm.proxy.management_endpoints.common_utils import (

39
tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py
View file

@ -3377,30 +3377,25 @@ async def test_validate_key_team_change_with_member_permissions():
"litellm.proxy.management_endpoints.key_management_endpoints._get_user_in_team"
) as mock_get_user:
with patch(
"litellm.proxy.management_endpoints.key_management_endpoints._is_user_team_admin"
) as mock_is_admin:
with patch(
"litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint"
) as mock_has_perms:
"litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint"
) as mock_has_perms:
mock_get_user.return_value = mock_member_object
mock_has_perms.return_value = True
mock_get_user.return_value = mock_member_object
mock_is_admin.return_value = False
mock_has_perms.return_value = True
# This should not raise an exception due to member permissions
await validate_key_team_change(
key=mock_key,
team=mock_team,
change_initiated_by=mock_change_initiator,
llm_router=mock_router,
)
# This should not raise an exception due to member permissions
await validate_key_team_change(
key=mock_key,
team=mock_team,
change_initiated_by=mock_change_initiator,
llm_router=mock_router,
)
# Verify the permission check was called with correct parameters
mock_has_perms.assert_called_once_with(
team_member_role=mock_member_object.role,
team_table=mock_team,
route=KeyManagementRoutes.KEY_UPDATE.value,
)
# Verify the permission check was called with correct parameters
mock_has_perms.assert_called_once_with(
team_member_role=mock_member_object.role,
team_table=mock_team,
route=KeyManagementRoutes.KEY_UPDATE.value,
)
@pytest.mark.asyncio

85
tests/test_litellm/proxy/management_endpoints/test_org_admin_team_access.py
View file

@ -2,7 +2,6 @@
Tests for org admin access to team management endpoints.
Covers:
- _is_user_org_admin_for_team helper
- validate_membership allowing org admins
- _user_is_org_admin route-level check (no privilege escalation)
"""
@ -68,7 +67,7 @@ def _make_caller_user(
def _patch_org_admin_deps(get_user_return):
"""Context manager that patches the lazy imports inside _is_user_org_admin_for_team."""
"""Context manager that patches the lazy imports inside PrismaOrgRoles.is_org_admin."""
return (
patch(
"litellm.proxy.auth.auth_checks.get_user_object",
@ -83,88 +82,6 @@ def _patch_org_admin_deps(get_user_return):
)
# ---------------------------------------------------------------------------
# _is_user_org_admin_for_team
# ---------------------------------------------------------------------------
class TestIsUserOrgAdminForTeam:
"""Tests for the reusable _is_user_org_admin_for_team helper."""
@pytest.mark.asyncio
async def test_org_admin_for_teams_org_returns_true(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = _make_team(organization_id="org-1")
key = _make_user_key(user_id="org-admin-user")
caller = _make_caller_user(user_id="org-admin-user", org_id="org-1")
p1, p2, p3, p4 = _patch_org_admin_deps(caller)
with p1, p2, p3, p4:
result = await _is_user_org_admin_for_team(
user_api_key_dict=key, team_obj=team
)
assert result is True
@pytest.mark.asyncio
async def test_org_admin_different_org_returns_false(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = _make_team(organization_id="org-1")
key = _make_user_key(user_id="other-admin")
caller = _make_caller_user(user_id="other-admin", org_id="org-2")
p1, p2, p3, p4 = _patch_org_admin_deps(caller)
with p1, p2, p3, p4:
result = await _is_user_org_admin_for_team(
user_api_key_dict=key, team_obj=team
)
assert result is False
@pytest.mark.asyncio
async def test_team_without_org_returns_false(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = _make_team(organization_id=None)
key = _make_user_key()
result = await _is_user_org_admin_for_team(user_api_key_dict=key, team_obj=team)
assert result is False
@pytest.mark.asyncio
async def test_org_member_not_admin_returns_false(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = _make_team(organization_id="org-1")
key = _make_user_key(user_id="regular")
caller = _make_caller_user(user_id="regular", org_id="org-1", org_role="user")
p1, p2, p3, p4 = _patch_org_admin_deps(caller)
with p1, p2, p3, p4:
result = await _is_user_org_admin_for_team(
user_api_key_dict=key, team_obj=team
)
assert result is False
@pytest.mark.asyncio
async def test_no_user_id_returns_false(self):
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
team = _make_team(organization_id="org-1")
key = _make_user_key(user_id=None)
result = await _is_user_org_admin_for_team(user_api_key_dict=key, team_obj=team)
assert result is False
# ---------------------------------------------------------------------------
# validate_membership
# ---------------------------------------------------------------------------

22
tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py
View file

@ -7,6 +7,7 @@ redacted audit rows for callback mutations.
"""
import json
from typing import Final
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
@ -20,6 +21,7 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.common_utils.callback_config_validation import cross_entry_family_error
from litellm.proxy.management.teams.access import TeamAccess
from litellm.proxy.management_endpoints.team_callback_endpoints import (
add_team_callbacks,
delete_team_callback,
@ -28,6 +30,14 @@ from litellm.proxy.management_endpoints.team_callback_endpoints import (
)
class _NoOrgAdmins:
async def is_org_admin(self, user_id: str, organization_id: str) -> bool:
return False
NO_ORG_ADMINS: Final = TeamAccess(org_roles=_NoOrgAdmins())
def _team_row(
*,
team_id: str = "team-victim",
@ -99,9 +109,8 @@ def patched_prisma():
with (
patch("litellm.proxy.proxy_server.prisma_client") as mock_client,
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
"litellm.proxy.management_endpoints.team_callback_endpoints.get_team_access",
return_value=NO_ORG_ADMINS,
),
):
mock_client.get_data = AsyncMock(return_value=_team_row())
@ -1488,10 +1497,9 @@ async def test_unknown_team_is_indistinguishable_from_no_access(call_handler, un
): # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through
mock_client.get_data = AsyncMock(return_value=_team_row())
mock_client.db.litellm_teamtable.update = AsyncMock()
with patch( # test-quality-ok: _verify_team_access calls this module-level helper directly, so there is no seam to inject through
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
with patch( # test-quality-ok: the handler builds its TeamAccess through this module-level provider, so it is the seam to inject through
"litellm.proxy.management_endpoints.team_callback_endpoints.get_team_access",
return_value=NO_ORG_ADMINS,
):
with pytest.raises(HTTPException) as no_access:
await call_handler(unauthorized_caller)

205
tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py
View file

@ -1,6 +1,7 @@
import asyncio
import json
from contextlib import asynccontextmanager, contextmanager
from contextlib import AbstractContextManager, asynccontextmanager, contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from types import SimpleNamespace
from collections.abc import Sequence
@ -39,6 +40,7 @@ from litellm.proxy._types import (
UpdateTeamRequest,
UserAPIKeyAuth, # Import UserAPIKeyAuth
)
from litellm.proxy.management.teams.access import TeamAccess
from litellm.proxy.management_endpoints.team_endpoints import (
_STRIP_DELETED_TEAM_FROM_USERS_SQL,
GetTeamMemberPermissionsResponse,
@ -51,7 +53,6 @@ from litellm.proxy.management_endpoints.team_endpoints import (
_update_model_table,
_validate_and_populate_member_user_info,
_validate_team_member_reset_spend_value,
_verify_team_access,
delete_team,
list_available_teams,
reset_team_member_budget_fn,
@ -103,15 +104,29 @@ def _team_admin_may_edit(*fields: str):
yield
def _not_org_admin():
"""update_team asks whether the caller administers the team's org before it settles for team admin;
a MagicMock prisma cannot answer that lookup, so pin it to False."""
return patch( # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
AsyncMock(return_value=False),
@dataclass(frozen=True, slots=True)
class OrgAdmins:
of: frozenset[tuple[str, str]]
async def is_org_admin(self, user_id: str, organization_id: str) -> bool:
return (user_id, organization_id) in self.of
def _org_admins(*user_org_pairs: tuple[str, str]) -> AbstractContextManager[object]:
"""Answer the team handlers' org-admin lookup from ``(user_id, organization_id)`` pairs instead of prisma."""
team_access: Final = TeamAccess(org_roles=OrgAdmins(of=frozenset(user_org_pairs)))
return patch( # test-quality-ok: this file's MagicMock prisma cannot answer the org-admin lookup
"litellm.proxy.management_endpoints.team_endpoints.get_team_access",
lambda: team_access,
)
def _not_org_admin() -> AbstractContextManager[object]:
"""update_team and team_info ask whether the caller administers the team's org before settling for team admin;
a MagicMock prisma cannot answer that lookup, so nobody is an org admin."""
return _org_admins()
def _wire_team_create_tx(prisma_client):
"""`/team/new` inserts the team and mirrors it onto the access groups in one transaction,
so a mocked client has to hand its team table back out of `db.tx()`.
@ -1398,10 +1413,6 @@ async def test_validate_team_member_add_permissions_non_admin():
team.organization_id = None
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=False,
@ -1440,10 +1451,6 @@ async def test_available_team_self_join_with_caller_user_id_allowed():
team.organization_id = None
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1471,10 +1478,6 @@ async def test_available_team_self_join_blocks_admin_role():
team.organization_id = None
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1506,10 +1509,6 @@ async def test_available_team_self_join_blocks_other_user_id():
team.organization_id = None
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1542,10 +1541,6 @@ async def test_available_team_self_join_blocks_when_caller_has_no_user_id():
team.organization_id = None
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1582,10 +1577,6 @@ async def test_available_team_self_join_blocks_email_only_member():
)
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1625,10 +1616,6 @@ async def test_available_team_self_join_blocks_admin_role_in_member_list():
)
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1676,10 +1663,6 @@ async def test_available_team_self_join_blocks_member_budget_controls(budget_con
)
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1717,10 +1700,6 @@ async def test_available_team_self_join_allows_no_budget_controls():
)
with (
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_available_team",
return_value=True,
@ -1770,10 +1749,6 @@ async def test_update_team_member_permissions_blocks_non_admin_via_available_tea
new_callable=AsyncMock,
return_value=existing_row,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
return_value=False,
),
patch(
# Even with the available-team bypass mocked True, the endpoint
# must NOT consult it any more — the gate should reject the
@ -7198,7 +7173,7 @@ async def test_update_team_standalone_models_not_gated_by_user_limit(
Test that /team/update for a standalone team does NOT gate the team's models
by the caller's personal allowed models.
A team admin authorized via _verify_team_access() may set the team's models
A team admin authorized via TeamAccess.strongest_role() may set the team's models
independently of their own personal model list on update.
Scenario:
@ -7326,10 +7301,7 @@ async def test_update_team_org_scoped_budget_bypasses_user_limit(
mock_org.litellm_budget_table = mock_budget_table
with (
patch( # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
AsyncMock(return_value=True),
),
_org_admins(("org-admin-update-budget-test", "test-org-update-budget")),
patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma,
patch("litellm.proxy.proxy_server.user_api_key_cache") as mock_cache,
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
@ -7716,7 +7688,7 @@ async def test_update_team_tpm_limit_not_gated_by_user_limit(
Test that /team/update does NOT gate the team's tpm_limit by the caller's
personal tpm_limit.
A team admin authorized via _verify_team_access() may raise the team's
A team admin authorized via TeamAccess.strongest_role() may raise the team's
tpm_limit above their own personal tpm_limit on update.
Scenario:
@ -9493,10 +9465,6 @@ async def test_team_member_delete_persists_deleted_keys(monkeypatch):
"litellm.proxy.proxy_server.prisma_client",
mock_prisma_client,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
lambda **kwargs: True,
)
cache: Final = UserApiKeyCache()
revoked_cache_keys: Final = (
@ -9588,7 +9556,6 @@ async def test_team_member_delete_evicts_jwt_key_mapping_cache_of_the_keys_it_de
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", cache)
monkeypatch.setattr("litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin", lambda **kwargs: True)
await team_member_delete(
data=TeamMemberDeleteRequest(team_id="team-1", user_id="user-123"),
@ -11137,45 +11104,11 @@ class TestResolveTeamAccessGroupResources:
assert resolved.access_group_models is None
@pytest.mark.asyncio
async def test_verify_team_access_denies_unauthorized_user():
"""
Test that _verify_team_access raises 403 when the caller is not a proxy admin,
not a team admin, and not an org admin for the team's organization.
"""
team_obj = LiteLLM_TeamTable(
team_id="team-123",
team_alias="test-team",
members_with_roles=[
Member(role="admin", user_id="other_admin_user"),
],
organization_id="org-456",
)
# Caller is an internal user with no admin role and not in the team
caller = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="unauthorized_user",
)
with patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
):
with pytest.raises(HTTPException) as exc_info:
await _verify_team_access(
team_obj=team_obj,
user_api_key_dict=caller,
)
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_update_team_rejects_unauthorized_caller():
"""
Test that /team/update returns 403 when the caller is not a proxy admin,
not a team admin, and not an org admin — exercising the _verify_team_access
not a team admin, and not an org admin — exercising the TeamAccess.strongest_role
guard added to the update_team endpoint.
"""
from unittest.mock import Mock
@ -11196,11 +11129,7 @@ async def test_update_team_rejects_unauthorized_caller():
patch("litellm.proxy.proxy_server.user_api_key_cache"),
patch("litellm.proxy.proxy_server.proxy_logging_obj"),
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
),
_not_org_admin(),
):
mock_existing_team = MagicMock()
mock_existing_team.model_dump.return_value = {
@ -11669,20 +11598,17 @@ async def test_new_team_blocks_non_admin_passthrough_routes(mock_db_client):
@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."""
/team/update — the gate runs after TeamAccess.strongest_role."""
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"}
existing.model_dump.return_value = {"team_id": "t1", "organization_id": "org-1"}
mock_db_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=existing)
with patch(
"litellm.proxy.management_endpoints.team_endpoints._resolve_team_access",
AsyncMock(return_value="org_admin"),
):
with _org_admins(("u-team-admin", "org-1")):
with pytest.raises(ProxyException) as exc:
await update_team(
data=UpdateTeamRequest(
@ -11755,13 +11681,10 @@ async def test_update_team_blocks_non_admin_disable_global_guardrails(mock_db_cl
from litellm.proxy.management_endpoints.team_endpoints import update_team
existing = MagicMock()
existing.model_dump.return_value = {"team_id": "t1"}
existing.model_dump.return_value = {"team_id": "t1", "organization_id": "org-1"}
mock_db_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=existing)
with patch(
"litellm.proxy.management_endpoints.team_endpoints._resolve_team_access",
AsyncMock(return_value="org_admin"),
):
with _org_admins(("u-team-admin", "org-1")):
with pytest.raises(ProxyException) as exc:
await update_team(
data=UpdateTeamRequest(team_id="t1", disable_global_guardrails=True),
@ -14477,7 +14400,7 @@ def _wire_update_team(stack, existing_metadata):
@pytest.mark.asyncio
async def test_update_team_output_token_estimate_lowered_rejected_for_team_admin():
"""End-to-end wiring: _verify_team_access admits a team admin, so the gate
"""End-to-end wiring: TeamAccess.strongest_role admits a team admin, so the gate
has to fire inside update_team itself."""
import contextlib
from unittest.mock import Mock
@ -14569,7 +14492,7 @@ _TEAM_BATCH_LIMIT = "batch_enqueued_token_limit"
@pytest.mark.asyncio
async def test_update_team_batch_enqueued_token_limit_raised_rejected_for_team_admin():
"""_verify_team_access admits a team admin, so the gate has to fire inside
"""TeamAccess.strongest_role admits a team admin, so the gate has to fire inside
update_team itself to keep the team's batch quota admin-owned."""
import contextlib
from unittest.mock import Mock
@ -15141,7 +15064,6 @@ async def test_new_team_and_delete_team_both_drive_the_mirror(
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
patch("litellm.proxy.proxy_server.llm_router", None),
patch("litellm.proxy.management_endpoints.team_endpoints._persist_deleted_team_records", new_callable=AsyncMock),
patch("litellm.proxy.management_endpoints.team_endpoints._verify_team_access", new_callable=AsyncMock),
patch(
"litellm.proxy.management_endpoints.team_endpoints.sync_team_access_group_membership",
new_callable=AsyncMock,
@ -15391,7 +15313,7 @@ async def test_reset_team_member_spend_fn_forbidden_for_non_admin(monkeypatch):
@pytest.mark.asyncio
async def test_reset_team_member_spend_fn_team_admin_cannot_reset_own_spend(monkeypatch):
"""_verify_team_access authorizes a team admin over their own team with no check that the
"""TeamAccess.allows authorizes a team admin over their own team with no check that the
target differs from the caller. Unchecked, that admin could target their own membership row
and repeatedly zero it right before it crosses their per-member cap, consuming the shared
team budget without the configured limit ever binding (Veria finding on PR #37971)."""
@ -16101,9 +16023,7 @@ async def test_team_info_reports_parent_organization_models_only_to_team_manager
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: no seam on team_info
patch.object(team_endpoints, "get_all_team_memberships", AsyncMock(return_value=[])), # test-quality-ok: no seam on team_info
patch.object( # test-quality-ok: no seam on team_info
team_endpoints, "_is_user_org_admin_for_team", AsyncMock(return_value=False)
),
_not_org_admin(),
):
response = await team_endpoints.team_info(
http_request=MagicMock(spec=Request),
@ -16598,12 +16518,7 @@ async def test_update_team_holds_a_team_admin_to_the_org_tpm_limit(disable_audit
prisma = _wire_update_team(stack, {})
prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=org_team)
stack.enter_context(_team_admin_may_edit("tpm_limit"))
stack.enter_context(
patch( # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
AsyncMock(return_value=False),
)
)
stack.enter_context(_not_org_admin())
stack.enter_context(
patch( # test-quality-ok: update_team reads orgs through this module-level import; no seam to inject
"litellm.proxy.management_endpoints.team_endpoints.get_org_object",
@ -16745,15 +16660,21 @@ async def test_update_team_org_admin_is_not_filtered_by_the_team_admin_field_lis
"""A caller who is both org admin and roster admin keeps unrestricted edits."""
import contextlib
org_team = MagicMock()
org_team.metadata = {}
org_team.model_dump.return_value = {
"team_id": "test_team_id",
"team_alias": "test_team",
"organization_id": "org-1",
"metadata": {},
"members_with_roles": [{"user_id": "team-admin", "role": "admin"}],
}
with contextlib.ExitStack() as stack:
prisma = _wire_update_team(stack, {})
prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=org_team)
stack.enter_context(_team_admin_may_edit())
stack.enter_context(
patch( # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
AsyncMock(return_value=True),
)
)
stack.enter_context(_org_admins(("team-admin", "org-1")))
result = await update_team(
data=UpdateTeamRequest(team_id="test_team_id", team_alias="renamed"),
http_request=_update_request_stub(),
@ -16792,28 +16713,6 @@ async def test_update_team_unknown_team_is_403_for_non_proxy_admins_and_404_for_
assert str(missing.value.code) == "404"
@pytest.mark.asyncio
async def test_resolve_team_access_ranks_proxy_admin_then_org_admin_then_team_admin():
from litellm.proxy.management_endpoints.team_endpoints import _resolve_team_access
team = LiteLLM_TeamTable(
team_id="team-1",
organization_id="org-1",
members_with_roles=[Member(user_id="team-admin", role="admin")],
)
roster_admin = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="team-admin")
outsider = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="someone-else")
org_lookup = AsyncMock(return_value=False)
with patch("litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team", org_lookup): # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
assert await _resolve_team_access(team_obj=team, user_api_key_dict=_PROXY_ADMIN_CALLER) == "proxy_admin"
assert org_lookup.await_count == 0
assert await _resolve_team_access(team_obj=team, user_api_key_dict=roster_admin) == "team_admin"
assert await _resolve_team_access(team_obj=team, user_api_key_dict=outsider) is None
org_lookup.return_value = True
assert await _resolve_team_access(team_obj=team, user_api_key_dict=roster_admin) == "org_admin"
_ROSTER_ADMIN_CALLER = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="admin-1")
_MEMBER_CALLER = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="member-1")
@ -16862,9 +16761,7 @@ async def test_team_info_reports_what_the_caller_may_edit(caller, org_admin, ena
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: no seam on team_info
patch.object(team_endpoints, "get_all_team_memberships", AsyncMock(return_value=[])), # test-quality-ok: no seam on team_info
patch.object( # test-quality-ok: the org-admin lookup needs a real prisma client this file's MagicMock cannot provide
team_endpoints, "_is_user_org_admin_for_team", AsyncMock(return_value=org_admin)
),
_org_admins(("admin-1", "org-1")) if org_admin else _not_org_admin(),
_team_admin_may_edit(*enabled_fields),
):
response = await team_endpoints.team_info(

10
tests/test_litellm/proxy/spend_tracking/test_spend_management_endpoints.py
View file

@ -264,7 +264,7 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.hooks.proxy_track_cost_callback import _ProxyDBLogger
from litellm.proxy.management_endpoints import common_utils
from litellm.proxy.management.teams import access as team_access
from litellm.proxy.proxy_server import app
from litellm.proxy.spend_tracking import spend_management_endpoints
from litellm.router import Router
@ -335,8 +335,8 @@ async def test_can_team_member_view_log_team_not_found(monkeypatch):
prisma = MockPrisma()
# Even if admin check would return True, no team means False
monkeypatch.setattr(
common_utils,
"_is_user_team_admin",
team_access,
"is_team_admin",
lambda user_api_key_dict, team_obj: True,
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
@ -373,8 +373,8 @@ async def test_can_team_member_view_log_not_admin(monkeypatch):
prisma = MockPrisma()
monkeypatch.setattr(
common_utils,
"_is_user_team_admin",
team_access,
"is_team_admin",
lambda user_api_key_dict, team_obj: False,
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")

0
tests/unit/proxy/management/__init__.py Normal file
View file

0
tests/unit/proxy/management/teams/__init__.py Normal file
View file

136
tests/unit/proxy/management/teams/test_access.py Normal file
View file

@ -0,0 +1,136 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Final
import pytest
from fastapi import HTTPException
from litellm.proxy._types import LiteLLM_TeamTable, LitellmUserRoles, Member, UserAPIKeyAuth
from litellm.proxy.management.teams.access import (
TEAM_ADMIN_ONLY,
TEAM_OR_ORG_ADMIN,
TeamAccess,
TeamRole,
is_team_admin,
team_access_denied,
)
ADMIN: Final = Member(user_id="admin", role="admin")
MEMBER: Final = Member(user_id="member", role="user")
@dataclass(frozen=True, slots=True)
class OrgAdmins:
of: frozenset[tuple[str, str]]
async def is_org_admin(self, user_id: str, organization_id: str) -> bool:
return (user_id, organization_id) in self.of
class NoOrgLookup:
async def is_org_admin(self, user_id: str, organization_id: str) -> bool:
raise AssertionError(f"org lookup ran for {user_id} in {organization_id}")
def team(*members: Member, organization_id: str | None = "org-1") -> LiteLLM_TeamTable:
return LiteLLM_TeamTable(team_id="team-1", organization_id=organization_id, members_with_roles=list(members))
def caller(user_id: str | None, role: LitellmUserRoles = LitellmUserRoles.INTERNAL_USER) -> UserAPIKeyAuth:
return UserAPIKeyAuth(user_id=user_id, api_key="sk-x", user_role=role)
BOSS_OF_ORG_1: Final = OrgAdmins(of=frozenset({("boss", "org-1")}))
@pytest.mark.parametrize(
("who", "allow", "expected"),
[
(caller("root", LitellmUserRoles.PROXY_ADMIN), TEAM_ADMIN_ONLY, True),
(caller("root", LitellmUserRoles.PROXY_ADMIN), TEAM_OR_ORG_ADMIN, True),
(caller("root", LitellmUserRoles.PROXY_ADMIN), frozenset({"team_admin"}), False),
(caller("admin"), TEAM_ADMIN_ONLY, True),
(caller("admin"), frozenset({"proxy_admin"}), False),
(caller("member"), TEAM_ADMIN_ONLY, False),
],
)
async def test_allows_answers_proxy_and_team_admins_without_an_org_lookup(
who: UserAPIKeyAuth, allow: frozenset[TeamRole], expected: bool
) -> None:
assert await TeamAccess(org_roles=NoOrgLookup()).allows(who, team(ADMIN, MEMBER), allow) is expected
async def test_allows_checks_the_roster_before_the_org_lookup() -> None:
assert await TeamAccess(org_roles=NoOrgLookup()).allows(caller("admin"), team(ADMIN), TEAM_OR_ORG_ADMIN)
@pytest.mark.parametrize(
("who", "on_team", "allow", "expected"),
[
(caller("boss"), team(ADMIN, organization_id="org-1"), TEAM_OR_ORG_ADMIN, True),
(caller("boss"), team(ADMIN, organization_id="org-1"), TEAM_ADMIN_ONLY, False),
(caller("boss"), team(ADMIN, organization_id="org-2"), TEAM_OR_ORG_ADMIN, False),
(caller("member"), team(MEMBER, organization_id="org-1"), TEAM_OR_ORG_ADMIN, False),
],
)
async def test_allows_admits_org_admins_only_of_the_teams_org_and_only_when_asked(
who: UserAPIKeyAuth, on_team: LiteLLM_TeamTable, allow: frozenset[TeamRole], expected: bool
) -> None:
assert await TeamAccess(org_roles=BOSS_OF_ORG_1).allows(who, on_team, allow) is expected
@pytest.mark.parametrize(
("who", "on_team"),
[
pytest.param(caller(None), team(organization_id="org-1"), id="caller-without-user-id"),
pytest.param(caller(""), team(organization_id="org-1"), id="caller-with-empty-user-id"),
pytest.param(caller("boss"), team(organization_id=None), id="team-without-org"),
pytest.param(caller("boss"), team(organization_id=""), id="team-with-empty-org"),
],
)
async def test_allows_skips_the_org_lookup_without_a_user_and_an_org(
who: UserAPIKeyAuth, on_team: LiteLLM_TeamTable
) -> None:
assert await TeamAccess(org_roles=NoOrgLookup()).allows(who, on_team, TEAM_OR_ORG_ADMIN) is False
@pytest.mark.parametrize(
("who", "on_team", "org_roles", "expected"),
[
(caller("root", LitellmUserRoles.PROXY_ADMIN), team(), NoOrgLookup(), "proxy_admin"),
(caller("boss"), team(Member(user_id="boss", role="admin")), BOSS_OF_ORG_1, "org_admin"),
(caller("boss"), team(), BOSS_OF_ORG_1, "org_admin"),
(caller("admin"), team(ADMIN), BOSS_OF_ORG_1, "team_admin"),
(caller("member"), team(ADMIN, MEMBER), BOSS_OF_ORG_1, None),
],
)
async def test_strongest_role_ranks_org_admin_above_team_admin(
who: UserAPIKeyAuth,
on_team: LiteLLM_TeamTable,
org_roles: OrgAdmins | NoOrgLookup,
expected: TeamRole | None,
) -> None:
assert await TeamAccess(org_roles=org_roles).strongest_role(who, on_team) == expected
@pytest.mark.parametrize(
("members", "user_id", "expected"),
[
((ADMIN,), "admin", True),
((MEMBER,), "member", False),
((MEMBER, ADMIN), "admin", True),
((), "admin", False),
((ADMIN,), "someone-else", False),
((Member(user_id=None, user_email="a@b.c", role="admin"),), None, False),
],
)
def test_is_team_admin_reads_the_roster(members: tuple[Member, ...], user_id: str | None, expected: bool) -> None:
assert is_team_admin(caller(user_id), team(*members)) is expected
def test_team_access_denied_is_the_403_management_routes_have_always_raised() -> None:
with pytest.raises(HTTPException) as denied:
team_access_denied()
assert denied.value.status_code == 403
assert denied.value.detail == "You do not have access to this team"

0
tests/unit/proxy/management/users/__init__.py Normal file
View file

53
tests/unit/proxy/management/users/test_service.py Normal file
View file

@ -0,0 +1,53 @@
from __future__ import annotations
from datetime import datetime, timezone
from typing import Final
import pytest
from litellm.caching.dual_cache import DualCache
from litellm.proxy._types import LiteLLM_OrganizationMembershipTable, LiteLLM_UserTable, LitellmUserRoles
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.management.users.service import PrismaOrgRoles, holds_org_admin
from litellm.proxy.utils import ProxyLogging
NOW: Final = datetime.now(timezone.utc)
def user_in(*memberships: tuple[str, str]) -> LiteLLM_UserTable:
return LiteLLM_UserTable(
user_id="u1",
organization_memberships=[
LiteLLM_OrganizationMembershipTable(
user_id="u1", organization_id=organization_id, user_role=role, created_at=NOW, updated_at=NOW
)
for organization_id, role in memberships
],
)
@pytest.mark.parametrize(
("user", "expected"),
[
(user_in(("org-1", LitellmUserRoles.ORG_ADMIN.value)), True),
(user_in(("org-2", LitellmUserRoles.ORG_ADMIN.value)), False),
(user_in(("org-1", LitellmUserRoles.INTERNAL_USER.value)), False),
(user_in(("org-2", LitellmUserRoles.ORG_ADMIN.value), ("org-1", LitellmUserRoles.ORG_ADMIN.value)), True),
(user_in(), False),
(LiteLLM_UserTable(user_id="u1", organization_memberships=None), False),
(None, False),
],
)
def test_holds_org_admin_needs_the_org_admin_role_in_that_org(user: LiteLLM_UserTable | None, expected: bool) -> None:
assert holds_org_admin(user, "org-1") is expected
@pytest.mark.parametrize(
("organization_id", "expected"),
[("org-1", True), ("org-2", False)],
)
async def test_prisma_org_roles_answers_from_the_cached_user_row(organization_id: str, expected: bool) -> None:
cache: Final = UserApiKeyCache()
await cache.async_set_cache(key="u1", value=user_in(("org-1", LitellmUserRoles.ORG_ADMIN.value)))
roles: Final = PrismaOrgRoles(None, cache, ProxyLogging(user_api_key_cache=DualCache()))
assert await roles.is_org_admin("u1", organization_id) is expected