ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix SCIM user lookup filters (#27308)

Browse Source 
* Fix SCIM Okta userName lookup

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* fix scim user filter typing

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

---------

Co-authored-by: oss-agent-shin <279349115+oss-agent-shin@users.noreply.github.com>
Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>
This commit is contained in:
oss-agent-shin 2026-05-06 11:58:47 -07:00 committed by GitHub
parent c92a08a307
commit d90cf56245
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 148 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
litellm/proxy/management_endpoints/scim/scim_v2.py
View File

@ -4,6 +4,7 @@
This is an enterprise feature and requires a premium license.
"""
import re
from typing import Any, Dict, List, Optional, Set, Tuple
from fastapi import (
@ -843,6 +844,18 @@ async def get_service_provider_config(request: Request):
return SCIMServiceProviderConfig(meta=meta)
def _parse_scim_eq_filter(scim_filter: str) -> Optional[Tuple[str, str]]:
"""Parse the SCIM equality filters Okta uses before user lifecycle changes."""
match = re.match(
r"""\s*([\w.]+)\s+eq\s+(['"]?)(.*?)\2\s*$""",
scim_filter,
flags=re.IGNORECASE,
)
if not match:
return None
return match.group(1).lower(), match.group(3)
# User Endpoints
@scim_router.get(
"/Users",
@ -867,15 +880,21 @@ async def get_users(
try:
prisma_client = await _get_prisma_client_or_raise_exception()
# Parse filter if provided (basic support)
where_conditions = {}
where_conditions: Dict[str, Any] = {}
if filter:
# Very basic filter support - only handling userName eq and emails.value eq
if "userName eq" in filter:
user_id = filter.split("userName eq ")[1].strip("\"'")
where_conditions["user_id"] = user_id
elif "emails.value eq" in filter:
email = filter.split("emails.value eq ")[1].strip("\"'")
where_conditions["user_email"] = email
# Okta locates users by userName before deprovisioning. LiteLLM
# exposes SCIM userName from user_email, while older SCIM-created
# users may still have user_id == userName, so support both.
parsed_filter = _parse_scim_eq_filter(filter)
if parsed_filter:
filter_attribute, filter_value = parsed_filter
if filter_attribute == "username":
where_conditions["OR"] = [
{"user_email": filter_value},
{"user_id": filter_value},
]
elif filter_attribute == "emails.value":
where_conditions["user_email"] = filter_value
# Get users from database
users: List[LiteLLM_UserTable] = (

123
tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py
View File

@ -4,6 +4,7 @@ import pytest
from fastapi import HTTPException
from litellm.proxy._types import (
LiteLLM_UserTable,
LitellmUserRoles,
NewUserRequest,
NewUserResponse,
@ -16,8 +17,8 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
_process_group_patch_operations,
create_group,
create_user,
get_users,
get_service_provider_config,
patch_group,
patch_user,
update_group,
update_user,
@ -259,6 +260,124 @@ async def test_scim_create_user_respects_default_role_set_via_ui(mocker, monkeyp
)
@pytest.mark.asyncio
async def test_get_users_filters_username_by_exposed_scim_username_for_okta(mocker):
"""
Okta deprovisioning first locates a user with `userName eq "<email>"`.
LiteLLM exposes SCIM userName from user_email, so the lookup must match
user_email even when the internal user_id is a UUID.
"""
user = LiteLLM_UserTable(
user_id="internal-user-id",
user_email="okta.user@example.com",
user_alias="Okta User",
teams=[],
metadata={},
)
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_many = AsyncMock(return_value=[user])
mock_prisma_client.db.litellm_usertable.count = AsyncMock(return_value=1)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=mock_prisma_client),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user",
AsyncMock(
return_value=SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id="internal-user-id",
userName="okta.user@example.com",
emails=[SCIMUserEmail(value="okta.user@example.com")],
)
),
)
response = await get_users(
startIndex=1,
count=10,
filter='userName eq "okta.user@example.com"',
)
expected_where = {
"OR": [
{"user_email": "okta.user@example.com"},
{"user_id": "okta.user@example.com"},
]
}
mock_prisma_client.db.litellm_usertable.find_many.assert_awaited_once_with(
where=expected_where,
skip=0,
take=10,
order={"created_at": "desc"},
)
mock_prisma_client.db.litellm_usertable.count.assert_awaited_once_with(
where=expected_where
)
assert response.totalResults == 1
assert response.Resources[0].id == "internal-user-id"
@pytest.mark.asyncio
async def test_get_users_filters_email_value_by_user_email(mocker):
"""
SCIM clients can locate users with `emails.value eq "<email>"`; keep that
filter as a direct user_email lookup alongside the userName fallback query.
"""
user = LiteLLM_UserTable(
user_id="internal-user-id",
user_email="scim.user@example.com",
user_alias="SCIM User",
teams=[],
metadata={},
)
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_many = AsyncMock(return_value=[user])
mock_prisma_client.db.litellm_usertable.count = AsyncMock(return_value=1)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
AsyncMock(return_value=mock_prisma_client),
)
mocker.patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user",
AsyncMock(
return_value=SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id="internal-user-id",
userName="scim.user@example.com",
emails=[SCIMUserEmail(value="scim.user@example.com")],
)
),
)
response = await get_users(
startIndex=1,
count=10,
filter='emails.value eq "scim.user@example.com"',
)
expected_where = {"user_email": "scim.user@example.com"}
mock_prisma_client.db.litellm_usertable.find_many.assert_awaited_once_with(
where=expected_where,
skip=0,
take=10,
order={"created_at": "desc"},
)
mock_prisma_client.db.litellm_usertable.count.assert_awaited_once_with(
where=expected_where
)
assert response.totalResults == 1
assert response.Resources[0].id == "internal-user-id"
@pytest.mark.asyncio
async def test_handle_existing_user_by_email_no_email(mocker):
"""Should return None when new_user_request has no email"""
@ -1337,7 +1456,7 @@ async def test_create_group_with_nonexistent_users_creates_when_flag_true(
)
# Execute the create_group function - should succeed
result = await create_group(group=scim_group)
await create_group(group=scim_group)
# Verify users were created
assert mock_create_user.call_count == 2