scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 35

fix(scim): stop provisioning nested group ids as internal users (#34997)

Browse source 
* fix(scim): stop provisioning nested group ids as internal users

POST/PUT/PATCH /scim/v2/Groups treated every member.value as a user id, so
with the default scim_upsert_user=true an unknown id was auto-created as an
internal user. Entra sends nested groups as members carrying "type": "Group",
which meant every nested group produced a phantom internal user whose id and
email were the group GUID, and those users counted toward licensed seats.

Group members are now classified before they are used: members typed "Group"
are skipped without a database hit, an id that names an existing team is
skipped too (Okta sends untyped ids through filtered paths, so the type alone
is not enough), and only ids that resolve to a user, or that resolve to
nothing at all, keep today's behavior. The user lookup runs before the team
lookup so a user whose id collides with a team id keeps syncing.

The type was previously dropped at parse time on POST/PUT because SCIMMember
had no such field, and on PATCH because the raw member dicts were reduced to
bare ids; both paths now share one resolver and one parser that preserves it.

Member removals no longer upsert: a remove of an id we do not know is an
idempotent no-op rather than a reason to create a user and immediately drop
it, and strict mode (scim_upsert_user=false) no longer rejects it. Removal of
an id that is on the roster but has no user row still cleans up membership.

Responses now state members are of type "User" instead of emitting a null,
and the advertised Group schema documents the members.type sub-attribute.

* fix(scim): harden group member classification after adversarial review

Removals now bypass classification and drop exactly the ids they name,
restoring cleanup of roster entries the old bug left behind. The
team-id fallback only applies to untyped members, so an explicit User
type always provisions even when the id collides with a team. Member
types are normalized before matching; a type other than User or Group
only skips when the id is not an existing user. Non-string type values
are tolerated as absent on every verb instead of failing validation.
Admitted member ids are deduped order-preserving, which also closes a
pre-existing duplicate-row hazard on group creation.

* fix(scim): only treat scim-managed teams as nested groups

A PR reviewer flagged that an untyped SCIM member whose id collides
with an admin-created team was silently skipped, suppressing that
user's provisioning. SCIM group writes (POST, PUT, and every PATCH)
now stamp the team with scim_managed metadata, and the typeless
team-id skip only applies to teams carrying that marker or the
scim_data blob older PUTs already wrote. Admin-created teams stay
unmarked, so a colliding untyped member provisions the user in
permissive mode and returns the standard unknown-user 400 in strict
mode. Teams SCIM touched before this change adopt the marker on their
next group write.
This commit is contained in:
ryan-crabbe-berri 2026-07-29 09:38:17 -07:00 • committed by GitHub
parent c274cf321c
commit 25ebe5600a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1230 additions and 120 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/proxy/management_endpoints/scim/scim_transformations.py
View file

@ -175,6 +175,7 @@ class ScimTransformations:
SCIMMember(
value=ScimTransformations._get_scim_member_value(member),
display=ScimTransformations._get_scim_member_display(member),
type="User",
)
)

425
litellm/proxy/management_endpoints/scim/scim_v2.py
View file

