From c7badb9bdc5aeaa216ff083e9a4012c0371a0390 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Sat, 20 Jun 2026 14:29:47 -0700 Subject: [PATCH] feat(scim): drive global proxy role from a SCIM admin group Adds an optional litellm_settings.scim_admin_group. When configured, the global proxy role is recomputed from a user's resulting groups on every SCIM write that can change membership: user create/PUT/PATCH and group create/PUT/PATCH/DELETE, plus the existing-email upsert path. Membership in the admin group grants PROXY_ADMIN and its absence demotes to the non-admin default, enabling just-in-time elevation and automatic demotion without a re-login. When the setting is unset the role is never touched, so current behavior is preserved and a misconfigured IdP can never unexpectedly grant admin. --- .../management_endpoints/scim/scim_v2.py | 197 +++- .../scim/test_scim_v2_endpoints.py | 1015 +++++++++++++++++ 2 files changed, 1197 insertions(+), 15 deletions(-) diff --git a/litellm/proxy/management_endpoints/scim/scim_v2.py b/litellm/proxy/management_endpoints/scim/scim_v2.py index 0798d1a510d..0e1fce51dad 100644 --- a/litellm/proxy/management_endpoints/scim/scim_v2.py +++ b/litellm/proxy/management_endpoints/scim/scim_v2.py @@ -5,7 +5,7 @@ This is an enterprise feature and requires a premium license. """ import re -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Any, Dict, Iterable, List, Optional, Set, Tuple from fastapi import ( APIRouter, @@ -69,14 +69,21 @@ class UserProvisionerHelpers: @staticmethod async def handle_existing_user_by_email( - prisma_client, new_user_request: NewUserRequest + prisma_client, + new_user_request: NewUserRequest, + admin_group: Optional[str] = None, ) -> Optional[SCIMUser]: """ Check if a user with the given email already exists and update them if found. + When admin_group is configured the resolved global role on new_user_request + is persisted too, so re-upserting an existing email demotes a user who is no + longer in the admin group instead of leaving the stale role. + Args: prisma_client: Database client new_user_request: New user request data + admin_group: Configured SCIM admin group, or None to leave role untouched Returns: SCIMUser if user was updated, None if no existing user found @@ -100,6 +107,11 @@ class UserProvisionerHelpers: "user_alias": new_user_request.user_alias, "teams": new_user_request.teams, "metadata": safe_dumps(new_user_request.metadata), + **( + {"user_role": new_user_request.user_role} + if admin_group is not None + else {} + ), }, ) @@ -244,6 +256,117 @@ async def _get_scim_upsert_user_setting() -> bool: return True +ScimUserRole = Literal[ + LitellmUserRoles.PROXY_ADMIN, + LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, + LitellmUserRoles.INTERNAL_USER, + LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, +] + + +def _default_scim_user_role() -> ScimUserRole: + """Non-admin default role for SCIM-provisioned users.""" + if litellm.default_internal_user_params: + configured_role = litellm.default_internal_user_params.get("user_role") + if configured_role is not None: + return configured_role + return LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +async def _get_scim_admin_group() -> Optional[str]: + """ + Get the scim_admin_group setting from litellm_settings. + + Returns the configured admin group identifier, or None when unset so callers + leave a user's global role untouched (default-safe). + """ + try: + from litellm.proxy.proxy_server import proxy_config + + config = await proxy_config.get_config() + litellm_settings = config.get("litellm_settings", {}) or {} + return litellm_settings.get("scim_admin_group") or None + except Exception as e: + verbose_proxy_logger.warning( + f"Error reading scim_admin_group setting, defaulting to None: {e}" + ) + return None + + +def _resolve_scim_user_role( + groups: list[SCIMUserGroup], + admin_group: Optional[str], + default_role: ScimUserRole, +) -> Optional[LitellmUserRoles]: + """ + Resolve a user's global proxy role from their SCIM groups. + + Returns None when no admin group is configured, signalling callers to leave + the role unchanged. Otherwise grants PROXY_ADMIN when any group matches the + admin group by value or display, and falls back to the non-admin default. + """ + if admin_group is None: + return None + for group in groups: + if group.value == admin_group or group.display == admin_group: + return LitellmUserRoles.PROXY_ADMIN + return default_role + + +async def _scim_groups_from_team_ids( + prisma_client: Any, team_ids: list[str] +) -> list[SCIMUserGroup]: + """ + Build SCIMUserGroup objects from team ids, populating display from each + team's alias so admin-group matching by display name works the same way it + does on PUT (where SCIM groups carry display names natively). + """ + teams = [ + await TeamRepository(prisma_client).table.find_unique( + where={"team_id": team_id} + ) + for team_id in team_ids + ] + return [ + SCIMUserGroup( + value=team_id, + display=team.team_alias if team is not None else None, + ) + for team_id, team in zip(team_ids, teams) + ] + + +async def _recompute_scim_member_roles( + prisma_client: Any, user_ids: Iterable[str] +) -> None: + """ + Recompute and persist each user's global proxy role from their resulting team + membership. No-op unless scim_admin_group is configured, so a SCIM group write + that drops a member from the admin group demotes them just like the user + endpoints do, and the role is left untouched when the feature is off. + """ + admin_group = await _get_scim_admin_group() + if admin_group is None: + return + + default_role = _default_scim_user_role() + for user_id in user_ids: + user = await UserRepository(prisma_client).table.find_unique( + where={"user_id": user_id} + ) + if user is None: + continue + resolved_role = _resolve_scim_user_role( + await _scim_groups_from_team_ids(prisma_client, user.teams or []), + admin_group, + default_role, + ) + await UserRepository(prisma_client).table.update( + where={"user_id": user_id}, + data={"user_role": resolved_role}, + ) + + async def _extract_group_member_ids(group: SCIMGroup) -> GroupMemberExtractionResult: """ Extract member IDs from SCIMGroup, validating that all users exist. @@ -1002,16 +1125,11 @@ async def create_user( user_data["given_name"], user_data["family_name"] ) - default_role: Optional[ - Literal[ - LitellmUserRoles.PROXY_ADMIN, - LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, - LitellmUserRoles.INTERNAL_USER, - LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, - ] - ] = LitellmUserRoles.INTERNAL_USER_VIEW_ONLY - if litellm.default_internal_user_params: - default_role = litellm.default_internal_user_params.get("user_role") + default_role = _default_scim_user_role() + admin_group = await _get_scim_admin_group() + resolved_role = _resolve_scim_user_role( + user.groups or [], admin_group, default_role + ) new_user_request = NewUserRequest( user_id=user_id, @@ -1020,12 +1138,14 @@ async def create_user( teams=user_data["teams"], metadata=metadata, auto_create_key=False, - user_role=default_role, + user_role=resolved_role if admin_group is not None else default_role, ) # Check if user with email already exists and update if found existing_user_scim = await UserProvisionerHelpers.handle_existing_user_by_email( - prisma_client=prisma_client, new_user_request=new_user_request + prisma_client=prisma_client, + new_user_request=new_user_request, + admin_group=admin_group, ) if existing_user_scim: @@ -1104,6 +1224,12 @@ async def update_user( "metadata": safe_dumps(metadata), } + admin_group = await _get_scim_admin_group() + if admin_group is not None: + update_data["user_role"] = _resolve_scim_user_role( + user.groups or [], admin_group, _default_scim_user_role() + ) + updated_user = await UserRepository(prisma_client).table.update( where={"user_id": user_id}, data=update_data, @@ -1417,6 +1543,14 @@ async def patch_user( update_data["teams"] = list(final_team_set) + admin_group = await _get_scim_admin_group() + if admin_group is not None: + update_data["user_role"] = _resolve_scim_user_role( + await _scim_groups_from_team_ids(prisma_client, list(final_team_set)), + admin_group, + _default_scim_user_role(), + ) + # Serialize metadata to JSON string for Prisma to avoid GraphQL parsing issues if "metadata" in update_data and isinstance(update_data["metadata"], dict): from litellm.litellm_core_utils.safe_json_dumps import safe_dumps @@ -1599,6 +1733,8 @@ async def create_group( user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN), ) + await _recompute_scim_member_roles(prisma_client, member_result.all_member_ids) + scim_group = await ScimTransformations.transform_litellm_team_to_scim_group( created_team ) @@ -1665,6 +1801,19 @@ async def update_group( final_members=final_members, ) + # A rename can flip whether this group matches scim_admin_group by display + # name, so retained members must be re-resolved too, not just the ones whose + # membership changed. + alias_changed = existing_team.team_alias != group.displayName + await _recompute_scim_member_roles( + prisma_client, + ( + current_members | final_members + if alias_changed + else current_members ^ final_members + ), + ) + # Convert to SCIM format and return scim_group = await ScimTransformations.transform_litellm_team_to_scim_group( updated_team @@ -1691,8 +1840,10 @@ async def delete_group( prisma_client = await _get_prisma_client_or_raise_exception() existing_team = await _check_team_exists(group_id) + member_ids = await _get_team_member_user_ids_from_team(existing_team) + # For each member, remove this team from their teams list - for member_id in existing_team.members or []: + for member_id in member_ids: user = await UserRepository(prisma_client).table.find_unique( where={"user_id": member_id} ) @@ -1704,6 +1855,8 @@ async def delete_group( where={"user_id": member_id}, data={"teams": new_teams} ) + await _recompute_scim_member_roles(prisma_client, member_ids) + # Delete team await TeamRepository(prisma_client).table.delete(where={"team_id": group_id}) @@ -1903,6 +2056,20 @@ async def patch_group( # Handle user-team relationship changes await _handle_group_membership_changes(group_id, current_members, final_members) + # A rename can flip whether this group matches scim_admin_group by display + # name, so retained members must be re-resolved too, not just the ones whose + # membership changed. + new_alias = update_data.get("team_alias", existing_team.team_alias) + alias_changed = new_alias != existing_team.team_alias + await _recompute_scim_member_roles( + prisma_client, + ( + current_members | final_members + if alias_changed + else current_members ^ final_members + ), + ) + # Refresh team one more time to get final state after membership changes final_team = await TeamRepository(prisma_client).table.find_unique( where={"team_id": group_id} diff --git a/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py b/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py index ad893012807..d5cd2f87e59 100644 --- a/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py @@ -15,10 +15,13 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import ( _extract_group_member_ids, _handle_team_membership_changes, _process_group_patch_operations, + _recompute_scim_member_roles, create_group, create_user, + delete_group, get_users, get_service_provider_config, + patch_group, patch_user, update_group, update_user, @@ -1720,3 +1723,1015 @@ async def test_process_group_patch_operations_with_flag_false_rejects( assert exc_info.value.status_code == 400 assert "does not exist" in str(exc_info.value.detail) assert "new-user-1" in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_create_user_grants_admin_when_in_scim_admin_group(mocker, monkeypatch): + """When scim_admin_group is configured and a created user's groups include it, + the user is provisioned as PROXY_ADMIN.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="new-admin", + emails=[SCIMUserEmail(value="new-admin@example.com")], + groups=[SCIMUserGroup(value="litellm-admins", display="LiteLLM Admins")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_usertable.find_first = AsyncMock(return_value=None) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + new_user_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.new_user", + AsyncMock(return_value=NewUserRequest(user_id="new-admin")), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await create_user(user=scim_user) + + called_args = new_user_mock.call_args.kwargs["data"] + assert called_args.user_role == LitellmUserRoles.PROXY_ADMIN + + +@pytest.mark.asyncio +async def test_create_user_keeps_default_when_not_in_scim_admin_group( + mocker, monkeypatch +): + """When scim_admin_group is configured but the user's groups don't include it, + the user keeps the non-admin default role.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="regular-user", + emails=[SCIMUserEmail(value="regular@example.com")], + groups=[SCIMUserGroup(value="engineering", display="Engineering")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_usertable.find_first = AsyncMock(return_value=None) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + new_user_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.new_user", + AsyncMock(return_value=NewUserRequest(user_id="regular-user")), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await create_user(user=scim_user) + + called_args = new_user_mock.call_args.kwargs["data"] + assert called_args.user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_update_user_demotes_admin_when_removed_from_scim_admin_group( + mocker, monkeypatch +): + """Core demotion test: a PUT whose new groups no longer include the configured + admin group must re-evaluate the role and write the non-admin default, so an + admin removed from the IdP group is demoted without re-login.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + existing_user = mocker.MagicMock() + existing_user.teams = ["litellm-admins"] + existing_user.metadata = {} + + updated_user = { + "user_id": "demote-me", + "user_email": "demote@example.com", + "user_alias": None, + "teams": ["engineering"], + "metadata": "{}", + } + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="demote-me", + emails=[SCIMUserEmail(value="demote@example.com")], + groups=[SCIMUserGroup(value="engineering", display="Engineering")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value=updated_user + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._check_user_exists", + AsyncMock(return_value=existing_user), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._handle_team_membership_changes", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await update_user(user_id="demote-me", user=scim_user) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_update_user_does_not_force_role_when_scim_admin_group_unset( + mocker, monkeypatch +): + """When scim_admin_group is unset, PUT must not touch user_role (current + behavior preserved).""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + existing_user = mocker.MagicMock() + existing_user.teams = ["litellm-admins"] + existing_user.metadata = {} + + updated_user = { + "user_id": "no-touch", + "user_email": "no-touch@example.com", + "user_alias": None, + "teams": ["litellm-admins"], + "metadata": "{}", + } + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="no-touch", + emails=[SCIMUserEmail(value="no-touch@example.com")], + groups=[SCIMUserGroup(value="litellm-admins", display="LiteLLM Admins")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value=updated_user + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._check_user_exists", + AsyncMock(return_value=existing_user), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._handle_team_membership_changes", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await update_user(user_id="no-touch", user=scim_user) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert "user_role" not in call_args[1]["data"] + + +@pytest.mark.asyncio +async def test_update_user_demotes_when_default_params_lack_user_role( + mocker, monkeypatch +): + """Regression: default_internal_user_params set without a user_role key must + still resolve to the non-admin default on demotion, not silently skip and + leave the user PROXY_ADMIN.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr( + "litellm.default_internal_user_params", {"max_budget": 10}, raising=False + ) + + existing_user = mocker.MagicMock() + existing_user.teams = ["litellm-admins"] + existing_user.metadata = {} + + updated_user = { + "user_id": "demote-me", + "user_email": "demote@example.com", + "user_alias": None, + "teams": ["engineering"], + "metadata": "{}", + } + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="demote-me", + emails=[SCIMUserEmail(value="demote@example.com")], + groups=[SCIMUserGroup(value="engineering", display="Engineering")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value=updated_user + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._check_user_exists", + AsyncMock(return_value=existing_user), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._handle_team_membership_changes", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await update_user(user_id="demote-me", user=scim_user) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_patch_user_demotes_admin_when_removed_from_scim_admin_group( + mocker, monkeypatch +): + """PATCH that drops the admin team from the resulting team set must write the + non-admin default, mirroring the PUT demotion path.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + existing_user = mocker.MagicMock() + existing_user.teams = ["litellm-admins"] + existing_user.metadata = {} + + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[ + SCIMPatchOperation( + op="replace", path="groups", value=[{"value": "engineering"}] + ) + ], + ) + + updated_user = { + "user_id": "demote-me", + "user_alias": None, + "teams": ["engineering"], + "metadata": "{}", + } + + engineering_team = mocker.MagicMock() + engineering_team.team_alias = "Engineering" + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value=updated_user + ) + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=engineering_team + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._check_user_exists", + AsyncMock(return_value=existing_user), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._handle_team_membership_changes", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock( + return_value=SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="demote-me", + ) + ), + ) + + await patch_user(user_id="demote-me", patch_ops=patch_ops) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_patch_user_grants_admin_by_team_display_name(mocker, monkeypatch): + """PATCH carries groups as team ids, so admin-group matching must fall back to + each team's display name; an admin group configured as a human-readable alias + grants PROXY_ADMIN even when the team id differs.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "LiteLLM Admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + existing_user = mocker.MagicMock() + existing_user.teams = [] + existing_user.metadata = {} + + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[ + SCIMPatchOperation( + op="replace", path="groups", value=[{"value": "team-abc-123"}] + ) + ], + ) + + updated_user = { + "user_id": "promote-me", + "user_alias": None, + "teams": ["team-abc-123"], + "metadata": "{}", + } + + admin_team = mocker.MagicMock() + admin_team.team_alias = "LiteLLM Admins" + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value=updated_user + ) + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=admin_team + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._check_user_exists", + AsyncMock(return_value=existing_user), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._handle_team_membership_changes", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock( + return_value=SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="promote-me", + ) + ), + ) + + await patch_user(user_id="promote-me", patch_ops=patch_ops) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.PROXY_ADMIN + + +def _scim_admin_prisma(mocker, *, user_teams): + """Prisma double whose user resolves to user_teams and whose teams expose an + alias equal to their id, used by the role-recompute helper tests.""" + user = mocker.MagicMock() + user.user_id = "member-1" + user.teams = user_teams + + def _team_find_unique(where): + team = mocker.MagicMock() + team.team_alias = where["team_id"] + return team + + prisma = mocker.MagicMock() + prisma.db = mocker.MagicMock() + prisma.db.litellm_usertable = mocker.MagicMock() + prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=user) + prisma.db.litellm_usertable.update = AsyncMock(return_value=user) + prisma.db.litellm_teamtable = mocker.MagicMock() + prisma.db.litellm_teamtable.find_unique = AsyncMock(side_effect=_team_find_unique) + return prisma + + +@pytest.mark.asyncio +async def test_recompute_scim_member_roles_demotes_when_not_in_admin_group( + mocker, monkeypatch +): + """The shared recompute helper writes the non-admin default for a member whose + resulting teams no longer include the configured admin group.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + prisma = _scim_admin_prisma(mocker, user_teams=["engineering"]) + + await _recompute_scim_member_roles(prisma, ["member-1"]) + + call_args = prisma.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_recompute_scim_member_roles_grants_when_in_admin_group( + mocker, monkeypatch +): + """The shared recompute helper grants PROXY_ADMIN when a member's resulting + teams include the configured admin group.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + prisma = _scim_admin_prisma(mocker, user_teams=["litellm-admins"]) + + await _recompute_scim_member_roles(prisma, ["member-1"]) + + call_args = prisma.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.PROXY_ADMIN + + +@pytest.mark.asyncio +async def test_recompute_scim_member_roles_noop_when_admin_group_unset( + mocker, monkeypatch +): + """With scim_admin_group unset the recompute helper must not touch any role, + preserving current behavior for SCIM group writes.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + + prisma = _scim_admin_prisma(mocker, user_teams=["litellm-admins"]) + + await _recompute_scim_member_roles(prisma, ["member-1"]) + + prisma.db.litellm_usertable.update.assert_not_called() + + +@pytest.mark.asyncio +async def test_update_group_recomputes_roles_for_changed_members(mocker): + """PUT /Groups must recompute the global role for every member whose + membership changed, so an admin dropped from the admin group is demoted.""" + from litellm.proxy._types import LiteLLM_TeamTable, Member + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "test-team-123" + existing_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Admins", + members=["user1", "user2"], + members_with_roles=[ + Member(user_id="user1", role="user"), + Member(user_id="user2", role="user"), + ], + metadata={}, + ) + scim_group_update = SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Admins", + members=[SCIMMember(value="user2"), SCIMMember(value="user3")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( + return_value=mocker.MagicMock() + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock(return_value=scim_group_update), + ) + + await update_group(group_id=group_id, group=scim_group_update) + + recompute_mock.assert_awaited_once() + assert set(recompute_mock.call_args[0][1]) == {"user1", "user3"} + + +@pytest.mark.asyncio +async def test_patch_group_recomputes_roles_for_changed_members(mocker): + """PATCH /Groups must recompute the global role for every member whose + membership changed, mirroring the PUT path.""" + from litellm.proxy._types import LiteLLM_TeamTable, Member + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "test-team-123" + existing_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Admins", + members=["user1", "user2"], + members_with_roles=[ + Member(user_id="user1", role="user"), + Member(user_id="user2", role="user"), + ], + metadata={}, + ) + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[ + SCIMPatchOperation(op="remove", path="members", value=[{"value": "user1"}]) + ], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( + return_value=mocker.MagicMock() + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock( + return_value=SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Admins", + ) + ), + ) + + await patch_group(group_id=group_id, patch_ops=patch_ops) + + recompute_mock.assert_awaited_once() + assert set(recompute_mock.call_args[0][1]) == {"user1"} + + +@pytest.mark.asyncio +async def test_delete_group_recomputes_roles_for_members(mocker): + """DELETE /Groups must recompute the global role for the team's members, so + deleting the admin group demotes everyone who was only admin through it.""" + from litellm.proxy._types import Member + + existing_team = mocker.MagicMock() + existing_team.members_with_roles = [ + Member(user_id="user1", role="user"), + Member(user_id="user2", role="user"), + ] + + member = mocker.MagicMock() + member.teams = ["test-team-123"] + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_teamtable.delete = AsyncMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=member) + mock_prisma_client.db.litellm_usertable.update = AsyncMock() + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + + await delete_group(group_id="test-team-123") + + recompute_mock.assert_awaited_once() + assert list(recompute_mock.call_args[0][1]) == ["user1", "user2"] + + +@pytest.mark.asyncio +async def test_handle_existing_user_by_email_applies_role_when_admin_group_set(mocker): + """When admin_group is configured, re-upserting an existing email persists the + resolved role so a now-non-admin user can't keep a stale PROXY_ADMIN.""" + existing_user = mocker.MagicMock() + existing_user.user_id = "old-user-id" + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_first = AsyncMock( + return_value=existing_user + ) + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value={"user_id": "new-user-id"} + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=mocker.MagicMock()), + ) + + new_user_request = NewUserRequest( + user_id="new-user-id", + user_email="test@example.com", + teams=["engineering"], + metadata={}, + auto_create_key=False, + user_role=LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, + ) + + await UserProvisionerHelpers.handle_existing_user_by_email( + prisma_client=mock_prisma_client, + new_user_request=new_user_request, + admin_group="litellm-admins", + ) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_handle_existing_user_by_email_leaves_role_when_admin_group_unset(mocker): + """With admin_group unset, the existing-email upsert must not write user_role, + preserving current behavior when the feature is off.""" + existing_user = mocker.MagicMock() + existing_user.user_id = "old-user-id" + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_first = AsyncMock( + return_value=existing_user + ) + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value={"user_id": "new-user-id"} + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=mocker.MagicMock()), + ) + + new_user_request = NewUserRequest( + user_id="new-user-id", + user_email="test@example.com", + teams=["engineering"], + metadata={}, + auto_create_key=False, + user_role=LitellmUserRoles.PROXY_ADMIN, + ) + + await UserProvisionerHelpers.handle_existing_user_by_email( + prisma_client=mock_prisma_client, + new_user_request=new_user_request, + ) + + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert "user_role" not in call_args[1]["data"] + + +@pytest.mark.asyncio +async def test_create_user_existing_email_upsert_demotes_when_admin_group_set( + mocker, monkeypatch +): + """End-to-end create wiring: a SCIM POST that upserts an existing email while + the user is not in the admin group must write the non-admin default, not leave + a stale PROXY_ADMIN.""" + from litellm.proxy.proxy_server import proxy_config + + async def mock_get_config(): + return {"litellm_settings": {"scim_admin_group": "litellm-admins"}} + + monkeypatch.setattr(proxy_config, "get_config", mock_get_config) + monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False) + + scim_user = SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + userName="returning-user", + emails=[SCIMUserEmail(value="returning@example.com")], + groups=[SCIMUserGroup(value="engineering", display="Engineering")], + ) + + existing_user = mocker.MagicMock() + existing_user.user_id = "returning-user" + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_usertable.find_first = AsyncMock( + return_value=existing_user + ) + mock_prisma_client.db.litellm_usertable.update = AsyncMock( + return_value={"user_id": "returning-user"} + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + new_user_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.new_user", + AsyncMock(return_value=NewUserRequest(user_id="returning-user")), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user", + AsyncMock(return_value=scim_user), + ) + + await create_user(user=scim_user) + + new_user_mock.assert_not_called() + call_args = mock_prisma_client.db.litellm_usertable.update.call_args + assert call_args[1]["data"]["user_role"] == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + +@pytest.mark.asyncio +async def test_create_group_recomputes_roles_for_members(mocker): + """POST /Groups must recompute the global role for the new team's members, so a + team created with the admin-group display name elevates its members.""" + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "admin-team-1" + scim_group = SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="LiteLLM Admins", + members=[SCIMMember(value="user1"), SCIMMember(value="user2")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( + return_value=mocker.MagicMock() + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.new_team", + AsyncMock(return_value=mocker.MagicMock()), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock(return_value=scim_group), + ) + + await create_group(group=scim_group) + + recompute_mock.assert_awaited_once() + assert set(recompute_mock.call_args[0][1]) == {"user1", "user2"} + + +@pytest.mark.asyncio +async def test_update_group_rename_recomputes_retained_members(mocker): + """A PUT that renames the group (alias changes) but leaves membership unchanged + must still recompute retained members, since a rename can flip whether the + group matches scim_admin_group by display name.""" + from litellm.proxy._types import LiteLLM_TeamTable, Member + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "test-team-123" + existing_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="LiteLLM Admins", + members=["user1"], + members_with_roles=[Member(user_id="user1", role="user")], + metadata={}, + ) + scim_group_update = SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Engineering", + members=[SCIMMember(value="user1")], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( + return_value=mocker.MagicMock() + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock(return_value=scim_group_update), + ) + + await update_group(group_id=group_id, group=scim_group_update) + + recompute_mock.assert_awaited_once() + assert set(recompute_mock.call_args[0][1]) == {"user1"} + + +@pytest.mark.asyncio +async def test_patch_group_rename_recomputes_retained_members(mocker): + """A PATCH that renames the group (displayName op) but leaves membership + unchanged must still recompute retained members, mirroring the PUT path.""" + from litellm.proxy._types import LiteLLM_TeamTable, Member + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "test-team-123" + existing_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="LiteLLM Admins", + members=["user1"], + members_with_roles=[Member(user_id="user1", role="user")], + metadata={}, + ) + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[ + SCIMPatchOperation(op="replace", path="displayName", value="Engineering") + ], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock( + return_value=existing_team + ) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( + return_value=mocker.MagicMock() + ) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + recompute_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock( + return_value=SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Engineering", + ) + ), + ) + + await patch_group(group_id=group_id, patch_ops=patch_ops) + + recompute_mock.assert_awaited_once() + assert set(recompute_mock.call_args[0][1]) == {"user1"}