feat(scim): ingest enterprise extension attributes into user metadata

Map the SCIM enterprise extension block
(urn:ietf:params:scim:schemas:extension:enterprise:2.0:User) onto SCIMUser
so create and PUT persist employeeNumber, costCenter, organization,
division, department, and manager into LiteLLM_UserTable.metadata under
scim_enterprise, and round-trip them back out on read. This lets financial
reporting group spend by fields like cost center and department.

The enterprise block holds directory-only HR attributes, so it is kept out
of the generic user management responses (/user/info, /v2/user/info, and
/user/list), which non-proxy-admin callers such as team and org admins can
use to read other users. The data still lands in metadata for reporting and
still round-trips through the SCIM read endpoints, which build their response
from the user row directly.

Resolves LIT-3617
This commit is contained in:
Yassin Kortam 2026-06-20 14:24:34 -07:00
parent 15aa40b36e
commit b25496efea
7 changed files with 364 additions and 6 deletions

View file

@ -57,6 +57,9 @@ from litellm.repositories.verification_token_repository import (
from litellm.types.proxy.management_endpoints.common_daily_activity import ( from litellm.types.proxy.management_endpoints.common_daily_activity import (
SpendAnalyticsPaginatedResponse, SpendAnalyticsPaginatedResponse,
) )
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_METADATA_KEY,
)
from litellm.types.proxy.management_endpoints.internal_user_endpoints import ( from litellm.types.proxy.management_endpoints.internal_user_endpoints import (
BulkUpdateUserRequest, BulkUpdateUserRequest,
BulkUpdateUserResponse, BulkUpdateUserResponse,
@ -719,6 +722,17 @@ async def _get_user_info_teams(
return team_list, teams_1 return team_list, teams_1
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:
return metadata
return {k: v for k, v in metadata.items() if k != SCIM_ENTERPRISE_METADATA_KEY}
def _build_user_info_response( def _build_user_info_response(
user_id: Optional[str], user_id: Optional[str],
user_info: Optional[Any], user_info: Optional[Any],
@ -739,6 +753,9 @@ def _build_user_info_response(
) )
if isinstance(_user_info, dict): if isinstance(_user_info, dict):
_user_info.pop("password", None) _user_info.pop("password", None)
_user_info["metadata"] = _redact_scim_enterprise_metadata(
_user_info.get("metadata")
)
return UserInfoResponse( return UserInfoResponse(
user_id=user_id, user_id=user_id,
@ -983,7 +1000,7 @@ async def user_info_v2(
models=user_data.get("models") or [], models=user_data.get("models") or [],
budget_duration=user_data.get("budget_duration"), budget_duration=user_data.get("budget_duration"),
budget_reset_at=user_data.get("budget_reset_at"), budget_reset_at=user_data.get("budget_reset_at"),
metadata=user_data.get("metadata"), metadata=_redact_scim_enterprise_metadata(user_data.get("metadata")),
created_at=user_data.get("created_at"), created_at=user_data.get("created_at"),
updated_at=user_data.get("updated_at"), updated_at=user_data.get("updated_at"),
sso_user_id=user_data.get("sso_user_id"), sso_user_id=user_data.get("sso_user_id"),
@ -2098,9 +2115,13 @@ async def get_users(
user_list: List[LiteLLM_UserTableWithKeyCount] = [] user_list: List[LiteLLM_UserTableWithKeyCount] = []
if users is not None: if users is not None:
for user in users: for user in users:
user_dump = user.model_dump()
user_dump["metadata"] = _redact_scim_enterprise_metadata(
user_dump.get("metadata")
)
user_list.append( user_list.append(
LiteLLM_UserTableWithKeyCount( LiteLLM_UserTableWithKeyCount(
**user.model_dump(), key_count=user_key_counts.get(user.user_id, 0) **user_dump, key_count=user_key_counts.get(user.user_id, 0)
) )
) )
else: else:

View file

@ -50,8 +50,16 @@ class ScimTransformations:
scim_active = metadata.get("scim_active") scim_active = metadata.get("scim_active")
active = True if scim_active is None else bool(scim_active) active = True if scim_active is None else bool(scim_active)
schemas = ["urn:ietf:params:scim:schemas:core:2.0:User"]
enterprise_user = None
if metadata.get(SCIM_ENTERPRISE_METADATA_KEY):
enterprise_user = SCIMEnterpriseUser.model_validate(
metadata[SCIM_ENTERPRISE_METADATA_KEY]
)
schemas.append(SCIM_ENTERPRISE_USER_SCHEMA)
return SCIMUser( return SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], schemas=schemas,
id=user.user_id, id=user.user_id,
userName=ScimTransformations._get_scim_user_name(user), userName=ScimTransformations._get_scim_user_name(user),
displayName=ScimTransformations._get_scim_user_name(user), displayName=ScimTransformations._get_scim_user_name(user),
@ -62,6 +70,7 @@ class ScimTransformations:
emails=emails, emails=emails,
groups=groups, groups=groups,
active=active, active=active,
enterprise_user=enterprise_user,
meta={ meta={
"resourceType": "User", "resourceType": "User",
"created": user_created_at, "created": user_created_at,

View file

@ -118,6 +118,7 @@ class ScimUserData(TypedDict):
given_name: Optional[str] given_name: Optional[str]
family_name: Optional[str] family_name: Optional[str]
active: Optional[bool] active: Optional[bool]
enterprise: Optional[SCIMEnterpriseUser]
class GroupMemberExtractionResult(BaseModel): class GroupMemberExtractionResult(BaseModel):
@ -199,11 +200,15 @@ def _extract_scim_user_data(user: SCIMUser) -> ScimUserData:
"given_name": user.name.givenName if user.name else None, "given_name": user.name.givenName if user.name else None,
"family_name": user.name.familyName if user.name else None, "family_name": user.name.familyName if user.name else None,
"active": user.active, "active": user.active,
"enterprise": user.enterprise_user,
} }
def _build_scim_metadata( def _build_scim_metadata(
given_name: Optional[str], family_name: Optional[str], active: Optional[bool] = None given_name: Optional[str],
family_name: Optional[str],
active: Optional[bool] = None,
enterprise: Optional[SCIMEnterpriseUser] = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Build metadata dictionary with SCIM data.""" """Build metadata dictionary with SCIM data."""
metadata: Dict[str, Any] = { metadata: Dict[str, Any] = {
@ -216,6 +221,11 @@ def _build_scim_metadata(
if active is not None: if active is not None:
metadata["scim_active"] = active metadata["scim_active"] = active
if enterprise is not None:
metadata[SCIM_ENTERPRISE_METADATA_KEY] = enterprise.model_dump(
by_alias=True, exclude_none=True
)
return metadata return metadata
@ -999,7 +1009,9 @@ async def create_user(
# Create user in database # Create user in database
user_id = user.userName or str(uuid.uuid4()) user_id = user.userName or str(uuid.uuid4())
metadata = _build_scim_metadata( metadata = _build_scim_metadata(
user_data["given_name"], user_data["family_name"] user_data["given_name"],
user_data["family_name"],
enterprise=user_data["enterprise"],
) )
default_role: Optional[ default_role: Optional[
@ -1088,6 +1100,7 @@ async def update_user(
user_data["given_name"], user_data["given_name"],
user_data["family_name"], user_data["family_name"],
scim_active_for_metadata, scim_active_for_metadata,
enterprise=user_data["enterprise"],
) )
await _handle_team_membership_changes( await _handle_team_membership_changes(

View file

@ -1,7 +1,20 @@
from typing import Any, Dict, List, Literal, Optional, Union from typing import Any, Dict, List, Literal, Optional, Union
from fastapi import HTTPException from fastapi import HTTPException
from pydantic import BaseModel, ConfigDict, EmailStr, field_validator from pydantic import (
BaseModel,
ConfigDict,
EmailStr,
Field,
field_validator,
model_serializer,
)
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"
class LiteLLM_UserScimMetadata(BaseModel): class LiteLLM_UserScimMetadata(BaseModel):
@ -42,13 +55,49 @@ class SCIMUserGroup(BaseModel):
type: Optional[str] = "direct" # direct or indirect type: Optional[str] = "direct" # direct or indirect
class SCIMUserManager(BaseModel):
model_config = ConfigDict(populate_by_name=True)
value: Optional[str] = None
displayName: Optional[str] = None
ref: Optional[str] = Field(default=None, alias="$ref")
class SCIMEnterpriseUser(BaseModel):
model_config = ConfigDict(populate_by_name=True)
employeeNumber: Optional[str] = None
costCenter: Optional[str] = None
organization: Optional[str] = None
division: Optional[str] = None
department: Optional[str] = None
manager: Optional[SCIMUserManager] = None
class SCIMUser(SCIMResource): class SCIMUser(SCIMResource):
model_config = ConfigDict(populate_by_name=True)
userName: Optional[str] = None userName: Optional[str] = None
name: Optional[SCIMUserName] = None name: Optional[SCIMUserName] = None
displayName: Optional[str] = None displayName: Optional[str] = None
active: bool = True active: bool = True
emails: Optional[List[SCIMUserEmail]] = None emails: Optional[List[SCIMUserEmail]] = None
groups: Optional[List[SCIMUserGroup]] = None groups: Optional[List[SCIMUserGroup]] = None
enterprise_user: Optional[SCIMEnterpriseUser] = Field(
default=None,
alias=SCIM_ENTERPRISE_USER_SCHEMA,
serialization_alias=SCIM_ENTERPRISE_USER_SCHEMA,
)
@model_serializer(mode="wrap")
def _omit_absent_enterprise(
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)
return dumped
class SCIMMember(BaseModel): class SCIMMember(BaseModel):

View file

@ -13,7 +13,10 @@ from litellm.proxy.management_endpoints.scim.scim_transformations import (
ScimTransformations, ScimTransformations,
) )
from litellm.types.proxy.management_endpoints.scim_v2 import ( from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_USER_SCHEMA,
SCIMEnterpriseUser,
SCIMPatchOperation, SCIMPatchOperation,
SCIMUser,
) )
@ -149,6 +152,77 @@ class TestScimTransformations:
assert scim_user.name.givenName == "Test" assert scim_user.name.givenName == "Test"
assert scim_user.name.familyName == "User" assert scim_user.name.familyName == "User"
@pytest.mark.asyncio
async def test_transform_user_with_enterprise_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-ent",
user_email="ent@example.com",
user_alias=None,
teams=[],
created_at=None,
updated_at=None,
metadata={
"scim_enterprise": {"costCenter": "CC-42", "department": "Platform"}
},
)
with patch("litellm.proxy.proxy_server.prisma_client", mock_client):
scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
user
)
assert scim_user.enterprise_user is not None
assert scim_user.enterprise_user.costCenter == "CC-42"
assert scim_user.enterprise_user.department == "Platform"
assert SCIM_ENTERPRISE_USER_SCHEMA in scim_user.schemas
@pytest.mark.asyncio
async def test_transform_user_without_enterprise_metadata_omits_schema(
self, mock_user, mock_prisma_client
):
mock_client, mock_find_unique = mock_prisma_client
team1 = LiteLLM_TeamTable(
team_id="team-1", team_alias="Team One", members_with_roles=[]
)
team2 = LiteLLM_TeamTable(
team_id="team-2", team_alias="Team Two", members_with_roles=[]
)
mock_find_unique.side_effect = [team1, team2]
with patch("litellm.proxy.proxy_server.prisma_client", mock_client):
scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
mock_user
)
assert scim_user.enterprise_user is None
assert SCIM_ENTERPRISE_USER_SCHEMA not in scim_user.schemas
def test_scim_user_serialization_omits_absent_enterprise_urn(self):
without_enterprise = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id="user-1",
userName="user@example.com",
)
dumped = without_enterprise.model_dump(by_alias=True)
assert SCIM_ENTERPRISE_USER_SCHEMA not in dumped
assert "enterprise_user" not in dumped
assert SCIM_ENTERPRISE_USER_SCHEMA not in dumped["schemas"]
with_enterprise = SCIMUser(
schemas=[
"urn:ietf:params:scim:schemas:core:2.0:User",
SCIM_ENTERPRISE_USER_SCHEMA,
],
id="user-2",
userName="ent@example.com",
enterprise_user=SCIMEnterpriseUser(costCenter="CC-42"),
)
dumped_ent = with_enterprise.model_dump(by_alias=True)
assert dumped_ent[SCIM_ENTERPRISE_USER_SCHEMA]["costCenter"] == "CC-42"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_transform_litellm_team_to_scim_group( async def test_transform_litellm_team_to_scim_group(
self, mock_team, mock_prisma_client self, mock_team, mock_prisma_client

View file

@ -24,6 +24,7 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
update_user, update_user,
) )
from litellm.types.proxy.management_endpoints.scim_v2 import ( from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIM_ENTERPRISE_USER_SCHEMA,
SCIMGroup, SCIMGroup,
SCIMMember, SCIMMember,
SCIMPatchOp, SCIMPatchOp,
@ -115,6 +116,59 @@ async def test_create_user_defaults_to_viewer(mocker, monkeypatch):
assert called_args.user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY assert called_args.user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY
@pytest.mark.asyncio
async def test_create_user_ingests_enterprise_extension(mocker, monkeypatch):
"""A SCIM create payload carrying the enterprise extension block should land
in the created user's metadata under scim_enterprise"""
scim_user = SCIMUser.model_validate(
{
"schemas": [
"urn:ietf:params:scim:schemas:core:2.0:User",
SCIM_ENTERPRISE_USER_SCHEMA,
],
"userName": "ent-user",
"name": {"familyName": "User", "givenName": "Ent"},
"emails": [{"value": "ent@example.com"}],
SCIM_ENTERPRISE_USER_SCHEMA: {
"costCenter": "CC-42",
"department": "Platform",
},
}
)
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="ent-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_enterprise"] == {
"costCenter": "CC-42",
"department": "Platform",
}
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_create_user_uses_default_internal_user_params_role(mocker, monkeypatch): 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""" """If role is set in default_internal_user_params, new user should use that role"""

View file

@ -657,6 +657,56 @@ async def test_get_users_includes_timestamps(mocker):
assert user_response.key_count == 0 assert user_response.key_count == 0
@pytest.mark.asyncio
async def test_get_users_redacts_scim_enterprise_metadata(mocker):
"""
/user/list must strip scim_enterprise from each user's metadata while leaving
the rest of the metadata intact, matching the user-info endpoints.
"""
mock_prisma_client = mocker.MagicMock()
mock_user_row = mocker.MagicMock()
mock_user_row.user_id = "listed-user"
mock_user_row.model_dump.return_value = {
"user_id": "listed-user",
"user_email": "listed@example.com",
"user_role": "internal_user",
"metadata": {
"scim_metadata": {"givenName": "Jane", "familyName": "Doe"},
"scim_enterprise": {"costCenter": "CC-42", "department": "Platform"},
},
}
async def mock_find_many(*args, **kwargs):
return [mock_user_row]
async def mock_count(*args, **kwargs):
return 1
mock_prisma_client.db.litellm_usertable.find_many = mock_find_many
mock_prisma_client.db.litellm_usertable.count = mock_count
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
async def mock_get_user_key_counts(*args, **kwargs):
return {"listed-user": 0}
mocker.patch(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_user_key_counts",
mock_get_user_key_counts,
)
admin_key = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
response = await get_users(
page=1, page_size=1, user_api_key_dict=admin_key, organization_ids=None
)
listed = response["users"][0]
assert listed.metadata == {
"scim_metadata": {"givenName": "Jane", "familyName": "Doe"}
}
assert "scim_enterprise" not in (listed.metadata or {})
def test_validate_sort_params(): def test_validate_sort_params():
""" """
Test that validate_sort_params returns None if sort_by is None Test that validate_sort_params returns None if sort_by is None
@ -2167,6 +2217,94 @@ async def test_user_info_v2_proxy_admin_can_query_any_user(mocker):
assert response.metadata == {"team": "engineering"} assert response.metadata == {"team": "engineering"}
@pytest.mark.asyncio
async def test_user_info_v2_redacts_scim_enterprise_metadata(mocker):
"""
SCIM enterprise attributes are persisted in metadata for reporting, but
/v2/user/info must not surface them; the rest of metadata is preserved.
"""
from fastapi import Request
from litellm.proxy._types import UserInfoV2Response
from litellm.proxy.management_endpoints.internal_user_endpoints import user_info_v2
mock_prisma_client = mocker.MagicMock()
mock_user_row = mocker.MagicMock()
mock_user_row.model_dump.return_value = {
"user_id": "target-user-123",
"user_email": "target@example.com",
"metadata": {
"scim_metadata": {"givenName": "Jane", "familyName": "Doe"},
"scim_enterprise": {
"costCenter": "CC-42",
"department": "Platform",
"employeeNumber": "E-1001",
},
},
"teams": ["team-1"],
}
async def mock_find_unique(*args, **kwargs):
if kwargs.get("where", {}).get("user_id") == "target-user-123":
return mock_user_row
return None
mock_prisma_client.db.litellm_usertable.find_unique = mocker.AsyncMock(
side_effect=mock_find_unique
)
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mock_request = mocker.MagicMock(spec=Request)
admin_key = UserAPIKeyAuth(
user_id="admin-user", user_role=LitellmUserRoles.PROXY_ADMIN
)
response = await user_info_v2(
request=mock_request,
user_id="target-user-123",
user_api_key_dict=admin_key,
)
assert isinstance(response, UserInfoV2Response)
assert response.metadata == {
"scim_metadata": {"givenName": "Jane", "familyName": "Doe"}
}
assert "scim_enterprise" not in (response.metadata or {})
def test_build_user_info_response_redacts_scim_enterprise_metadata():
"""
The shared /user/info builder strips scim_enterprise from the returned user row
while leaving every other metadata key intact.
"""
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_build_user_info_response,
)
user_row = {
"user_id": "target-user-123",
"metadata": {
"scim_metadata": {"givenName": "Jane"},
"scim_enterprise": {"costCenter": "CC-42"},
},
}
response = _build_user_info_response(
user_id="target-user-123",
user_info=user_row,
keys=None,
team_list=[],
teams_1=None,
)
assert response.user_info is not None
assert response.user_info["metadata"] == {"scim_metadata": {"givenName": "Jane"}}
assert "scim_enterprise" not in response.user_info["metadata"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_user_info_v2_internal_user_can_query_self(mocker): async def test_user_info_v2_internal_user_can_query_self(mocker):
""" """