@ -5,7 +5,9 @@ This is an enterprise feature and requires a premium license.
"""
import re
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
from collections.abc import Mapping, Sequence
from itertools import chain
from typing import Any, Dict, Iterable, List, NamedTuple, Optional, Set, Tuple
from fastapi import (
APIRouter,
@ -17,8 +19,8 @@ from fastapi import (
Request,
Response,
)
from pydantic import BaseModel, ValidationError
from typing_extensions import TypedDict
from pydantic import BaseModel, TypeAdapter, ValidationError
from typing_extensions import TypedDict, assert_never
import litellm
from litellm._logging import verbose_proxy_logger
@ -50,7 +52,11 @@ from litellm.proxy.management_endpoints.team_endpoints import (
team_member_add,
team_member_delete,
)
from litellm.proxy.utils import _premium_user_check, handle_exception_on_proxy
from litellm.proxy.utils import (
PrismaClient,
_premium_user_check,
handle_exception_on_proxy,
)
from litellm.repositories.table_repositories import (
InvitationLinkRepository,
OrganizationMembershipRepository,
@ -143,7 +149,11 @@ class ScimUserData(TypedDict):
class GroupMemberExtractionResult(BaseModel):
"""Result of extracting and processing group members."""
"""Result of extracting and processing group members.
``all_member_ids`` is deduped order-preserving; ``existing_member_ids`` is not,
so a repeated resolved id appears once in the former and twice in the latter.
"""
existing_member_ids: List[str]
created_users: List[NewUserResponse]
@ -371,6 +381,216 @@ async def _recompute_scim_member_roles(prisma_client: Any, user_ids: Iterable[st
)
class _ResolvedUserMember(NamedTuple):
user_id: str
class _SkippedGroupMember(NamedTuple):
value: str
reason: Literal["nested_group", "non_user_type", "existing_team"]
class _UnknownMember(NamedTuple):
value: str
_ClassifiedGroupMember = Union[_ResolvedUserMember, _SkippedGroupMember, _UnknownMember]
class _PartitionedMembers(NamedTuple):
resolved_ids: tuple[str, ...]
skipped: tuple[_SkippedGroupMember, ...]
unknown_ids: tuple[str, ...]
def _member_value(member: SCIMMember) -> str:
"""A member id is opaque to us but has to be there; an empty one is a client error."""
if not member.value or not member.value.strip():
raise HTTPException(
status_code=400,
detail={"error": "Invalid member: user ID cannot be empty."},
)
return member.value
def _normalized_member_type(member: SCIMMember) -> str | None:
"""The canonical ``type`` a member declares, lowercased; blank or absent means none."""
normalized = (member.type or "").strip().lower()
return normalized or None
_JSON_OBJECT_ADAPTER = TypeAdapter(Dict[str, object])
def _json_object_fields(raw: object) -> Mapping[str, object] | None:
"""A typed, read-only view of a JSON object, or None when it is not one."""
try:
return _JSON_OBJECT_ADAPTER.validate_python(raw)
except ValidationError:
return None
def _team_metadata_has_scim_provenance(team_metadata: object) -> bool:
"""Whether a group write from the identity provider left its mark on this team.
``SCIM_TEAM_DATA_METADATA_KEY`` counts because PUT has been writing it since
long before the explicit marker, so a team the identity provider already
syncs is recognized without waiting to be written again.
"""
fields = _json_object_fields(team_metadata)
if fields is None:
return False
return bool(fields.get(SCIM_MANAGED_TEAM_METADATA_KEY)) or fields.get(SCIM_TEAM_DATA_METADATA_KEY) is not None
async def _classify_group_member(member: SCIMMember, prisma_client: PrismaClient) -> _ClassifiedGroupMember:
"""
Decide what a single SCIM group member refers to.
A LiteLLM team only holds users, so a member is dropped when it declares a type
other than ``User`` or when its id names an existing team. Both of those checks
are placed around the user lookup rather than before it, because the id of a
real user is the one thing that outranks them:
- ``"type": "Group"`` (what Entra sends for a nested group) is dropped without
a lookup. This bug provisioned nested group GUIDs as users, so those rows
exist in the wild and would otherwise resolve as members all over again.
- any other unrecognized type is dropped only after the user lookup misses.
Clients do send non-canonical types on real members (RFC 7643 defines
``direct`` for ``User.groups``), and dropping a live user over one would
revoke that user's team access on the next full sync.
- an id that names an existing team is dropped only when the member arrives
untyped, which is how Okta sends nested groups, and only when that team is
one the identity provider writes. An id the IdP called a User is a user
even if some team happens to share the id, and a team created here rather
than through SCIM is not evidence of anything about the member.
"""
value = _member_value(member)
member_type = _normalized_member_type(member)
if member_type == "group":
return _SkippedGroupMember(value=value, reason="nested_group")
user = await UserRepository(prisma_client).table.find_unique(where={"user_id": value})
if user is not None:
return _ResolvedUserMember(user_id=value)
if member_type is not None and member_type != "user":
return _SkippedGroupMember(value=value, reason="non_user_type")
if member_type is None:
team = await TeamRepository(prisma_client).table.find_unique(where={"team_id": value})
if team is not None and _team_metadata_has_scim_provenance(team.metadata):
return _SkippedGroupMember(value=value, reason="existing_team")
return _UnknownMember(value=value)
def _bucketed_member(entry: _ClassifiedGroupMember) -> _PartitionedMembers:
"""The single-member partition one classified entry contributes."""
match entry:
case _ResolvedUserMember(user_id=user_id):
return _PartitionedMembers(resolved_ids=(user_id,), skipped=(), unknown_ids=())
case _SkippedGroupMember():
return _PartitionedMembers(resolved_ids=(), skipped=(entry,), unknown_ids=())
case _UnknownMember(value=value):
return _PartitionedMembers(resolved_ids=(), skipped=(), unknown_ids=(value,))
case _:
assert_never(entry)
def _partition_classified_members(classified: Iterable[_ClassifiedGroupMember]) -> _PartitionedMembers:
"""Split classified members into the buckets the resolver acts on, keeping request order."""
bucketed = tuple(_bucketed_member(entry) for entry in classified)
return _PartitionedMembers(
resolved_ids=tuple(chain.from_iterable(bucket.resolved_ids for bucket in bucketed)),
skipped=tuple(chain.from_iterable(bucket.skipped for bucket in bucketed)),
unknown_ids=tuple(chain.from_iterable(bucket.unknown_ids for bucket in bucketed)),
)
def _admitted_member_id(entry: _ClassifiedGroupMember, created_ids: frozenset[str]) -> str | None:
match entry:
case _ResolvedUserMember(user_id=user_id):
return user_id
case _UnknownMember(value=value):
return value if value in created_ids else None
case _SkippedGroupMember():
return None
case _:
assert_never(entry)
def _admitted_member_ids(classified: Iterable[_ClassifiedGroupMember], created_ids: frozenset[str]) -> tuple[str, ...]:
"""Member ids that survive resolution, in the order the request listed them.
An id the request repeats is one member: the roster these ids are written to
holds one row per member, and a second creation attempt for the same id fails
against the real unique constraint even though the first one succeeded.
"""
return tuple(
dict.fromkeys(
member_id for entry in classified if (member_id := _admitted_member_id(entry, created_ids)) is not None
)
)
async def _resolve_group_member_ids(
members: Sequence[SCIMMember],
created_via: str,
prisma_client: PrismaClient,
) -> GroupMemberExtractionResult:
"""
Resolve SCIM group members to LiteLLM user ids, dropping members that are not users.
Only the operations that put ids onto a roster resolve their members: an id
that resolves to nothing is created when litellm_settings.scim_upsert_user is
True (default) and rejected per SCIM 2.0 otherwise. Removals do not come
through here; dropping an id is idempotent, so it needs neither a lookup nor a
user to drop.
Raises:
HTTPException: 400 when a member id is empty, or when scim_upsert_user is
False and a member id is neither an existing user, an existing team, nor a
member declared to be something other than a user.
"""
classified = tuple([await _classify_group_member(member, prisma_client) for member in members])
partition = _partition_classified_members(classified)
for skipped in partition.skipped:
verbose_proxy_logger.info(
"SCIM: ignoring non-user group member '%s' (%s); LiteLLM teams contain users only",
skipped.value,
skipped.reason,
)
if partition.unknown_ids and not await _get_scim_upsert_user_setting():
raise HTTPException(
status_code=400,
detail={
"error": f"User with ID '{partition.unknown_ids[0]}' does not exist. "
"Please create the user first via POST /Users before adding to group."
},
)
creations = tuple(
[
(user_id, await _create_user_if_not_exists(user_id=user_id, created_via=created_via))
for user_id in partition.unknown_ids
]
)
created_users = tuple(created for _, created in creations if created is not None)
return GroupMemberExtractionResult(
existing_member_ids=partition.resolved_ids,
created_users=created_users,
all_member_ids=_admitted_member_ids(
classified,
frozenset(user_id for user_id, created in creations if created is not None),
),
)
async def _extract_group_member_ids(group: SCIMGroup) -> GroupMemberExtractionResult:
"""
Extract member IDs from SCIMGroup, validating that all users exist.
@ -386,56 +606,10 @@ async def _extract_group_member_ids(group: SCIMGroup) -> GroupMemberExtractionRe
HTTPException: If scim_upsert_user is False and any member user does not exist (400 Bad Request)
"""
prisma_client = await _get_prisma_client_or_raise_exception()
existing_member_ids = []
created_users = []
all_member_ids = []
# Check the feature flag
scim_upsert_user = await _get_scim_upsert_user_setting()
if group.members:
for member in group.members:
user_id = member.value
# Validate user_id is not empty
if not user_id or not user_id.strip():
raise HTTPException(
status_code=400,
detail={"error": "Invalid member: user ID cannot be empty."},
)
# Check if user exists
user = await UserRepository(prisma_client).table.find_unique(where={"user_id": user_id})
if user:
existing_member_ids.append(user_id)
all_member_ids.append(user_id)
else:
if scim_upsert_user:
# Create the user if they don't exist (backward compatible behavior)
created_user = await _create_user_if_not_exists(
user_id=user_id, created_via="scim_group_membership"
)
if created_user:
created_users.append(created_user)
all_member_ids.append(user_id)
# If creation failed, user is skipped (logged in helper)
else:
# User doesn't exist - reject per SCIM 2.0 protocol
# This prevents security issues where users not assigned to app
# get provisioned via group membership
raise HTTPException(
status_code=400,
detail={
"error": f"User with ID '{user_id}' does not exist. "
"Please create the user first via POST /Users before adding to group."
},
)
return GroupMemberExtractionResult(
existing_member_ids=existing_member_ids,
created_users=created_users,
all_member_ids=all_member_ids,
return await _resolve_group_member_ids(
members=group.members or [],
created_via="scim_group_membership",
prisma_client=prisma_client,
)
@ -448,7 +622,7 @@ async def _get_team_members_display(member_ids: List[str]) -> List[SCIMMember]:
user = await UserRepository(prisma_client).table.find_unique(where={"user_id": member_id})
if user:
display_name = user.user_email or user.user_id
members.append(SCIMMember(value=user.user_id, display=display_name))
members.append(SCIMMember(value=user.user_id, display=display_name, type="User"))
return members
@ -863,6 +1037,14 @@ def _get_schemas() -> list:
type="string",
description="Member display name.",
),
SCIMSchemaAttribute(
name="type",
type="string",
description=(
'The type of member; canonical values are "User" and "Group". '
"Only members of type User are honored, LiteLLM teams contain users only."
),
),
],
),
],
@ -1336,21 +1518,42 @@ async def delete_user(
raise handle_exception_on_proxy(e)
def _extract_group_values(value: Any) -> List[str]:
def _parse_member_entry(entry: object) -> SCIMMember | None:
"""Parse one entry of a SCIM patch value, or None when it carries no id."""
if isinstance(entry, str):
return SCIMMember(value=entry)
fields = _json_object_fields(entry)
if fields is None:
return None
entry_value = fields.get("value")
if not entry_value:
return None
entry_display = fields.get("display")
entry_type = fields.get("type")
return SCIMMember(
value=str(entry_value),
display=str(entry_display) if entry_display is not None else None,
type=entry_type if isinstance(entry_type, str) else None,
)
def _parse_member_entries(value: object) -> tuple[SCIMMember, ...]:
"""Parse a SCIM patch value into members, keeping each entry's ``type``.
PATCH bodies bypass SCIMGroup parsing (SCIMPatchOperation.value is untyped),
so member objects arrive as raw dicts and the ``type`` that marks a nested
group would otherwise be lost.
"""
entries: tuple[object, ...] = tuple(value) if isinstance(value, list) else (value,)
return tuple(member for member in (_parse_member_entry(entry) for entry in entries) if member is not None)
def _extract_group_values(value: object) -> List[str]:
"""Return group ids from a SCIM patch value."""
group_values: List[str] = []
if isinstance(value, list):
for v in value:
if isinstance(v, dict) and v.get("value"):
group_values.append(str(v.get("value")))
elif isinstance(v, str):
group_values.append(v)
elif isinstance(value, dict):
if value.get("value"):
group_values.append(str(value.get("value")))
elif isinstance(value, str):
group_values.append(value)
return group_values
return [member.value for member in _parse_member_entries(value)]
def _extract_ids_from_path_filter(path: str | None, attribute: str) -> List[str]:
@ -1833,6 +2036,7 @@ async def create_group(
team_id=team_id,
team_alias=group.displayName,
members_with_roles=members_with_roles,
metadata={SCIM_MANAGED_TEAM_METADATA_KEY: True},
),
http_request=Request(scope={"type": "http", "path": "/scim/v2/Groups"}),
user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN),
@ -1875,7 +2079,11 @@ async def update_group(
# Prepare update data
existing_metadata = existing_team.metadata if existing_team.metadata else {}
updated_metadata = {**existing_metadata, "scim_data": group.model_dump()}
updated_metadata = {
**existing_metadata,
SCIM_TEAM_DATA_METADATA_KEY: group.model_dump(),
SCIM_MANAGED_TEAM_METADATA_KEY: True,
}
update_data = {
"team_alias": group.displayName,
@ -1968,12 +2176,17 @@ async def _process_group_patch_operations(
is absolute: it declares the roster is exactly this set, so the caller must
reconcile against it as a set-to-target rather than rebasing it onto a
concurrently-mutated roster.
A ``remove`` drops the ids it names without resolving them first. Removal is
idempotent and cannot put anything on a roster, while resolving would make it
conditional on what the id turns out to be and leave members we should never
have admitted - the phantom users this endpoint used to create for nested
groups - impossible to clean up.
"""
update_data: Dict[str, Any] = {}
# Create a fresh copy of existing metadata to avoid Prisma issues
existing_metadata = existing_team.metadata or {}
metadata = dict(existing_metadata) if existing_metadata else {}
metadata = {**(existing_team.metadata or {}), SCIM_MANAGED_TEAM_METADATA_KEY: True}
# Track member changes. members_with_roles is the source of truth for team
# membership; the legacy `members` column is not populated by team creation
@ -2001,50 +2214,26 @@ async def _process_group_patch_operations(
metadata["externalId"] = str(value)
elif path.startswith("members"):
# Handle member operations
member_values = _extract_group_values(value)
if not member_values and value is None:
member_values = _extract_ids_from_path_filter(op.path, "members")
# Check the feature flag
scim_upsert_user = await _get_scim_upsert_user_setting()
# Validate all users exist or create them based on feature flag
valid_members = []
for member_id in member_values:
# Validate member_id is not empty
if not member_id or not member_id.strip():
raise HTTPException(
status_code=400,
detail={"error": "Invalid member: user ID cannot be empty."},
)
patched_members = (
_parse_member_entries(value)
if value is not None
else tuple(
SCIMMember(value=member_id) for member_id in _extract_ids_from_path_filter(op.path, "members")
)
)
user = await UserRepository(prisma_client).table.find_unique(where={"user_id": member_id})
if user:
valid_members.append(member_id)
else:
if scim_upsert_user:
# Create the user if they don't exist (backward compatible behavior)
created_user = await _create_user_if_not_exists(
user_id=member_id, created_via="scim_group_patch"
)
if created_user:
valid_members.append(member_id)
# If creation failed, user is skipped (logged in helper)
else:
# User doesn't exist - reject per SCIM 2.0 protocol
raise HTTPException(
status_code=400,
detail={
"error": f"User with ID '{member_id}' does not exist. "
"Please create the user first via POST /Users before adding to group."
},
)
if op_type == "replace":
final_members = set(valid_members)
elif op_type == "add":
final_members.update(valid_members)
elif op_type == "remove":
for member_id in valid_members:
final_members.discard(member_id)
if op_type == "remove":
final_members = final_members - {_member_value(member) for member in patched_members}
else:
member_result = await _resolve_group_member_ids(
members=patched_members,
created_via="scim_group_patch",
prisma_client=prisma_client,
)
if op_type == "replace":
final_members = set(member_result.all_member_ids)
elif op_type == "add":
final_members = final_members | set(member_result.all_member_ids)
else:
# Handle other generic metadata
if op_type == "remove":
@ -2052,9 +2241,7 @@ async def _process_group_patch_operations(
else:
metadata[path] = value
# Include metadata in update data if it exists
if metadata:
update_data["metadata"] = metadata
update_data["metadata"] = metadata
member_replace_present = any(
op.op == "replace" and (op.path or "").lower().startswith("members") for op in patch_ops.Operations

12
litellm/types/proxy/management_endpoints/scim_v2.py
View file

@ -18,6 +18,9 @@ SCIM_ENTERPRISE_METADATA_KEY = "scim_enterprise"
SCIM_ENTITLEMENTS_METADATA_KEY = "scim_entitlements"
SCIM_ROLES_METADATA_KEY = "scim_roles"
SCIM_MANAGED_TEAM_METADATA_KEY = "scim_managed"
SCIM_TEAM_DATA_METADATA_KEY = "scim_data"
class LiteLLM_UserScimMetadata(BaseModel):
"""
@ -131,6 +134,15 @@ class SCIMUser(SCIMResource):
class SCIMMember(BaseModel):
value: str # User ID
display: Optional[str] = None # Username or email
type: str | None = None
@field_validator("type", mode="before")
@classmethod
def normalize_type(cls, v: object) -> str | None:
"""Anything that is not a string carries no canonical type, and rejecting the
request over it would be a regression: before this field existed the value was
parsed away silently."""
return v if isinstance(v, str) else None
class SCIMGroup(SCIMResource):

15
tests/test_litellm/proxy/management_endpoints/scim/test_scim_transformations.py
View file

@ -332,6 +332,21 @@ class TestScimTransformations:
assert scim_group.members[1].value == "test2@example.com"
assert scim_group.members[1].display == "test2@example.com"
@pytest.mark.asyncio
async def test_transform_team_marks_members_as_users(
self, mock_team, mock_prisma_client
):
"""A LiteLLM team only holds users, and stating the member type keeps the
response from emitting a null ``type`` now that SCIMMember carries one."""
mock_client, _ = mock_prisma_client
with patch("litellm.proxy.proxy_server.prisma_client", mock_client):
scim_group = await ScimTransformations.transform_litellm_team_to_scim_group(
mock_team
)
assert [member.type for member in scim_group.members] == ["User", "User"]
def test_get_scim_user_name(self, mock_user, mock_user_minimal):
# User with email
result = ScimTransformations._get_scim_user_name(mock_user)

13
tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_discovery.py
View file

@ -108,6 +108,19 @@ class TestGetSchemas:
assert "displayName" in attr_names
assert "members" in attr_names
def test_group_schema_advertises_member_type(self):
"""IdPs read the schema to learn we understand ``members.type``, which is how
a nested group announces itself."""
schemas = _get_schemas()
group_schema = next(
s for s in schemas if s.id == "urn:ietf:params:scim:schemas:core:2.0:Group"
)
members = next(a for a in group_schema.attributes if a.name == "members")
member_type = next(a for a in members.subAttributes or [] if a.name == "type")
assert member_type.type == "string"
assert member_type.multiValued is False
assert "Group" in (member_type.description or "")
def test_schema_meta_fields(self):
schemas = _get_schemas()
user_schema = next(

884
tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py
View file

@ -20,8 +20,10 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
_extract_group_member_ids,
_extract_ids_from_path_filter,
_handle_team_membership_changes,
_parse_member_entries,
_process_group_patch_operations,
_recompute_scim_member_roles,
_resolve_group_member_ids,
create_group,
create_user,
delete_group,
@ -36,6 +38,8 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
)
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_USER_SCHEMA,
SCIM_MANAGED_TEAM_METADATA_KEY,
SCIM_TEAM_DATA_METADATA_KEY,
SCIMGroup,
SCIMMember,
SCIMPatchOp,
@ -1611,7 +1615,10 @@ async def test_update_group_with_nonexistent_users_rejects(mocker, monkeypatch):
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
# Mock team operations
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=mock_existing_team)
def mock_team_lookup(where):
return mock_existing_team if where["team_id"] == group_id else None
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(side_effect=mock_team_lookup)
# Mock updated team response
mock_updated_team = mocker.MagicMock()
@ -1775,6 +1782,7 @@ async def test_extract_group_member_ids_with_flag_true_creates_users(mocker, mon
mock_prisma_client = mocker.MagicMock()
mock_prisma_client.db = mocker.MagicMock()
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
# Mock user lookup - only existing-user exists initially
def mock_user_lookup(where):
@ -1842,6 +1850,7 @@ async def test_extract_group_member_ids_with_flag_false_rejects(mocker, monkeypa
mock_prisma_client = mocker.MagicMock()
mock_prisma_client.db = mocker.MagicMock()
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
# Mock user lookup - only existing-user exists
def mock_user_lookup(where):
@ -1902,6 +1911,7 @@ async def test_process_group_patch_operations_with_flag_true_creates_users(mocke
# Mock user lookup - new-user-1 doesn't exist
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
# Mock user creation
created_user = NewUserResponse(user_id="new-user-1", key="test-key-1")
@ -1956,6 +1966,7 @@ async def test_process_group_patch_operations_with_flag_false_rejects(mocker, mo
# Mock user lookup - new-user-1 doesn't exist
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
# Execute the function - should raise HTTPException
with pytest.raises(HTTPException) as exc_info:
@ -3519,3 +3530,874 @@ async def test_process_group_patch_replace_empty_value_does_not_use_path_filter(
)
assert final_members == set()
def _member_resolution_prisma(mocker, *, users: set, teams: set, unmanaged_teams: frozenset = frozenset()):
"""Prisma mock where only the given ids resolve to a user row / team row.
``teams`` are teams a SCIM group write created, so they carry provenance;
``unmanaged_teams`` resolve too but look like a team an admin created here.
"""
def team_row(team_id: str) -> LiteLLM_TeamTable | None:
if team_id in teams:
return LiteLLM_TeamTable(team_id=team_id, metadata={SCIM_MANAGED_TEAM_METADATA_KEY: True})
if team_id in unmanaged_teams:
return LiteLLM_TeamTable(team_id=team_id, metadata={})
return None
prisma_client = mocker.MagicMock()
prisma_client.db = mocker.MagicMock()
prisma_client.db.litellm_usertable = mocker.MagicMock()
prisma_client.db.litellm_usertable.find_unique = AsyncMock(
side_effect=lambda where: (
LiteLLM_UserTable(user_id=where["user_id"]) if where["user_id"] in users else None
)
)
prisma_client.db.litellm_teamtable = mocker.MagicMock()
prisma_client.db.litellm_teamtable.find_unique = AsyncMock(side_effect=lambda where: team_row(where["team_id"]))
return prisma_client
@pytest.fixture
def scim_upsert_user_enabled(monkeypatch):
from litellm.proxy.proxy_server import proxy_config
async def mock_get_config():
return {"litellm_settings": {"scim_upsert_user": True}}
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
@pytest.fixture
def scim_upsert_user_disabled(monkeypatch):
from litellm.proxy.proxy_server import proxy_config
async def mock_get_config():
return {"litellm_settings": {"scim_upsert_user": False}}
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
@pytest.mark.asyncio
async def test_create_group_ignores_nested_group_members(mocker, scim_upsert_user_enabled):
"""Entra sends nested groups as members with ``type: "Group"``. Treating that
GUID as a user id provisioned a phantom internal user per nested group."""
nested_group_id = "8f1e9d70-0000-4a0e-9a1e-nested"
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id="parent-group",
displayName="Parent Group",
members=[
SCIMMember(value="real-user", display="Real User", type="User"),
SCIMMember(value=nested_group_id, display="Nested Group", type="Group"),
],
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=_member_resolution_prisma(mocker, users={"real-user"}, teams=set())),
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
new_team_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.new_team",
AsyncMock(return_value=mocker.MagicMock()),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
AsyncMock(return_value=scim_group),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles",
AsyncMock(),
)
await create_group(group=scim_group)
create_user_mock.assert_not_called()
assert new_team_mock.call_args.kwargs["data"].members_with_roles == [Member(user_id="real-user", role="user")]
@pytest.mark.asyncio
async def test_update_group_ignores_nested_group_members(mocker, scim_upsert_user_enabled):
"""PUT /Groups must drop nested-group members too, so a full sync from the IdP
neither provisions nor enrolls the nested group's GUID."""
group_id = "parent-group"
nested_group_id = "8f1e9d70-0000-4a0e-9a1e-nested"
existing_team = LiteLLM_TeamTable(
team_id=group_id,
team_alias="Parent Group",
members=[],
members_with_roles=[],
metadata={},
)
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id=group_id,
displayName="Parent Group",
members=[
SCIMMember(value="real-user", display="Real User", type="User"),
SCIMMember(value=nested_group_id, display="Nested Group", type="Group"),
],
)
prisma_client = _member_resolution_prisma(mocker, users={"real-user"}, teams={group_id})
prisma_client.db.litellm_teamtable.update = AsyncMock(return_value=existing_team)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=prisma_client),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._check_team_exists",
AsyncMock(return_value=existing_team),
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
patch_membership_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership",
AsyncMock(),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles",
AsyncMock(),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
AsyncMock(return_value=scim_group),
)
await update_group(group_id=group_id, group=scim_group)
create_user_mock.assert_not_called()
enrolled = {call.kwargs["user_id"] for call in patch_membership_mock.call_args_list}
assert enrolled == {"real-user"}
@pytest.mark.asyncio
async def test_process_group_patch_operations_ignores_nested_group_members(mocker, scim_upsert_user_enabled):
"""PATCH bodies bypass SCIMGroup parsing, so ``type`` must be read off the raw
member dicts; otherwise a nested group is indistinguishable from a user id."""
nested_group_id = "8f1e9d70-0000-4a0e-9a1e-nested"
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[
SCIMPatchOperation(
op="add",
path="members",
value=[
{"value": "real-user", "display": "Real User", "type": "User"},
{"value": nested_group_id, "display": "Nested Group", "type": "Group"},
],
)
],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[Member(user_id="incumbent", role="user")],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"real-user"}, teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == {"incumbent", "real-user"}
@pytest.mark.asyncio
async def test_process_group_patch_operations_ignores_lowercase_group_type(mocker, scim_upsert_user_enabled):
"""The ``type`` comparison is case-insensitive; IdPs are not consistent about it."""
nested_group_id = "8f1e9d70-0000-4a0e-9a1e-nested"
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[
SCIMPatchOperation(op="add", path="members", value=[{"value": nested_group_id, "type": "group"}])
],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users=set(), teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == set()
@pytest.mark.asyncio
async def test_process_group_patch_operations_skips_member_matching_existing_team(mocker, scim_upsert_user_enabled):
"""Okta sends filtered paths and untyped ids, so a nested group arrives with no
``type`` at all; an id that names an existing team is still not a user."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "child-team"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[Member(user_id="incumbent", role="user")],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users=set(), teams={"child-team", "parent-group"}),
)
create_user_mock.assert_not_called()
assert final_members == {"incumbent"}
@pytest.mark.asyncio
async def test_process_group_patch_operations_prefers_user_over_team_for_colliding_id(
mocker, scim_upsert_user_enabled
):
"""Nothing stops a user id from also being a team id, so the user lookup has to
win; ordering the team check first would silently stop syncing that user."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "dual-id"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"dual-id"}, teams={"dual-id"}),
)
assert final_members == {"dual-id"}
@pytest.mark.asyncio
async def test_create_group_strict_mode_accepts_group_and_team_members(mocker, scim_upsert_user_disabled):
"""Strict mode (scim_upsert_user=False) rejects unknown *users*; a nested group
is not a user, so it must be dropped rather than 400 the whole sync."""
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id="parent-group",
displayName="Parent Group",
members=[
SCIMMember(value="real-user", type="User"),
SCIMMember(value="nested-group-guid", type="Group"),
SCIMMember(value="child-team"),
],
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=_member_resolution_prisma(mocker, users={"real-user"}, teams={"child-team"})),
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
new_team_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.new_team",
AsyncMock(return_value=mocker.MagicMock()),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
AsyncMock(return_value=scim_group),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles",
AsyncMock(),
)
await create_group(group=scim_group)
create_user_mock.assert_not_called()
assert new_team_mock.call_args.kwargs["data"].members_with_roles == [Member(user_id="real-user", role="user")]
@pytest.mark.asyncio
async def test_create_group_strict_mode_still_rejects_unknown_user(mocker, scim_upsert_user_disabled):
"""The strict-mode 400 must name the unknown *user* and stay quiet about the
nested group sharing the request."""
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id="parent-group",
displayName="Parent Group",
members=[
SCIMMember(value="nested-group-guid", type="Group"),
SCIMMember(value="unknown-user"),
],
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=_member_resolution_prisma(mocker, users=set(), teams=set())),
)
with pytest.raises(ProxyException) as exc_info:
await create_group(group=scim_group)
assert int(exc_info.value.code) == 400
assert "unknown-user" in str(exc_info.value.message)
assert "nested-group-guid" not in str(exc_info.value.message)
@pytest.mark.asyncio
async def test_process_group_patch_remove_unknown_member_does_not_create_user(mocker, scim_upsert_user_enabled):
"""A ``remove`` of an id we don't know is an idempotent no-op. Upserting the id
first, only to drop it from the roster, made removals a phantom-user factory."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="remove", path="members", value=[{"value": "long-gone"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[Member(user_id="keep-user", role="user")],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"keep-user"}, teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == {"keep-user"}
@pytest.mark.parametrize(
"operation",
[
SCIMPatchOperation(op="remove", path='members[value eq "long-gone"]', value=None),
SCIMPatchOperation(op="remove", path="members", value=[{"value": "long-gone"}]),
SCIMPatchOperation(op="remove", path="members", value=[{"value": "long-gone", "type": "Group"}]),
],
ids=["path-filter", "unknown-id", "nested-group"],
)
@pytest.mark.asyncio
async def test_process_group_patch_remove_unknown_member_does_not_reject_in_strict_mode(
mocker, scim_upsert_user_disabled, operation
):
"""Strict mode must not 400 a removal: refusing to drop an id the IdP already
forgot leaves the roster permanently out of sync."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[operation],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[Member(user_id="keep-user", role="user")],
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"keep-user"}, teams=set()),
)
assert final_members == {"keep-user"}
@pytest.mark.asyncio
async def test_process_group_patch_remove_drops_member_without_user_row(mocker, scim_upsert_user_enabled):
"""Phantom members already on a roster (their user row is gone) must still be
removable, so the removal id is honoured even though it resolves to nothing."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="remove", path="members", value=[{"value": "phantom"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[
Member(user_id="keep-user", role="user"),
Member(user_id="phantom", role="user"),
],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"keep-user"}, teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == {"keep-user"}
_NESTED_GROUP_ID = "8f1e9d70-0000-4a0e-9a1e-nested"
@pytest.mark.parametrize(
"member_entry, user_rows, team_rows",
[
({"value": _NESTED_GROUP_ID, "type": "Group"}, {"keep-user", _NESTED_GROUP_ID}, set()),
({"value": _NESTED_GROUP_ID, "type": "Group"}, {"keep-user"}, set()),
({"value": _NESTED_GROUP_ID, "type": "Group"}, {"keep-user"}, {_NESTED_GROUP_ID}),
({"value": _NESTED_GROUP_ID}, {"keep-user"}, {_NESTED_GROUP_ID}),
],
ids=["phantom-user-row-exists", "user-row-already-deleted", "child-group-is-a-team", "untyped-team-id"],
)
@pytest.mark.asyncio
async def test_process_group_patch_remove_discards_non_user_member(
mocker, scim_upsert_user_enabled, member_entry, user_rows, team_rows
):
"""Rosters written before nested groups were understood still carry those ids,
and the IdP removes them exactly as it added them; a removal that resolved its
ids first would classify them as non-users and leave them stuck on the team."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="remove", path="members", value=[member_entry])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[
Member(user_id="keep-user", role="user"),
Member(user_id=_NESTED_GROUP_ID, role="user"),
],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users=user_rows, teams=team_rows),
)
create_user_mock.assert_not_called()
assert final_members == {"keep-user"}
@pytest.mark.asyncio
async def test_process_group_patch_add_keeps_member_typed_user_that_collides_with_team_id(
mocker, scim_upsert_user_enabled
):
"""The team lookup only exists to catch nested groups that arrive untyped. An id
the IdP calls a User is a user, and IdP ids collide with team ids easily."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "123456", "type": "User"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="123456", key="new-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users=set(), teams={"123456", "parent-group"}),
)
assert create_user_mock.call_args.kwargs["user_id"] == "123456"
assert final_members == {"123456"}
@pytest.mark.parametrize("member_type", ["Device", " group ", "Machine"])
@pytest.mark.asyncio
async def test_process_group_patch_operations_skips_non_user_member_types(
mocker, scim_upsert_user_enabled, member_type
):
"""A team holds users, so a member that declares itself to be anything else is
dropped; enumerating the types worth skipping would leave the next one to leak."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "not-a-user", "type": member_type}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users=set(), teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == set()
@pytest.mark.parametrize(
"team_metadata, expect_provisioned",
[
({SCIM_MANAGED_TEAM_METADATA_KEY: True}, False),
({SCIM_TEAM_DATA_METADATA_KEY: {"displayName": "Child.Apps"}}, False),
({}, True),
(None, True),
({SCIM_MANAGED_TEAM_METADATA_KEY: False}, True),
({SCIM_TEAM_DATA_METADATA_KEY: None}, True),
],
ids=[
"scim-managed",
"legacy-scim-data",
"admin-created",
"no-metadata",
"marker-unset",
"legacy-key-without-value",
],
)
@pytest.mark.asyncio
async def test_process_group_patch_team_match_needs_scim_provenance(
mocker, scim_upsert_user_enabled, team_metadata, expect_provisioned
):
"""A bare member id that names a team is only evidence of a nested group when the
identity provider is what wrote that team. Teams created here can share an id with
a real user, and skipping those members stops provisioning them entirely."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "child-team"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
prisma_client = _member_resolution_prisma(mocker, users=set(), teams=set())
prisma_client.db.litellm_teamtable.find_unique = AsyncMock(
return_value=LiteLLM_TeamTable(team_id="child-team", metadata=team_metadata)
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="child-team", key="new-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=prisma_client,
)
assert create_user_mock.called is expect_provisioned
assert final_members == ({"child-team"} if expect_provisioned else set())
@pytest.mark.asyncio
async def test_create_group_strict_mode_rejects_id_matching_admin_created_team(mocker, scim_upsert_user_disabled):
"""Strict mode drops nested groups but reports unknown users. A team an admin
created here says nothing about the member, so the member is an unknown user."""
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id="parent-group",
displayName="Parent Group",
members=[SCIMMember(value="admin-team")],
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(
return_value=_member_resolution_prisma(
mocker, users=set(), teams=set(), unmanaged_teams=frozenset({"admin-team"})
)
),
)
with pytest.raises(ProxyException) as exc_info:
await create_group(group=scim_group)
assert int(exc_info.value.code) == 400
assert "admin-team" in str(exc_info.value.message)
@pytest.mark.asyncio
async def test_create_group_stamps_scim_provenance(mocker, scim_upsert_user_enabled):
"""The provenance the classifier reads only exists if the group writes stamp it;
a SCIM-created team that carries no mark looks admin-created forever after."""
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id="child-group",
displayName="Child.Apps",
members=[],
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=_member_resolution_prisma(mocker, users=set(), teams=set())),
)
new_team_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.new_team",
AsyncMock(return_value=mocker.MagicMock()),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
AsyncMock(return_value=scim_group),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles",
AsyncMock(),
)
await create_group(group=scim_group)
assert new_team_mock.call_args.kwargs["data"].metadata == {SCIM_MANAGED_TEAM_METADATA_KEY: True}
@pytest.mark.asyncio
async def test_update_group_stamps_scim_provenance(mocker, scim_upsert_user_enabled):
"""A PUT full sync adopts a team the identity provider now owns, and the stamp has
to land alongside the existing metadata rather than replacing it."""
import json
group_id = "child-group"
existing_team = LiteLLM_TeamTable(
team_id=group_id,
team_alias="Child.Apps",
members=[],
members_with_roles=[],
metadata={"existing_key": "kept"},
)
scim_group = SCIMGroup(
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
id=group_id,
displayName="Child.Apps",
members=[],
)
prisma_client = _member_resolution_prisma(mocker, users=set(), teams=set())
prisma_client.db.litellm_teamtable.update = AsyncMock(return_value=existing_team)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=prisma_client),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._check_team_exists",
AsyncMock(return_value=existing_team),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles",
AsyncMock(),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
AsyncMock(return_value=scim_group),
)
await update_group(group_id=group_id, group=scim_group)
written = json.loads(prisma_client.db.litellm_teamtable.update.call_args.kwargs["data"]["metadata"])
assert written[SCIM_MANAGED_TEAM_METADATA_KEY] is True
assert written["existing_key"] == "kept"
assert SCIM_TEAM_DATA_METADATA_KEY in written
@pytest.mark.asyncio
async def test_process_group_patch_stamps_scim_provenance(mocker, scim_upsert_user_enabled):
"""PATCH is how Okta adopts a group, so a membership-only patch has to stamp the
team too; otherwise the group it manages never gains provenance."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "real-user"}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
metadata={"existing_key": "kept"},
)
update_data, _, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"real-user"}, teams=set()),
)
assert update_data["metadata"][SCIM_MANAGED_TEAM_METADATA_KEY] is True
assert update_data["metadata"]["existing_key"] == "kept"
@pytest.mark.parametrize("member_type", ["direct", "Device"])
@pytest.mark.asyncio
async def test_process_group_patch_keeps_existing_user_with_unrecognized_type(
mocker, scim_upsert_user_enabled, member_type
):
"""Clients do stamp non-canonical types on real members (RFC 7643 defines
``direct`` for ``User.groups``). Dropping a member whose id is a live user would
revoke that user's team access on the next full sync, so the type is only
grounds for skipping once the user lookup has missed."""
patch_ops = SCIMPatchOp(
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "real-user", "type": member_type}])],
)
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[],
)
create_user_mock = mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(return_value=NewUserResponse(user_id="phantom-user", key="phantom-key")),
)
_, final_members, _ = await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"real-user"}, teams=set()),
)
create_user_mock.assert_not_called()
assert final_members == {"real-user"}
@pytest.mark.parametrize(
"second_creation",
[None, NewUserResponse(user_id="dup-user", key="second-key")],
ids=["second-creation-fails", "both-creations-succeed"],
)
@pytest.mark.asyncio
async def test_resolve_group_member_ids_dedupes_repeated_member(mocker, scim_upsert_user_enabled, second_creation):
"""An id the request lists twice is one member. Admitting it twice writes a
duplicate members_with_roles row, and the second creation of the same id fails
against the real unique constraint even when the first one succeeded."""
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
AsyncMock(side_effect=[NewUserResponse(user_id="dup-user", key="first-key"), second_creation]),
)
result = await _resolve_group_member_ids(
members=[SCIMMember(value="dup-user"), SCIMMember(value="dup-user")],
created_via="scim_group_membership",
prisma_client=_member_resolution_prisma(mocker, users=set(), teams=set()),
)
assert result.all_member_ids == ["dup-user"]
@pytest.mark.parametrize(
"operation",
[
SCIMPatchOperation(op="add", path="members", value=[{"value": " "}]),
SCIMPatchOperation(op="remove", path="members", value=[{"value": " "}]),
SCIMPatchOperation(op="remove", path='members[value eq " "]', value=None),
],
ids=["add", "remove", "remove-path-filter"],
)
@pytest.mark.asyncio
async def test_process_group_patch_rejects_blank_member_id(mocker, scim_upsert_user_enabled, operation):
"""A blank id names nobody. The removal path stopped resolving its members, so it
has to keep rejecting one on its own."""
patch_ops = SCIMPatchOp(schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], Operations=[operation])
existing_team = LiteLLM_TeamTable(
team_id="parent-group",
team_alias="Parent Group",
members=[],
members_with_roles=[Member(user_id="keep-user", role="user")],
)
with pytest.raises(HTTPException) as exc_info:
await _process_group_patch_operations(
patch_ops=patch_ops,
existing_team=existing_team,
prisma_client=_member_resolution_prisma(mocker, users={"keep-user"}, teams=set()),
)
assert exc_info.value.status_code == 400
def test_scim_member_round_trips_type():
"""``type`` has to survive parsing; dropping it is what made a nested group
look like a user id."""
assert SCIMMember.model_validate({"value": "x", "type": "Group"}).type == "Group"
assert SCIMMember(value="x").type is None
@pytest.mark.parametrize("junk_type", [123, True, {}, [], 1.5])
def test_scim_member_treats_non_string_type_as_absent(junk_type):
"""Before ``type`` was a field, junk in it was parsed away; typing the field must
not start rejecting those requests, and both parsers have to agree it is typeless."""
assert SCIMMember.model_validate({"value": "x", "type": junk_type}).type is None
assert _parse_member_entries([{"value": "x", "type": junk_type}])[0].type is None
@pytest.mark.asyncio
async def test_get_groups_members_are_typed_as_users(mocker):
"""Group members we report back are always users, and saying so keeps the
response from emitting a null ``type``."""
team = LiteLLM_TeamTable(
team_id="team-1",
team_alias="Team One",
members=[],
members_with_roles=[Member(user_id="member-1", role="user")],
)
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_many = AsyncMock(return_value=[team])
mock_prisma_client.db.litellm_teamtable.count = AsyncMock(return_value=1)
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(
return_value=mocker.MagicMock(user_id="member-1", user_email="member-1@example.com")
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=mock_prisma_client),
)
response = await get_groups(startIndex=1, count=10, filter=None)
assert [m.type for m in response.Resources[0].members] == ["User"]