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.
This commit is contained in:
Yassin Kortam 2026-06-20 14:29:47 -07:00
parent 15aa40b36e
commit c7badb9bdc
2 changed files with 1197 additions and 15 deletions

View file

@ -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}