fix(scim): harden PATCH multi-valued ops and fail-soft directory metadata reads

litellm/proxy/management_endpoints/scim/scim_transformations.py

@@ -1,5 +1,8 @@
-from typing import List, Union
+from typing import Callable, List, TypeVar, Union
 
+from pydantic import ValidationError
+
+from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import (
     LiteLLM_TeamTable,
     LiteLLM_UserTable,
@@ -9,6 +12,8 @@ from litellm.proxy._types import (
 from litellm.repositories.team_repository import TeamRepository
 from litellm.types.proxy.management_endpoints.scim_v2 import *
 
+T = TypeVar("T")
+
 
 class ScimTransformations:
     DEFAULT_SCIM_NAME = "Unknown User"
@@ -47,16 +52,18 @@ class ScimTransformations:
         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])
+        enterprise_user = ScimTransformations._parse_directory_metadata(
+            user, SCIM_ENTERPRISE_METADATA_KEY, SCIMEnterpriseUser.model_validate
+        )
+        if enterprise_user is not None:
             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
+        entitlements = ScimTransformations._parse_directory_metadata(
+            user, SCIM_ENTITLEMENTS_METADATA_KEY, SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python
+        )
+        roles = ScimTransformations._parse_directory_metadata(
+            user, SCIM_ROLES_METADATA_KEY, SCIM_MULTI_VALUED_LIST_ADAPTER.validate_python
+        )
 
         return SCIMUser(
             schemas=schemas,
@@ -80,6 +87,31 @@ class ScimTransformations:
             },
         )
 
+    @staticmethod
+    def _parse_directory_metadata(
+        user: Union[LiteLLM_UserTable, NewUserResponse],
+        key: str,
+        validate: Callable[[object], T],
+    ) -> T | None:
+        """A SCIM directory attribute parsed from user metadata, or None when absent or malformed.
+
+        Metadata is writable outside the SCIM surface, so a malformed value on one user must not
+        fail the whole directory response; the attribute is omitted and the corruption logged.
+        """
+        metadata = user.metadata or {}
+        raw = metadata.get(key)
+        if not raw:
+            return None
+        try:
+            return validate(raw)
+        except ValidationError:
+            verbose_proxy_logger.warning(
+                "Skipping malformed %s metadata on user %s in SCIM response",
+                key,
+                user.user_id,
+            )
+            return None
+
     @staticmethod
     def _get_scim_user_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
         """

litellm/proxy/management_endpoints/scim/scim_v2.py

@@ -1383,20 +1383,38 @@ def _handle_group_operations(op_type: str, value: Any, teams_set: Set[str]) -> O
     return None
 
 
+def _multi_valued_attribute_base(path: str) -> str:
+    """The attribute name a SCIM path targets, stripped of any value filter or sub-attribute."""
+    return path.split("[", 1)[0].split(".", 1)[0]
+
+
 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
+    base = _multi_valued_attribute_base(path)
+    metadata_key = SCIM_MULTI_VALUED_ATTRIBUTE_METADATA_KEYS[base]
+    if path != base:
+        raise HTTPException(
+            status_code=400,
+            detail={"error": f"Filtered or sub-attribute paths are not supported for {base}; PATCH the full attribute"},
+        )
+
     if op_type == "remove":
         metadata.pop(metadata_key, None)
         return
 
+    if value is None:
+        raise HTTPException(
+            status_code=400,
+            detail={"error": f"The {op_type} operation on {base} requires a 'value' member (RFC 7644 Section 3.5.2)"},
+        )
+
     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"},
+            detail={"error": f"Invalid value for {base}: expected a list of objects with a 'value' sub-attribute"},
         )
 
     dumped = [attr.model_dump(exclude_none=True) for attr in attrs]
@@ -1442,7 +1460,7 @@ 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"):
+                elif key_lower in SCIM_MULTI_VALUED_ATTRIBUTE_METADATA_KEYS:
                     _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():
@@ -1464,7 +1482,7 @@ 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"):
+        elif _multi_valued_attribute_base(path) in SCIM_MULTI_VALUED_ATTRIBUTE_METADATA_KEYS:
             _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)

litellm/types/proxy/management_endpoints/scim_v2.py

@@ -73,6 +73,11 @@ class SCIMMultiValuedAttribute(BaseModel):
 
 SCIM_MULTI_VALUED_LIST_ADAPTER = TypeAdapter(List[SCIMMultiValuedAttribute])
 
+SCIM_MULTI_VALUED_ATTRIBUTE_METADATA_KEYS = {
+    "entitlements": SCIM_ENTITLEMENTS_METADATA_KEY,
+    "roles": SCIM_ROLES_METADATA_KEY,
+}
+
 
 class SCIMUserManager(BaseModel):
     model_config = ConfigDict(populate_by_name=True)

tests/test_litellm/proxy/management_endpoints/scim/test_scim_patch_user.py

@@ -431,3 +431,37 @@ def test_apply_patch_ops_invalid_entitlements_value_raises_400():
         _apply_patch_ops(existing_user=_user_with_metadata({}), patch_ops=patch_ops)
 
     assert exc_info.value.status_code == 400
+
+
+def test_apply_patch_ops_add_without_value_raises_400_naming_value_member():
+    patch_ops = SCIMPatchOp(
+        Operations=[SCIMPatchOperation(op="add", path="entitlements")]
+    )
+
+    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
+    assert "value" in str(exc_info.value.detail)
+
+
+def test_apply_patch_ops_filtered_path_raises_400_instead_of_junk_metadata():
+    """A filtered path must fail loudly rather than fall through to the generic
+    handler, which would write a junk metadata key while reporting success"""
+    patch_ops = SCIMPatchOp(
+        Operations=[
+            SCIMPatchOperation(
+                op="remove", path='roles[value eq "engineering-admin"]'
+            )
+        ]
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        _apply_patch_ops(
+            existing_user=_user_with_metadata(
+                {"scim_roles": [{"value": "engineering-admin"}]}
+            ),
+            patch_ops=patch_ops,
+        )
+
+    assert exc_info.value.status_code == 400

tests/test_litellm/proxy/management_endpoints/scim/test_scim_transformations.py

@@ -214,6 +214,40 @@ class TestScimTransformations:
             assert scim_user.roles[0].value == "engineering-admin"
             assert scim_user.roles[0].primary is True
 
+    @pytest.mark.asyncio
+    async def test_transform_user_with_malformed_directory_metadata_fails_soft(
+        self, mock_prisma_client
+    ):
+        """Metadata is writable outside the SCIM surface; a corrupted value on one
+        user must omit the attribute, not fail the whole directory response"""
+        mock_client, mock_find_unique = mock_prisma_client
+        mock_find_unique.return_value = None
+
+        user = LiteLLM_UserTable(
+            user_id="user-corrupt",
+            user_email="corrupt@example.com",
+            user_alias=None,
+            teams=[],
+            created_at=None,
+            updated_at=None,
+            metadata={
+                "scim_entitlements": [{"display": 123}],
+                "scim_roles": {"value": "not-a-list"},
+                "scim_enterprise": {"manager": 42},
+            },
+        )
+
+        with patch("litellm.proxy.proxy_server.prisma_client", mock_client):
+            scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
+                user
+            )
+
+            assert scim_user.id == "user-corrupt"
+            assert scim_user.entitlements is None
+            assert scim_user.roles is None
+            assert scim_user.enterprise_user is None
+            assert SCIM_ENTERPRISE_USER_SCHEMA not in scim_user.schemas
+
     @pytest.mark.asyncio
     async def test_transform_user_without_enterprise_metadata_omits_schema(
         self, mock_user, mock_prisma_client