scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 109

feat(scim): ingest and round-trip SCIM entitlements and roles user attributes

Browse source
This commit is contained in:
Tin Chi Lo 2026-07-16 13:50:00 -07:00
parent a8ae515bee
commit ccfa78046a
7 changed files with 373 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -61,6 +61,8 @@ from litellm.types.proxy.management_endpoints.common_daily_activity import (
)
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_METADATA_KEY,
SCIM_ENTITLEMENTS_METADATA_KEY,
SCIM_ROLES_METADATA_KEY,
)
from litellm.types.proxy.management_endpoints.internal_user_endpoints import (
BulkUpdateUserRequest,
@ -690,15 +692,21 @@ async def _get_user_info_teams(
return team_list, teams_1
_SCIM_DIRECTORY_METADATA_KEYS = frozenset(
{SCIM_ENTERPRISE_METADATA_KEY, SCIM_ENTITLEMENTS_METADATA_KEY, SCIM_ROLES_METADATA_KEY}
)
def _redact_scim_enterprise_metadata(
metadata: Optional[Dict[str, Any]],
) -> Optional[Dict[str, Any]]:
"""SCIM enterprise attributes are persisted in user metadata so reporting can
group on them, but they are directory-only fields that generic user-info
endpoints must not surface; SCIM clients read them through the SCIM endpoints."""
if not isinstance(metadata, dict) or SCIM_ENTERPRISE_METADATA_KEY not in metadata:
"""SCIM enterprise attributes, entitlements, and roles are persisted in user
metadata so reporting can group on them, but they are directory-only fields
that generic user-info endpoints must not surface; SCIM clients read them
through the SCIM endpoints."""
if not isinstance(metadata, dict) or not _SCIM_DIRECTORY_METADATA_KEYS.intersection(metadata):
return metadata
return {k: v for k, v in metadata.items() if k != SCIM_ENTERPRISE_METADATA_KEY}
return {k: v for k, v in metadata.items() if k not in _SCIM_DIRECTORY_METADATA_KEYS}
def _build_user_info_response(

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

@ -52,6 +52,12 @@ class ScimTransformations:
enterprise_user = SCIMEnterpriseUser.model_validate(metadata[SCIM_ENTERPRISE_METADATA_KEY])
schemas.append(SCIM_ENTERPRISE_USER_SCHEMA)
raw_entitlements = metadata.get(SCIM_ENTITLEMENTS_METADATA_KEY)
entitlements = SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python(raw_entitlements) if raw_entitlements else None
raw_roles = metadata.get(SCIM_ROLES_METADATA_KEY)
roles = SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python(raw_roles) if raw_roles else None
return SCIMUser(
schemas=schemas,
id=user.user_id,
@ -64,6 +70,8 @@ class ScimTransformations:
emails=emails,
groups=groups,
active=active,
entitlements=entitlements,
roles=roles,
enterprise_user=enterprise_user,
meta={
"resourceType": "User",

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

@ -17,7 +17,7 @@ from fastapi import (
Request,
Response,
)
from pydantic import BaseModel
from pydantic import BaseModel, ValidationError
from typing_extensions import TypedDict
import litellm
@ -125,6 +125,8 @@ class ScimUserData(TypedDict):
family_name: Optional[str]
active: Optional[bool]
enterprise: Optional[SCIMEnterpriseUser]
entitlements: list[SCIMMultiValuedAttribute] | None
roles: list[SCIMMultiValuedAttribute] | None
class GroupMemberExtractionResult(BaseModel):
@ -199,6 +201,8 @@ def _extract_scim_user_data(user: SCIMUser) -> ScimUserData:
"family_name": user.name.familyName if user.name else None,
"active": user.active,
"enterprise": user.enterprise_user,
"entitlements": user.entitlements,
"roles": user.roles,
}
@ -207,6 +211,8 @@ def _build_scim_metadata(
family_name: Optional[str],
active: Optional[bool] = None,
enterprise: Optional[SCIMEnterpriseUser] = None,
entitlements: list[SCIMMultiValuedAttribute] | None = None,
roles: list[SCIMMultiValuedAttribute] | None = None,
) -> Dict[str, Any]:
"""Build metadata dictionary with SCIM data."""
metadata: Dict[str, Any] = {
@ -222,6 +228,12 @@ def _build_scim_metadata(
if enterprise is not None:
metadata[SCIM_ENTERPRISE_METADATA_KEY] = enterprise.model_dump(by_alias=True, exclude_none=True)
if entitlements is not None:
metadata[SCIM_ENTITLEMENTS_METADATA_KEY] = [e.model_dump(exclude_none=True) for e in entitlements]
if roles is not None:
metadata[SCIM_ROLES_METADATA_KEY] = [r.model_dump(exclude_none=True) for r in roles]
return metadata
@ -739,6 +751,62 @@ def _get_schemas() -> list:
),
],
),
SCIMSchemaAttribute(
name="entitlements",
type="complex",
multiValued=True,
description="A list of entitlements for the user.",
subAttributes=[
SCIMSchemaAttribute(
name="value",
type="string",
description="The value of an entitlement.",
),
SCIMSchemaAttribute(
name="display",
type="string",
description="A human-readable name for the entitlement.",
),
SCIMSchemaAttribute(
name="type",
type="string",
description="A label indicating the entitlement's function.",
),
SCIMSchemaAttribute(
name="primary",
type="boolean",
description="Whether this is the primary entitlement.",
),
],
),
SCIMSchemaAttribute(
name="roles",
type="complex",
multiValued=True,
description="A list of roles for the user.",
subAttributes=[
SCIMSchemaAttribute(
name="value",
type="string",
description="The value of a role.",
),
SCIMSchemaAttribute(
name="display",
type="string",
description="A human-readable name for the role.",
),
SCIMSchemaAttribute(
name="type",
type="string",
description="A label indicating the role's function.",
),
SCIMSchemaAttribute(
name="primary",
type="boolean",
description="Whether this is the primary role.",
),
],
),
],
meta={
"location": "/scim/v2/Schemas/urn:ietf:params:scim:schemas:core:2.0:User",
@ -1074,6 +1142,8 @@ async def create_user(
user_data["given_name"],
user_data["family_name"],
enterprise=user_data["enterprise"],
entitlements=user_data["entitlements"],
roles=user_data["roles"],
)
default_role = _default_scim_user_role()
@ -1152,6 +1222,8 @@ async def update_user(
user_data["family_name"],
scim_active_for_metadata,
enterprise=user_data["enterprise"],
entitlements=user_data["entitlements"],
roles=user_data["roles"],
)
await _handle_team_membership_changes(
@ -1311,6 +1383,30 @@ def _handle_group_operations(op_type: str, value: Any, teams_set: Set[str]) -> O
return None
def _handle_multi_valued_attribute_update(path: str, op_type: str, value: Any, metadata: dict[str, Any]) -> None:
"""Handle add/replace/remove for the entitlements and roles multi-valued attributes."""
metadata_key = SCIM_ENTITLEMENTS_METADATA_KEY if path == "entitlements" else SCIM_ROLES_METADATA_KEY
if op_type == "remove":
metadata.pop(metadata_key, None)
return
normalized = value if isinstance(value, list) else [value]
try:
attrs = SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python(normalized)
except ValidationError:
raise HTTPException(
status_code=400,
detail={"error": f"Invalid value for {path}: expected a list of objects with a 'value' sub-attribute"},
)
dumped = [attr.model_dump(exclude_none=True) for attr in attrs]
existing = metadata.get(metadata_key)
if op_type == "add" and isinstance(existing, list):
metadata[metadata_key] = existing + dumped
return
metadata[metadata_key] = dumped
def _handle_generic_metadata(path: str, op_type: str, value: Any, metadata: Dict[str, Any]) -> None:
"""Handle generic metadata operations for unknown paths."""
if op_type == "remove":
@ -1346,6 +1442,8 @@ def _apply_patch_ops(
_handle_displayname_update(op_type, val, update_data)
elif key_lower == "externalid":
_handle_externalid_update(op_type, val, update_data)
elif key_lower in ("entitlements", "roles"):
_handle_multi_valued_attribute_update(key_lower, op_type, val, metadata)
elif key_lower == "name" and isinstance(val, dict):
for name_key, name_val in val.items():
name_key_lower = name_key.lower()
@ -1366,6 +1464,8 @@ def _apply_patch_ops(
_handle_active_update(op_type, value, metadata)
elif path in ("name.givenname", "name.familyname"):
_handle_name_update(path, op_type, value, scim_metadata)
elif path in ("entitlements", "roles"):
_handle_multi_valued_attribute_update(path, op_type, value, metadata)
elif path.startswith("groups"):
new_replace_set = _handle_group_operations(op_type, value, teams_set)
if new_replace_set is not None:

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

@ -6,13 +6,17 @@ from pydantic import (
ConfigDict,
EmailStr,
Field,
TypeAdapter,
field_validator,
model_serializer,
model_validator,
)
from pydantic_core.core_schema import SerializerFunctionWrapHandler
SCIM_ENTERPRISE_USER_SCHEMA = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
SCIM_ENTERPRISE_METADATA_KEY = "scim_enterprise"
SCIM_ENTITLEMENTS_METADATA_KEY = "scim_entitlements"
SCIM_ROLES_METADATA_KEY = "scim_roles"
class LiteLLM_UserScimMetadata(BaseModel):
@ -53,6 +57,23 @@ class SCIMUserGroup(BaseModel):
type: Optional[str] = "direct" # direct or indirect
class SCIMMultiValuedAttribute(BaseModel):
value: str
display: Optional[str] = None
type: Optional[str] = None
primary: Optional[bool] = None
@model_validator(mode="before")
@classmethod
def coerce_bare_string(cls, data: object) -> object:
if isinstance(data, str):
return {"value": data}
return data
SCIM_MULTI_VALUED_LIST_ADAPTER = TypeAdapter(List[SCIMMultiValuedAttribute])
class SCIMUserManager(BaseModel):
model_config = ConfigDict(populate_by_name=True)
@ -81,6 +102,8 @@ class SCIMUser(SCIMResource):
active: bool = True
emails: Optional[List[SCIMUserEmail]] = None
groups: Optional[List[SCIMUserGroup]] = None
entitlements: Optional[List[SCIMMultiValuedAttribute]] = None
roles: Optional[List[SCIMMultiValuedAttribute]] = None
enterprise_user: Optional[SCIMEnterpriseUser] = Field(
default=None,
alias=SCIM_ENTERPRISE_USER_SCHEMA,
@ -88,11 +111,15 @@ class SCIMUser(SCIMResource):
)
@model_serializer(mode="wrap")
def _omit_absent_enterprise(self, handler: SerializerFunctionWrapHandler) -> Dict[str, Any]:
def _omit_absent_optional_blocks(self, handler: SerializerFunctionWrapHandler) -> Dict[str, Any]:
dumped = handler(self)
if self.enterprise_user is None:
dumped.pop(SCIM_ENTERPRISE_USER_SCHEMA, None)
dumped.pop("enterprise_user", None)
if self.entitlements is None:
dumped.pop("entitlements", None)
if self.roles is None:
dumped.pop("roles", None)
return dumped

104
tests/test_litellm/proxy/management_endpoints/scim/test_scim_patch_user.py
View file

@ -1,9 +1,10 @@
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import HTTPException
from litellm.proxy._types import LiteLLM_UserTable
from litellm.proxy.management_endpoints.scim.scim_v2 import patch_user
from litellm.proxy.management_endpoints.scim.scim_v2 import _apply_patch_ops, patch_user
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIMPatchOp,
SCIMPatchOperation,
@ -329,3 +330,104 @@ async def test_patch_user_multiple_fields_without_path():
assert update_data["user_alias"] == "New Display Name"
assert "" not in metadata # Ensure no empty string key
assert result.active is False
def _user_with_metadata(metadata):
return LiteLLM_UserTable(
user_id="user-mva",
user_email="mva@example.com",
user_alias=None,
teams=[],
metadata=metadata,
)
def test_apply_patch_ops_replace_entitlements_writes_canonical_key():
"""A PATCH on path=entitlements must persist under scim_entitlements, not
fall through to the generic handler's raw path key"""
patch_ops = SCIMPatchOp(
Operations=[
SCIMPatchOperation(
op="replace",
path="entitlements",
value=[{"value": "jira-software", "display": "Jira Software"}],
)
]
)
update_data, _ = _apply_patch_ops(
existing_user=_user_with_metadata({}), patch_ops=patch_ops
)
metadata = update_data["metadata"]
assert metadata["scim_entitlements"] == [
{"value": "jira-software", "display": "Jira Software"}
]
assert "entitlements" not in metadata
def test_apply_patch_ops_add_roles_appends_to_existing():
patch_ops = SCIMPatchOp(
Operations=[
SCIMPatchOperation(op="add", path="roles", value=[{"value": "admin"}])
]
)
update_data, _ = _apply_patch_ops(
existing_user=_user_with_metadata({"scim_roles": [{"value": "viewer"}]}),
patch_ops=patch_ops,
)
assert update_data["metadata"]["scim_roles"] == [
{"value": "viewer"},
{"value": "admin"},
]
def test_apply_patch_ops_remove_entitlements_clears_canonical_key():
patch_ops = SCIMPatchOp(
Operations=[SCIMPatchOperation(op="remove", path="entitlements")]
)
update_data, _ = _apply_patch_ops(
existing_user=_user_with_metadata(
{"scim_entitlements": [{"value": "jira-software"}]}
),
patch_ops=patch_ops,
)
assert "scim_entitlements" not in update_data["metadata"]
def test_apply_patch_ops_pathless_value_dict_handles_roles():
patch_ops = SCIMPatchOp(
Operations=[
SCIMPatchOperation(
op="replace",
value={"roles": [{"value": "engineering-admin", "primary": True}]},
)
]
)
update_data, _ = _apply_patch_ops(
existing_user=_user_with_metadata({}), patch_ops=patch_ops
)
assert update_data["metadata"]["scim_roles"] == [
{"value": "engineering-admin", "primary": True}
]
def test_apply_patch_ops_invalid_entitlements_value_raises_400():
patch_ops = SCIMPatchOp(
Operations=[
SCIMPatchOperation(
op="replace", path="entitlements", value=[{"display": "no value"}]
)
]
)
with pytest.raises(HTTPException) as exc_info:
_apply_patch_ops(existing_user=_user_with_metadata({}), patch_ops=patch_ops)
assert exc_info.value.status_code == 400

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

@ -15,6 +15,7 @@ from litellm.proxy.management_endpoints.scim.scim_transformations import (
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_USER_SCHEMA,
SCIMEnterpriseUser,
SCIMMultiValuedAttribute,
SCIMPatchOperation,
SCIMUser,
)
@ -179,6 +180,40 @@ class TestScimTransformations:
assert scim_user.enterprise_user.department == "Platform"
assert SCIM_ENTERPRISE_USER_SCHEMA in scim_user.schemas
@pytest.mark.asyncio
async def test_transform_user_with_entitlements_and_roles_metadata(
self, mock_prisma_client
):
mock_client, mock_find_unique = mock_prisma_client
mock_find_unique.return_value = None
user = LiteLLM_UserTable(
user_id="user-entitled",
user_email="entitled@example.com",
user_alias=None,
teams=[],
created_at=None,
updated_at=None,
metadata={
"scim_entitlements": [
{"value": "jira-software", "display": "Jira Software"}
],
"scim_roles": [{"value": "engineering-admin", "primary": True}],
},
)
with patch("litellm.proxy.proxy_server.prisma_client", mock_client):
scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
user
)
assert scim_user.entitlements is not None
assert scim_user.entitlements[0].value == "jira-software"
assert scim_user.entitlements[0].display == "Jira Software"
assert scim_user.roles is not None
assert scim_user.roles[0].value == "engineering-admin"
assert scim_user.roles[0].primary is True
@pytest.mark.asyncio
async def test_transform_user_without_enterprise_metadata_omits_schema(
self, mock_user, mock_prisma_client
@ -223,6 +258,27 @@ class TestScimTransformations:
dumped_ent = with_enterprise.model_dump(by_alias=True)
assert dumped_ent[SCIM_ENTERPRISE_USER_SCHEMA]["costCenter"] == "CC-42"
def test_scim_user_serialization_omits_absent_entitlements_and_roles(self):
without_attrs = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id="user-1",
userName="user@example.com",
)
dumped = without_attrs.model_dump(by_alias=True)
assert "entitlements" not in dumped
assert "roles" not in dumped
with_attrs = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id="user-2",
userName="entitled@example.com",
entitlements=[SCIMMultiValuedAttribute(value="jira-software")],
roles=[SCIMMultiValuedAttribute(value="engineering-admin")],
)
dumped_attrs = with_attrs.model_dump(by_alias=True)
assert dumped_attrs["entitlements"][0]["value"] == "jira-software"
assert dumped_attrs["roles"][0]["value"] == "engineering-admin"
@pytest.mark.asyncio
async def test_transform_litellm_team_to_scim_group(
self, mock_team, mock_prisma_client

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

@ -172,6 +172,70 @@ async def test_create_user_ingests_enterprise_extension(mocker, monkeypatch):
}
@pytest.mark.asyncio
async def test_create_user_ingests_entitlements_and_roles(mocker, monkeypatch):
"""A SCIM create payload carrying entitlements and roles should land in the
created user's metadata under scim_entitlements and scim_roles"""
scim_user = SCIMUser.model_validate(
{
"schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
"userName": "entitled-user",
"name": {"familyName": "User", "givenName": "Entitled"},
"emails": [{"value": "entitled@example.com"}],
"entitlements": [
{
"value": "jira-software",
"display": "Jira Software",
"type": "app",
"primary": True,
},
"bare-entitlement",
],
"roles": [{"value": "engineering-admin", "type": "role"}],
}
)
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)
monkeypatch.setattr("litellm.default_internal_user_params", None, raising=False)
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="entitled-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)
created_metadata = new_user_mock.call_args.kwargs["data"].metadata
assert created_metadata["scim_entitlements"] == [
{
"value": "jira-software",
"display": "Jira Software",
"type": "app",
"primary": True,
},
{"value": "bare-entitlement"},
]
assert created_metadata["scim_roles"] == [
{"value": "engineering-admin", "type": "role"}
]
@pytest.mark.asyncio
async def test_create_user_uses_default_internal_user_params_role(mocker, monkeypatch):
"""If role is set in default_internal_user_params, new user should use that role"""