refac

backend/open_webui/routers/scim.py

@@ -1,6 +1,8 @@
 """
-SCIM 2.0 Implementation for Open WebUI
+Experimental SCIM 2.0 Implementation for Open WebUI
 Provides System for Cross-domain Identity Management endpoints for users and groups
+
+NOTE: This is an experimental implementation and may not fully comply with SCIM 2.0 standards, and is subject to change.
 """
 
 import logging
@@ -15,7 +17,12 @@ from pydantic import BaseModel, Field, ConfigDict
 
 from open_webui.models.users import Users, UserModel
 from open_webui.models.groups import Groups, GroupModel
-from open_webui.utils.auth import get_admin_user, get_current_user, decode_token, get_verified_user
+from open_webui.utils.auth import (
+    get_admin_user,
+    get_current_user,
+    decode_token,
+    get_verified_user,
+)
 from open_webui.constants import ERROR_MESSAGES
 from open_webui.env import SRC_LOG_LEVELS
 
@@ -40,7 +47,7 @@ def scim_error(status_code: int, detail: str, scim_type: Optional[str] = None):
     error_body = {
         "schemas": [SCIM_ERROR_SCHEMA],
         "status": str(status_code),
-        "detail": detail
+        "detail": detail,
     }
 
     if scim_type:
@@ -52,14 +59,12 @@ def scim_error(status_code: int, detail: str, scim_type: Optional[str] = None):
     elif status_code == 400:
         error_body["scimType"] = "invalidSyntax"
 
-    return JSONResponse(
+    return JSONResponse(status_code=status_code, content=error_body)
-        status_code=status_code,
-        content=error_body
-    )
 
 
 class SCIMError(BaseModel):
     """SCIM Error Response"""
+
     schemas: List[str] = [SCIM_ERROR_SCHEMA]
     status: str
     scimType: Optional[str] = None
@@ -68,6 +73,7 @@ class SCIMError(BaseModel):
 
 class SCIMMeta(BaseModel):
     """SCIM Resource Metadata"""
+
     resourceType: str
     created: str
     lastModified: str
@@ -77,6 +83,7 @@ class SCIMMeta(BaseModel):
 
 class SCIMName(BaseModel):
     """SCIM User Name"""
+
     formatted: Optional[str] = None
     familyName: Optional[str] = None
     givenName: Optional[str] = None
@@ -87,6 +94,7 @@ class SCIMName(BaseModel):
 
 class SCIMEmail(BaseModel):
     """SCIM Email"""
+
     value: str
     type: Optional[str] = "work"
     primary: bool = True
@@ -95,6 +103,7 @@ class SCIMEmail(BaseModel):
 
 class SCIMPhoto(BaseModel):
     """SCIM Photo"""
+
     value: str
     type: Optional[str] = "photo"
     primary: bool = True
@@ -103,6 +112,7 @@ class SCIMPhoto(BaseModel):
 
 class SCIMGroupMember(BaseModel):
     """SCIM Group Member"""
+
     value: str  # User ID
     ref: Optional[str] = Field(None, alias="$ref")
     type: Optional[str] = "User"
@@ -111,6 +121,7 @@ class SCIMGroupMember(BaseModel):
 
 class SCIMUser(BaseModel):
     """SCIM User Resource"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_USER_SCHEMA]
@@ -128,6 +139,7 @@ class SCIMUser(BaseModel):
 
 class SCIMUserCreateRequest(BaseModel):
     """SCIM User Create Request"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_USER_SCHEMA]
@@ -143,6 +155,7 @@ class SCIMUserCreateRequest(BaseModel):
 
 class SCIMUserUpdateRequest(BaseModel):
     """SCIM User Update Request"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_USER_SCHEMA]
@@ -158,6 +171,7 @@ class SCIMUserUpdateRequest(BaseModel):
 
 class SCIMGroup(BaseModel):
     """SCIM Group Resource"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_GROUP_SCHEMA]
