diff --git a/litellm/proxy/management_endpoints/scim/README_SCIM.md b/litellm/proxy/management_endpoints/scim/README_SCIM.md index 8eb366789b..eb628a7806 100644 --- a/litellm/proxy/management_endpoints/scim/README_SCIM.md +++ b/litellm/proxy/management_endpoints/scim/README_SCIM.md @@ -68,13 +68,6 @@ Most identity providers will require authentication. You should use a valid Lite - Automatic synchronization of user-team relationships - Proper status codes and error handling per SCIM specification -## Limitations - -This is a basic implementation intended for testing and integration purposes. Some limitations include: - -- Limited filtering capabilities (only supports basic equality filters) -- No support for complex SCIM operations like PATCH -- Limited schema customization ## Example Usage diff --git a/litellm/proxy/management_endpoints/scim/scim_transformations.py b/litellm/proxy/management_endpoints/scim/scim_transformations.py new file mode 100644 index 0000000000..f68f728e2a --- /dev/null +++ b/litellm/proxy/management_endpoints/scim/scim_transformations.py @@ -0,0 +1,154 @@ +from typing import List, Union + +from litellm.proxy._types import ( + LiteLLM_TeamTable, + LiteLLM_UserTable, + Member, + NewUserResponse, +) +from litellm.types.proxy.management_endpoints.scim_v2 import * + + +class ScimTransformations: + DEFAULT_SCIM_NAME = "Unknown User" + DEFAULT_SCIM_FAMILY_NAME = "Unknown Family Name" + DEFAULT_SCIM_DISPLAY_NAME = "Unknown Display Name" + DEFAULT_SCIM_MEMBER_VALUE = "Unknown Member Value" + + @staticmethod + async def transform_litellm_user_to_scim_user( + user: Union[LiteLLM_UserTable, NewUserResponse], + ) -> SCIMUser: + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail={"error": "No database connected"} + ) + + # Get user's teams/groups + groups = [] + for team_id in user.teams or []: + team = await prisma_client.db.litellm_teamtable.find_unique( + where={"team_id": team_id} + ) + if team: + team_alias = getattr(team, "team_alias", team.team_id) + groups.append(SCIMUserGroup(value=team.team_id, display=team_alias)) + + user_created_at = user.created_at.isoformat() if user.created_at else None + user_updated_at = user.updated_at.isoformat() if user.updated_at else None + + emails = [] + if user.user_email: + emails.append(SCIMUserEmail(value=user.user_email, primary=True)) + + return SCIMUser( + schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], + id=user.user_id, + userName=ScimTransformations._get_scim_user_name(user), + displayName=ScimTransformations._get_scim_user_name(user), + name=SCIMUserName( + familyName=ScimTransformations._get_scim_family_name(user), + givenName=ScimTransformations._get_scim_given_name(user), + ), + emails=emails, + groups=groups, + active=True, + meta={ + "resourceType": "User", + "created": user_created_at, + "lastModified": user_updated_at, + }, + ) + + @staticmethod + def _get_scim_user_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: + """ + SCIM requires a display name with length > 0 + + We use the same userName and displayName for SCIM users + """ + if user.user_email and len(user.user_email) > 0: + return user.user_email + return ScimTransformations.DEFAULT_SCIM_DISPLAY_NAME + + @staticmethod + def _get_scim_family_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: + """ + SCIM requires a family name with length > 0 + """ + metadata = user.metadata or {} + if "scim_metadata" in metadata: + scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata( + **metadata["scim_metadata"] + ) + if scim_metadata.familyName and len(scim_metadata.familyName) > 0: + return scim_metadata.familyName + + if user.user_alias and len(user.user_alias) > 0: + return user.user_alias + return ScimTransformations.DEFAULT_SCIM_FAMILY_NAME + + @staticmethod + def _get_scim_given_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: + """ + SCIM requires a given name with length > 0 + """ + metadata = user.metadata or {} + if "scim_metadata" in metadata: + scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata( + **metadata["scim_metadata"] + ) + if scim_metadata.givenName and len(scim_metadata.givenName) > 0: + return scim_metadata.givenName + + if user.user_alias and len(user.user_alias) > 0: + return user.user_alias or ScimTransformations.DEFAULT_SCIM_NAME + return ScimTransformations.DEFAULT_SCIM_NAME + + @staticmethod + async def transform_litellm_team_to_scim_group( + team: Union[LiteLLM_TeamTable, dict], + ) -> SCIMGroup: + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail={"error": "No database connected"} + ) + + if isinstance(team, dict): + team = LiteLLM_TeamTable(**team) + + # Get team members + scim_members: List[SCIMMember] = [] + for member in team.members_with_roles or []: + scim_members.append( + SCIMMember( + value=ScimTransformations._get_scim_member_value(member), + display=member.user_email, + ) + ) + + team_alias = getattr(team, "team_alias", team.team_id) + team_created_at = team.created_at.isoformat() if team.created_at else None + team_updated_at = team.updated_at.isoformat() if team.updated_at else None + + return SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=team.team_id, + displayName=team_alias, + members=scim_members, + meta={ + "resourceType": "Group", + "created": team_created_at, + "lastModified": team_updated_at, + }, + ) + + @staticmethod + def _get_scim_member_value(member: Member) -> str: + if member.user_email: + return member.user_email + return ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE diff --git a/litellm/proxy/management_endpoints/scim/scim_v2.py b/litellm/proxy/management_endpoints/scim/scim_v2.py index 1b3cdd6fcd..bc0ba3eb46 100644 --- a/litellm/proxy/management_endpoints/scim/scim_v2.py +++ b/litellm/proxy/management_endpoints/scim/scim_v2.py @@ -1,12 +1,10 @@ """ SCIM v2 Endpoints for LiteLLM Proxy using Internal User/Team Management -Provides basic implementations of SCIM v2 endpoints for testing -and integration purposes. """ import uuid -from typing import List, Optional, Union +from typing import List, Optional from fastapi import ( APIRouter, @@ -21,17 +19,18 @@ from fastapi import ( from litellm._logging import verbose_proxy_logger from litellm.proxy._types import ( - LiteLLM_TeamTable, LiteLLM_UserTable, LitellmUserRoles, Member, NewTeamRequest, NewUserRequest, - NewUserResponse, UserAPIKeyAuth, ) from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.management_endpoints.internal_user_endpoints import new_user +from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, +) from litellm.proxy.management_endpoints.team_endpoints import new_team from litellm.types.proxy.management_endpoints.scim_v2 import * @@ -49,151 +48,6 @@ async def set_scim_content_type(response: Response): response.headers["Content-Type"] = "application/scim+json" -class ScimTransformations: - DEFAULT_SCIM_NAME = "Unknown User" - DEFAULT_SCIM_FAMILY_NAME = "Unknown Family Name" - DEFAULT_SCIM_DISPLAY_NAME = "Unknown Display Name" - DEFAULT_SCIM_MEMBER_VALUE = "Unknown Member Value" - - @staticmethod - async def transform_litellm_user_to_scim_user( - user: Union[LiteLLM_UserTable, NewUserResponse], - ) -> SCIMUser: - from litellm.proxy.proxy_server import prisma_client - - if prisma_client is None: - raise HTTPException( - status_code=500, detail={"error": "No database connected"} - ) - - # Get user's teams/groups - groups = [] - for team_id in user.teams or []: - team = await prisma_client.db.litellm_teamtable.find_unique( - where={"team_id": team_id} - ) - if team: - team_alias = getattr(team, "team_alias", team.team_id) - groups.append(SCIMUserGroup(value=team.team_id, display=team_alias)) - - user_created_at = user.created_at.isoformat() if user.created_at else None - user_updated_at = user.updated_at.isoformat() if user.updated_at else None - - emails = [] - if user.user_email: - emails.append(SCIMUserEmail(value=user.user_email, primary=True)) - - return SCIMUser( - schemas=["urn:ietf:params:scim:schemas:core:2.0:User"], - id=user.user_id, - userName=ScimTransformations._get_scim_user_name(user), - displayName=ScimTransformations._get_scim_user_name(user), - name=SCIMUserName( - familyName=ScimTransformations._get_scim_family_name(user), - givenName=ScimTransformations._get_scim_given_name(user), - ), - emails=emails, - groups=groups, - active=True, - meta={ - "resourceType": "User", - "created": user_created_at, - "lastModified": user_updated_at, - }, - ) - - @staticmethod - def _get_scim_user_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: - """ - SCIM requires a display name with length > 0 - - We use the same userName and displayName for SCIM users - """ - if user.user_email and len(user.user_email) > 0: - return user.user_email - return ScimTransformations.DEFAULT_SCIM_DISPLAY_NAME - - @staticmethod - def _get_scim_family_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: - """ - SCIM requires a family name with length > 0 - """ - metadata = user.metadata or {} - if "scim_metadata" in metadata: - scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata( - **metadata["scim_metadata"] - ) - if scim_metadata.familyName and len(scim_metadata.familyName) > 0: - return scim_metadata.familyName - - if user.user_alias and len(user.user_alias) > 0: - return user.user_alias - return ScimTransformations.DEFAULT_SCIM_FAMILY_NAME - - @staticmethod - def _get_scim_given_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str: - """ - SCIM requires a given name with length > 0 - """ - metadata = user.metadata or {} - if "scim_metadata" in metadata: - scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata( - **metadata["scim_metadata"] - ) - if scim_metadata.givenName and len(scim_metadata.givenName) > 0: - return scim_metadata.givenName - - if user.user_alias and len(user.user_alias) > 0: - return user.user_alias or ScimTransformations.DEFAULT_SCIM_NAME - return ScimTransformations.DEFAULT_SCIM_NAME - - @staticmethod - async def transform_litellm_team_to_scim_group( - team: Union[LiteLLM_TeamTable, dict], - ) -> SCIMGroup: - from litellm.proxy.proxy_server import prisma_client - - if prisma_client is None: - raise HTTPException( - status_code=500, detail={"error": "No database connected"} - ) - - if isinstance(team, dict): - team = LiteLLM_TeamTable(**team) - - # Get team members - scim_members: List[SCIMMember] = [] - for member in team.members_with_roles or []: - scim_members.append( - SCIMMember( - value=ScimTransformations._get_scim_member_value(member), - display=member.user_email, - ) - ) - - team_alias = getattr(team, "team_alias", team.team_id) - team_created_at = team.created_at.isoformat() if team.created_at else None - team_updated_at = team.updated_at.isoformat() if team.updated_at else None - - return SCIMGroup( - schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], - id=team.team_id, - displayName=team_alias, - members=scim_members, - meta={ - "resourceType": "Group", - "created": team_created_at, - "lastModified": team_updated_at, - }, - ) - - @staticmethod - def _get_scim_member_value(member: Member) -> str: - if member.user_email: - return member.user_email - return ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE - - # User Endpoints @scim_router.get( "/Users",