Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/scim/scim_transformations.py: 21%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1from collections.abc import Callable
2from typing import Final, TypeVar
4from pydantic import ValidationError
6from litellm._logging import verbose_proxy_logger
7from litellm.proxy._types import (
8 LiteLLM_TeamTable,
9 LiteLLM_UserTable,
10 Member,
11 NewUserResponse,
12)
13from litellm.repositories.team_repository import TeamRepository
14from litellm.types.proxy.management_endpoints.scim_v2 import *
16T = TypeVar("T")
19class ScimTransformations:
20 DEFAULT_SCIM_NAME = "Unknown User"
21 DEFAULT_SCIM_FAMILY_NAME = "Unknown Family Name"
22 DEFAULT_SCIM_DISPLAY_NAME = "Unknown Display Name"
23 DEFAULT_SCIM_MEMBER_VALUE = "Unknown Member Value"
25 @staticmethod
26 async def transform_litellm_user_to_scim_user(
27 user: LiteLLM_UserTable | NewUserResponse,
28 ) -> SCIMUser:
29 from litellm.proxy.proxy_server import prisma_client
31 if prisma_client is None:
32 raise HTTPException(status_code=500, detail={"error": "No database connected"})
34 # Get user's teams/groups
35 groups: Final = []
36 team_ids: Final[list[str]] = user.teams or [] # mutable-ok: scim reads the user row's team ids
37 for team_id in team_ids:
38 team = await TeamRepository(prisma_client).table.find_unique(where={"team_id": team_id})
39 if team:
40 team_alias = getattr(team, "team_alias", team.team_id)
41 groups.append(SCIMUserGroup(value=team.team_id, display=team_alias))
43 user_created_at: Final = user.created_at.isoformat() if user.created_at else None
44 user_updated_at: Final = user.updated_at.isoformat() if user.updated_at else None
46 emails: Final = []
47 # Only add email if it's a valid email address (contains @)
48 # user_email can be a UUID when users are created without an email
49 if user.user_email and "@" in user.user_email:
50 emails.append(SCIMUserEmail(value=user.user_email, primary=True))
52 metadata: Final = user.metadata or {}
53 scim_active: Final = metadata.get("scim_active")
54 active: Final = True if scim_active is None else bool(scim_active)
56 schemas: Final = ["urn:ietf:params:scim:schemas:core:2.0:User"]
57 enterprise_user: Final = ScimTransformations._parse_directory_metadata(
58 user, SCIM_ENTERPRISE_METADATA_KEY, SCIMEnterpriseUser.model_validate
59 )
60 if enterprise_user is not None:
61 schemas.append(SCIM_ENTERPRISE_USER_SCHEMA)
63 entitlements: Final = ScimTransformations._parse_directory_metadata(
64 user, SCIM_ENTITLEMENTS_METADATA_KEY, SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python
65 )
66 roles: Final = ScimTransformations._parse_directory_metadata(
67 user, SCIM_ROLES_METADATA_KEY, SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python
68 )
70 return SCIMUser(
71 schemas=schemas,
72 id=user.user_id,
73 userName=ScimTransformations._get_scim_user_name(user),
74 displayName=ScimTransformations._get_scim_user_name(user),
75 name=SCIMUserName(
76 familyName=ScimTransformations._get_scim_family_name(user),
77 givenName=ScimTransformations._get_scim_given_name(user),
78 ),
79 emails=emails,
80 groups=groups,
81 active=active,
82 entitlements=entitlements,
83 roles=roles,
84 enterprise_user=enterprise_user,
85 meta={
86 "resourceType": "User",
87 "created": user_created_at,
88 "lastModified": user_updated_at,
89 },
90 )
92 @staticmethod
93 def _parse_directory_metadata(
94 user: LiteLLM_UserTable | NewUserResponse,
95 key: str,
96 validate: Callable[[object], T],
97 ) -> T | None:
98 """A SCIM directory attribute parsed from user metadata, or None when absent or malformed.
100 Metadata is writable outside the SCIM surface, so a malformed value on one user must not
101 fail the whole directory response; the attribute is omitted and the corruption logged.
102 """
103 metadata: Final = user.metadata or {}
104 raw: Final = metadata.get(key)
105 if not raw:
106 return None
107 try:
108 return validate(raw)
109 except ValidationError:
110 verbose_proxy_logger.warning(
111 "Skipping malformed %s metadata on user %s in SCIM response",
112 key,
113 user.user_id,
114 )
115 return None
117 @staticmethod
118 def _get_scim_user_name(user: LiteLLM_UserTable | NewUserResponse) -> str:
119 """
120 SCIM requires a display name with length > 0
122 We use the same userName and displayName for SCIM users
123 """
124 if user.user_email and len(user.user_email) > 0:
125 return user.user_email
126 return ScimTransformations.DEFAULT_SCIM_DISPLAY_NAME
128 @staticmethod
129 def _get_scim_family_name(user: LiteLLM_UserTable | NewUserResponse) -> str:
130 """
131 SCIM requires a family name with length > 0
132 """
133 metadata: Final = user.metadata or {}
134 if "scim_metadata" in metadata:
135 scim_metadata: Final[LiteLLM_UserScimMetadata] = LiteLLM_UserScimMetadata(**metadata["scim_metadata"])
136 if scim_metadata.familyName and len(scim_metadata.familyName) > 0:
137 return scim_metadata.familyName
139 if user.user_alias and len(user.user_alias) > 0:
140 return user.user_alias
141 return ScimTransformations.DEFAULT_SCIM_FAMILY_NAME
143 @staticmethod
144 def _get_scim_given_name(user: LiteLLM_UserTable | NewUserResponse) -> str:
145 """
146 SCIM requires a given name with length > 0
147 """
148 metadata: Final = user.metadata or {}
149 if "scim_metadata" in metadata:
150 scim_metadata: Final[LiteLLM_UserScimMetadata] = LiteLLM_UserScimMetadata(**metadata["scim_metadata"])
151 if scim_metadata.givenName and len(scim_metadata.givenName) > 0:
152 return scim_metadata.givenName
154 if user.user_alias and len(user.user_alias) > 0:
155 return user.user_alias or ScimTransformations.DEFAULT_SCIM_NAME
156 return ScimTransformations.DEFAULT_SCIM_NAME
158 @staticmethod
159 async def transform_litellm_team_to_scim_group(
160 team: LiteLLM_TeamTable | dict,
161 ) -> SCIMGroup:
162 from litellm.proxy.proxy_server import prisma_client
164 if prisma_client is None:
165 raise HTTPException(status_code=500, detail={"error": "No database connected"})
167 if isinstance(team, dict):
168 team = LiteLLM_TeamTable(**team)
170 # Get team members with proper display names
171 scim_members: Final[list[SCIMMember]] = []
172 for member in team.members_with_roles or []:
173 if isinstance(member, dict):
174 member = Member(**member)
176 scim_members.append(
177 SCIMMember(
178 value=ScimTransformations._get_scim_member_value(member),
179 display=ScimTransformations._get_scim_member_display(member),
180 type="User",
181 )
182 )
184 team_alias: Final = getattr(team, "team_alias", team.team_id)
185 team_created_at: Final = team.created_at.isoformat() if team.created_at else None
186 team_updated_at: Final = team.updated_at.isoformat() if team.updated_at else None
188 return SCIMGroup(
189 schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
190 id=team.team_id,
191 displayName=team_alias,
192 members=scim_members,
193 meta={
194 "resourceType": "Group",
195 "created": team_created_at,
196 "lastModified": team_updated_at,
197 },
198 )
200 @staticmethod
201 def _get_scim_member_value(member: Member) -> str:
202 """The member's SCIM resource id, which LiteLLM serves as user_id (RFC 7643 §8.7.1)."""
203 return member.user_id or ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE
205 @staticmethod
206 def _get_scim_member_display(member: Member) -> str:
207 """
208 Get the SCIM member display. Use user_email if available, otherwise use user_id.
209 SCIM member display should be the display name for the user.
210 """
211 if hasattr(member, "user_email") and member.user_email:
212 return member.user_email
213 elif hasattr(member, "user_id"):
214 return member.user_id or ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE
215 return ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE