fix sso email case sensitivity

This commit is contained in:
yuneng-jiang
2026-01-26 10:33:57 -08:00
parent b920be3ee7
commit 4ee00cfda5
4 changed files with 151 additions and 9 deletions
+3 -1
View File
@@ -1014,8 +1014,10 @@ async def _get_fuzzy_user_object(
)
if response is None and user_email is not None:
# Use case-insensitive query to handle emails with different casing
# This matches the pattern used in _check_duplicate_user_email
response = await prisma_client.db.litellm_usertable.find_first(
where={"user_email": user_email},
where={"user_email": {"equals": user_email, "mode": "insensitive"}},
include={"organization_memberships": True},
)
+28 -8
View File
@@ -93,6 +93,26 @@ else:
router = APIRouter()
def normalize_email(email: Optional[str]) -> Optional[str]:
"""
Normalize email address to lowercase for consistent storage and comparison.
Email addresses should be treated as case-insensitive for SSO purposes,
even though RFC 5321 technically allows case-sensitive local parts.
This prevents issues where SSO providers return emails with different casing
than what's stored in the database.
Args:
email: Email address to normalize, can be None
Returns:
Lowercased email address, or None if input is None
"""
if email is None:
return None
return email.lower() if isinstance(email, str) else email
def determine_role_from_groups(
user_groups: List[str],
role_mappings: "RoleMappings",
@@ -390,7 +410,7 @@ def generic_response_convertor(
display_name=get_nested_value(
response, generic_user_display_name_attribute_name
),
email=get_nested_value(response, generic_user_email_attribute_name),
email=normalize_email(get_nested_value(response, generic_user_email_attribute_name)),
first_name=get_nested_value(response, generic_user_first_name_attribute_name),
last_name=get_nested_value(response, generic_user_last_name_attribute_name),
provider=get_nested_value(response, generic_provider_attribute_name),
@@ -688,7 +708,7 @@ async def get_user_info_from_db(
if _id is not None and isinstance(_id, str):
potential_user_ids.append(_id)
user_email = (
user_email = normalize_email(
getattr(result, "email", None)
if not isinstance(result, dict)
else result.get("email", None)
@@ -763,8 +783,8 @@ def _build_sso_user_update_data(
Returns:
dict: Update data containing user_email and optionally user_role if valid
"""
update_data: dict = {"user_email": user_email}
"""
update_data: dict = {"user_email": normalize_email(user_email)}
# Get SSO role from result and include if valid
sso_role = getattr(result, "user_role", None)
@@ -1266,7 +1286,7 @@ async def insert_sso_user(
new_user_request = NewUserRequest(
user_id=user_defined_values["user_id"],
user_email=user_defined_values["user_email"],
user_email=normalize_email(user_defined_values["user_email"]),
user_role=user_defined_values["user_role"], # type: ignore
max_budget=user_defined_values["max_budget"],
budget_duration=user_defined_values["budget_duration"],
@@ -1931,7 +1951,7 @@ class SSOAuthenticationHandler:
"""
Gets the user email and id from the OpenID result after validating the email domain
"""
user_email: Optional[str] = getattr(result, "email", None)
user_email: Optional[str] = normalize_email(getattr(result, "email", None))
user_id: Optional[str] = (
getattr(result, "id", None) if result is not None else None
)
@@ -1970,7 +1990,7 @@ class SSOAuthenticationHandler:
"GENERIC_USER_ROLE_ATTRIBUTE", "role"
)
user_id = getattr(result, "id", None)
user_email = getattr(result, "email", None)
user_email = normalize_email(getattr(result, "email", None))
if user_role is None:
_role_from_attr = getattr(result, generic_user_role_attribute_name, None) # type: ignore
if _role_from_attr is not None:
@@ -2363,7 +2383,7 @@ class MicrosoftSSOHandler:
response = response or {}
verbose_proxy_logger.debug(f"Microsoft SSO Callback Response: {response}")
openid_response = CustomOpenID(
email=response.get(MICROSOFT_USER_EMAIL_ATTRIBUTE) or response.get("mail"),
email=normalize_email(response.get(MICROSOFT_USER_EMAIL_ATTRIBUTE) or response.get("mail")),
display_name=response.get(MICROSOFT_USER_DISPLAY_NAME_ATTRIBUTE),
provider="microsoft",
id=response.get(MICROSOFT_USER_ID_ATTRIBUTE),
@@ -28,6 +28,7 @@ from litellm.proxy._types import (
from litellm.proxy.auth.auth_checks import (
ExperimentalUIJWTToken,
_can_object_call_vector_stores,
_get_fuzzy_user_object,
_get_team_db_check,
_virtual_key_max_budget_alert_check,
_virtual_key_soft_budget_check,
@@ -1277,3 +1278,42 @@ async def test_virtual_key_max_budget_alert_check_scenarios(
assert (
alert_triggered == expect_alert
), f"Expected alert_triggered to be {expect_alert} for spend={spend}, max_budget={max_budget}"
@pytest.mark.asyncio
async def test_get_fuzzy_user_object_case_insensitive_email():
"""Test that _get_fuzzy_user_object uses case-insensitive email lookup"""
# Setup mock Prisma client
mock_prisma = MagicMock()
mock_prisma.db = MagicMock()
mock_prisma.db.litellm_usertable = MagicMock()
# Mock user data with mixed case email
test_user = LiteLLM_UserTable(
user_id="test_123",
sso_user_id=None,
user_email="Test@Example.com", # Mixed case in DB
organization_memberships=[],
max_budget=None,
)
# Test: SSO ID not found, find by email with different casing
mock_prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
mock_prisma.db.litellm_usertable.find_first = AsyncMock(return_value=test_user)
# Search with lowercase email (different from DB)
result = await _get_fuzzy_user_object(
prisma_client=mock_prisma,
sso_user_id=None,
user_email="test@example.com", # Lowercase search
)
# Verify user was found despite case difference
assert result == test_user
# Verify the query used case-insensitive mode
mock_prisma.db.litellm_usertable.find_first.assert_called_once()
call_args = mock_prisma.db.litellm_usertable.find_first.call_args
assert call_args.kwargs["where"]["user_email"]["equals"] == "test@example.com"
assert call_args.kwargs["where"]["user_email"]["mode"] == "insensitive"
assert call_args.kwargs["include"] == {"organization_memberships": True}
@@ -24,6 +24,7 @@ from litellm.proxy.management_endpoints.ui_sso import (
GoogleSSOHandler,
MicrosoftSSOHandler,
SSOAuthenticationHandler,
normalize_email,
)
from litellm.types.proxy.management_endpoints.ui_sso import (
DefaultTeamSSOParams,
@@ -667,6 +668,85 @@ def test_build_sso_user_update_data_without_role():
assert "user_role" not in update_data
def test_normalize_email():
"""
Test that normalize_email correctly lowercases email addresses and handles edge cases.
"""
# Test with lowercase email
assert normalize_email("test@example.com") == "test@example.com"
# Test with uppercase email
assert normalize_email("TEST@EXAMPLE.COM") == "test@example.com"
# Test with mixed case email
assert normalize_email("Test.User@Example.COM") == "test.user@example.com"
# Test with None
assert normalize_email(None) is None
# Test with empty string
assert normalize_email("") == ""
def test_build_sso_user_update_data_normalizes_email():
"""
Test that _build_sso_user_update_data normalizes email addresses to lowercase.
"""
from litellm.proxy.management_endpoints.types import CustomOpenID
from litellm.proxy.management_endpoints.ui_sso import _build_sso_user_update_data
sso_result = CustomOpenID(
id="test-user-789",
email="Test.User@Example.COM",
display_name="Test User",
provider="microsoft",
team_ids=[],
user_role=None,
)
update_data = _build_sso_user_update_data(
result=sso_result,
user_email="Test.User@Example.COM",
user_id="test-user-789",
)
# Email should be normalized to lowercase
assert update_data["user_email"] == "test.user@example.com"
assert "user_role" not in update_data
def test_generic_response_convertor_normalizes_email():
"""
Test that generic_response_convertor normalizes email addresses.
"""
from litellm.proxy.management_endpoints.ui_sso import generic_response_convertor
mock_response = {
"preferred_username": "user123",
"email": "Test.User@Example.COM",
"sub": "Test User",
"first_name": "Test",
"last_name": "User",
"provider": "generic",
}
# Mock JWT handler
mock_jwt_handler = MagicMock(spec=JWTHandler)
mock_jwt_handler.get_team_ids_from_jwt.return_value = []
result = generic_response_convertor(
response=mock_response,
jwt_handler=mock_jwt_handler,
sso_jwt_handler=None,
role_mappings=None,
)
# Email should be normalized to lowercase
assert result.email == "test.user@example.com"
assert result.id == "user123"
assert result.display_name == "Test User"
@pytest.mark.asyncio
async def test_upsert_sso_user_updates_role_for_existing_user():
"""