@@ -169,6 +183,7 @@ class SCIMGroup(BaseModel):
 
 class SCIMGroupCreateRequest(BaseModel):
     """SCIM Group Create Request"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_GROUP_SCHEMA]
@@ -178,6 +193,7 @@ class SCIMGroupCreateRequest(BaseModel):
 
 class SCIMGroupUpdateRequest(BaseModel):
     """SCIM Group Update Request"""
+
     model_config = ConfigDict(populate_by_name=True)
 
     schemas: List[str] = [SCIM_GROUP_SCHEMA]
@@ -187,6 +203,7 @@ class SCIMGroupUpdateRequest(BaseModel):
 
 class SCIMListResponse(BaseModel):
     """SCIM List Response"""
+
     schemas: List[str] = [SCIM_LIST_RESPONSE_SCHEMA]
     totalResults: int
     itemsPerPage: int
@@ -196,6 +213,7 @@ class SCIMListResponse(BaseModel):
 
 class SCIMPatchOperation(BaseModel):
     """SCIM Patch Operation"""
+
     op: str  # "add", "replace", "remove"
     path: Optional[str] = None
     value: Optional[Any] = None
@@ -203,11 +221,14 @@ class SCIMPatchOperation(BaseModel):
 
 class SCIMPatchRequest(BaseModel):
     """SCIM Patch Request"""
+
     schemas: List[str] = ["urn:ietf:params:scim:api:messages:2.0:PatchOp"]
     Operations: List[SCIMPatchOperation]
 
 
-def get_scim_auth(request: Request, authorization: Optional[str] = Header(None)) -> bool:
+def get_scim_auth(
+    request: Request, authorization: Optional[str] = Header(None)
+) -> bool:
     """
     Verify SCIM authentication
     Checks for SCIM-specific bearer token configured in the system
@@ -236,9 +257,11 @@ def get_scim_auth(request: Request, authorization: Optional[str] = Header(None))
 
         # Check if SCIM is enabled
         scim_enabled = getattr(request.app.state.config, "SCIM_ENABLED", False)
-        log.info(f"SCIM auth check - raw SCIM_ENABLED: {scim_enabled}, type: {type(scim_enabled)}")
+        log.info(
+            f"SCIM auth check - raw SCIM_ENABLED: {scim_enabled}, type: {type(scim_enabled)}"
+        )
         # Handle both PersistentConfig and direct value
-        if hasattr(scim_enabled, 'value'):
+        if hasattr(scim_enabled, "value"):
             scim_enabled = scim_enabled.value
         log.info(f"SCIM enabled status after conversion: {scim_enabled}")
         if not scim_enabled:
@@ -250,7 +273,7 @@ def get_scim_auth(request: Request, authorization: Optional[str] = Header(None))
         # Verify the SCIM token
         scim_token = getattr(request.app.state.config, "SCIM_TOKEN", None)
         # Handle both PersistentConfig and direct value
-        if hasattr(scim_token, 'value'):
+        if hasattr(scim_token, "value"):
             scim_token = scim_token.value
         log.debug(f"SCIM token configured: {bool(scim_token)}")
         if not scim_token or token != scim_token:
@@ -266,6 +289,7 @@ def get_scim_auth(request: Request, authorization: Optional[str] = Header(None))
     except Exception as e:
         log.error(f"SCIM authentication error: {e}")
         import traceback
+
         log.error(f"Traceback: {traceback.format_exc()}")
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
@@ -273,7 +297,6 @@ def get_scim_auth(request: Request, authorization: Optional[str] = Header(None))
         )
 
 
-
 def user_to_scim(user: UserModel, request: Request) -> SCIMUser:
     """Convert internal User model to SCIM User"""
     # Parse display name into name components
@@ -288,7 +311,7 @@ def user_to_scim(user: UserModel, request: Request) -> SCIMUser:
             "value": group.id,
             "display": group.name,
             "$ref": f"{request.base_url}api/v1/scim/v2/Groups/{group.id}",
-            "type": "direct"
+            "type": "direct",
         }
         for group in user_groups
     ]
@@ -304,12 +327,20 @@ def user_to_scim(user: UserModel, request: Request) -> SCIMUser:
         displayName=user.name,
         emails=[SCIMEmail(value=user.email)],
         active=user.role != "pending",
-        photos=[SCIMPhoto(value=user.profile_image_url)] if user.profile_image_url else None,
+        photos=(
+            [SCIMPhoto(value=user.profile_image_url)]
+            if user.profile_image_url
+            else None
+        ),
         groups=groups if groups else None,
         meta=SCIMMeta(
             resourceType=SCIM_RESOURCE_TYPE_USER,
-            created=datetime.fromtimestamp(user.created_at, tz=timezone.utc).isoformat(),
+            created=datetime.fromtimestamp(
-            lastModified=datetime.fromtimestamp(user.updated_at, tz=timezone.utc).isoformat(),
+                user.created_at, tz=timezone.utc
+            ).isoformat(),
+            lastModified=datetime.fromtimestamp(
+                user.updated_at, tz=timezone.utc
+            ).isoformat(),
             location=f"{request.base_url}api/v1/scim/v2/Users/{user.id}",
         ),
     )
@@ -335,48 +366,36 @@ def group_to_scim(group: GroupModel, request: Request) -> SCIMGroup:
         members=members,
         meta=SCIMMeta(
             resourceType=SCIM_RESOURCE_TYPE_GROUP,
-            created=datetime.fromtimestamp(group.created_at, tz=timezone.utc).isoformat(),
+            created=datetime.fromtimestamp(
-            lastModified=datetime.fromtimestamp(group.updated_at, tz=timezone.utc).isoformat(),
+                group.created_at, tz=timezone.utc
+            ).isoformat(),
+            lastModified=datetime.fromtimestamp(
+                group.updated_at, tz=timezone.utc
+            ).isoformat(),
             location=f"{request.base_url}api/v1/scim/v2/Groups/{group.id}",
         ),
     )
 
 
-
 # SCIM Service Provider Config
 @router.get("/ServiceProviderConfig")
 async def get_service_provider_config():
     """Get SCIM Service Provider Configuration"""
     return {
         "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"],
-        "patch": {
+        "patch": {"supported": True},
-            "supported": True
+        "bulk": {"supported": False, "maxOperations": 1000, "maxPayloadSize": 1048576},
-        },
+        "filter": {"supported": True, "maxResults": 200},
-        "bulk": {
+        "changePassword": {"supported": False},
-            "supported": False,
+        "sort": {"supported": False},
-            "maxOperations": 1000,
+        "etag": {"supported": False},
-            "maxPayloadSize": 1048576
-        },
-        "filter": {
-            "supported": True,
-            "maxResults": 200
-        },
-        "changePassword": {
-            "supported": False
-        },
-        "sort": {
-            "supported": False
-        },
-        "etag": {
-            "supported": False
-        },
         "authenticationSchemes": [
             {
                 "type": "oauthbearertoken",
                 "name": "OAuth Bearer Token",
-                "description": "Authentication using OAuth 2.0 Bearer Token"
+                "description": "Authentication using OAuth 2.0 Bearer Token",
             }
-        ]
+        ],
     }
 
 
@@ -393,8 +412,8 @@ async def get_resource_types(request: Request):
             "schema": SCIM_USER_SCHEMA,
             "meta": {
                 "location": f"{request.base_url}api/v1/scim/v2/ResourceTypes/User",
-                "resourceType": "ResourceType"
+                "resourceType": "ResourceType",
-            }
+            },
         },
         {
             "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ResourceType"],
@@ -404,9 +423,9 @@ async def get_resource_types(request: Request):
             "schema": SCIM_GROUP_SCHEMA,
             "meta": {
                 "location": f"{request.base_url}api/v1/scim/v2/ResourceTypes/Group",
-                "resourceType": "ResourceType"
+                "resourceType": "ResourceType",
-            }
+            },
-        }
+        },
     ]
 
 
@@ -425,25 +444,17 @@ async def get_schemas():
                     "name": "userName",
                     "type": "string",
                     "required": True,
-                    "uniqueness": "server"
+                    "uniqueness": "server",
-                },
-                {
-                    "name": "displayName",
-                    "type": "string",
-                    "required": True
                 },
+                {"name": "displayName", "type": "string", "required": True},
                 {
                     "name": "emails",
                     "type": "complex",
                     "multiValued": True,
-                    "required": True
+                    "required": True,
                 },
-                {
+                {"name": "active", "type": "boolean", "required": False},
-                    "name": "active",
+            ],
-                    "type": "boolean",
-                    "required": False
-                }
-            ]
         },
         {
             "schemas": ["urn:ietf:params:scim:schemas:core:2.0:Schema"],
@@ -451,19 +462,15 @@ async def get_schemas():
             "name": "Group",
             "description": "Group",
             "attributes": [
-                {
+                {"name": "displayName", "type": "string", "required": True},
-                    "name": "displayName",
-                    "type": "string",
-                    "required": True
-                },
                 {
                     "name": "members",
                     "type": "complex",
                     "multiValued": True,
-                    "required": False
+                    "required": False,
-                }
+                },
-            ]
+            ],
-        }
+        },
     ]
 
 
@@ -519,8 +526,7 @@ async def get_user(
     user = Users.get_user_by_id(user_id)
     if not user:
         return scim_error(
-            status_code=status.HTTP_404_NOT_FOUND,
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"User {user_id} not found"
-            detail=f"User {user_id} not found"
         )
 
     return user_to_scim(user, request)
@@ -603,7 +609,9 @@ async def update_user(
         if user_data.name.formatted:
             update_data["name"] = user_data.name.formatted
         elif user_data.name.givenName or user_data.name.familyName:
-            update_data["name"] = f"{user_data.name.givenName or ''} {user_data.name.familyName or ''}".strip()
+            update_data["name"] = (
+                f"{user_data.name.givenName or ''} {user_data.name.familyName or ''}".strip()
+            )
 
     if user_data.emails and len(user_data.emails) > 0:
         update_data["email"] = user_data.emails[0].value
@@ -783,6 +791,7 @@ async def create_group(
     # Add members if provided
     if member_ids:
         from open_webui.models.groups import GroupUpdateForm
+
         update_form = GroupUpdateForm(
             name=new_group.name,
             description=new_group.description,