scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

feat(proxy): support deterministic fake SSO email when provider hides email

Browse source
This commit is contained in:
naaa760 2026-03-20 17:18:29 +05:30
parent d7c419bfee
commit 6e419a5e7a
2 changed files with 165 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
litellm/proxy/management_endpoints/ui_sso.py
View file

@ -133,6 +133,38 @@ def normalize_email(email: Optional[str]) -> Optional[str]:
return email.lower() if isinstance(email, str) else email
def _is_empty_email(email: Optional[str]) -> bool:
"""True if email is None or blank string."""
if email is None:
return True
return isinstance(email, str) and email.strip() == ""
def _derive_fake_email_from_sso_id(sso_user_id: str) -> str:
"""
Derive a deterministic fake email from SSO user ID.
Used when SSO provider does not reveal user email (e.g. Lark with email disabled).
"""
digest = hashlib.sha256(sso_user_id.encode()).hexdigest()[:24]
return f"{digest}@sso.example.com"
def _maybe_apply_fake_email(
email: Optional[str], sso_user_id: Optional[str]
) -> Optional[str]:
"""
If GENERIC_FAKE_EMAIL is enabled and email is empty but sso_user_id exists,
return a deterministic fake email. Otherwise return the original email.
"""
if os.getenv("GENERIC_FAKE_EMAIL", "").lower() != "true":
return email
if not sso_user_id or not isinstance(sso_user_id, str) or not sso_user_id.strip():
return email
if not _is_empty_email(email):
return email
return _derive_fake_email_from_sso_id(sso_user_id)
def determine_role_from_groups(
user_groups: List[str],
role_mappings: "RoleMappings",
@ -525,14 +557,19 @@ def generic_response_convertor(
attr_name = attr_name.strip()
extra_fields[attr_name] = get_nested_value(response, attr_name)
user_id = get_nested_value(response, generic_user_id_attribute_name)
raw_email = get_nested_value(response, generic_user_email_attribute_name)
email = normalize_email(raw_email)
email = _maybe_apply_fake_email(email, user_id)
if _is_empty_email(email):
email = None
return CustomOpenID(
id=get_nested_value(response, generic_user_id_attribute_name),
id=user_id,
display_name=get_nested_value(
response, generic_user_display_name_attribute_name
),
email=normalize_email(
get_nested_value(response, generic_user_email_attribute_name)
),
email=email,
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),
@ -1062,6 +1099,8 @@ async def get_user_info_from_db(
if not isinstance(result, dict)
else result.get("email", None)
)
sso_user_id = potential_user_ids[0] if potential_user_ids else None
user_email = _maybe_apply_fake_email(user_email, sso_user_id)
user_info: Optional[Union[LiteLLM_UserTable, NewUserResponse]] = None
@ -2287,10 +2326,15 @@ class SSOAuthenticationHandler:
"""
Gets the user email and id from the OpenID result after validating the email domain
"""
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
)
if result is None:
user_email = None
user_id = None
elif isinstance(result, dict):
user_email = normalize_email(result.get("email"))
user_id = result.get("id")
else:
user_email = normalize_email(getattr(result, "email", None))
user_id = getattr(result, "id", None)
user_role: Optional[str] = None
if user_email is not None and os.getenv("ALLOWED_EMAIL_DOMAINS") is not None:
@ -2308,7 +2352,11 @@ class SSOAuthenticationHandler:
# Extract user_role from result (works for all SSO providers)
if result is not None:
_user_role = getattr(result, "user_role", None)
_user_role = (
result.get("user_role")
if isinstance(result, dict)
else getattr(result, "user_role", None)
)
if _user_role is not None:
# Convert enum to string if needed
user_role = (
@ -2325,10 +2373,18 @@ class SSOAuthenticationHandler:
generic_user_role_attribute_name = os.getenv(
"GENERIC_USER_ROLE_ATTRIBUTE", "role"
)
user_id = getattr(result, "id", None)
user_email = normalize_email(getattr(result, "email", None))
if isinstance(result, dict):
user_id = result.get("id")
user_email = normalize_email(result.get("email"))
else:
user_id = getattr(result, "id", 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
_role_from_attr = (
result.get(generic_user_role_attribute_name)
if isinstance(result, dict)
else getattr(result, generic_user_role_attribute_name, None) # type: ignore
)
if _role_from_attr is not None:
# Convert enum to string if needed
user_role = (
@ -2338,13 +2394,19 @@ class SSOAuthenticationHandler:
)
if user_id is None and result is not None:
_first_name = getattr(result, "first_name", "") or ""
_last_name = getattr(result, "last_name", "") or ""
if isinstance(result, dict):
_first_name = result.get("first_name") or ""
_last_name = result.get("last_name") or ""
else:
_first_name = getattr(result, "first_name", "") or ""
_last_name = getattr(result, "last_name", "") or ""
user_id = _first_name + _last_name
if user_email is not None and (user_id is None or len(user_id) == 0):
user_id = user_email
user_email = _maybe_apply_fake_email(user_email, user_id)
return ParsedOpenIDResult(
user_email=user_email,
user_id=user_id,

89
tests/test_litellm/proxy/management_endpoints/test_ui_sso.py
View file

@ -23,6 +23,8 @@ from litellm.proxy.management_endpoints.ui_sso import (
GoogleSSOHandler,
MicrosoftSSOHandler,
SSOAuthenticationHandler,
_derive_fake_email_from_sso_id,
_maybe_apply_fake_email,
_setup_team_mappings,
determine_role_from_groups,
normalize_email,
@ -702,6 +704,63 @@ def test_normalize_email():
assert normalize_email("") == ""
def test_derive_fake_email_from_sso_id_deterministic():
"""Same SSO ID always yields same fake email."""
email1 = _derive_fake_email_from_sso_id("user-123")
email2 = _derive_fake_email_from_sso_id("user-123")
assert email1 == email2
assert email1.endswith("@sso.example.com")
def test_maybe_apply_fake_email_when_disabled():
"""When GENERIC_FAKE_EMAIL is not set, return original email."""
with patch.dict(os.environ, {}, clear=True):
os.environ.pop("GENERIC_FAKE_EMAIL", None)
result = _maybe_apply_fake_email(None, "user-123")
assert result is None
result = _maybe_apply_fake_email("", "user-123")
assert result == ""
def test_maybe_apply_fake_email_when_enabled_empty_email():
"""When GENERIC_FAKE_EMAIL is true and email is empty, derive fake email."""
with patch.dict(os.environ, {"GENERIC_FAKE_EMAIL": "true"}):
result = _maybe_apply_fake_email(None, "lark-user-456")
assert result is not None
assert result.endswith("@sso.example.com")
assert result == _derive_fake_email_from_sso_id("lark-user-456")
def test_maybe_apply_fake_email_when_enabled_has_email():
"""When GENERIC_FAKE_EMAIL is true but email exists, keep original."""
with patch.dict(os.environ, {"GENERIC_FAKE_EMAIL": "true"}):
result = _maybe_apply_fake_email("real@example.com", "user-123")
assert result == "real@example.com"
def test_get_user_email_and_id_with_empty_email_generic_fake_enabled():
"""When SSO returns empty email and GENERIC_FAKE_EMAIL is true, use derived fake email."""
result = {"id": "lark-user-789", "email": "", "display_name": "Test User"}
with patch.dict(os.environ, {"GENERIC_FAKE_EMAIL": "true"}):
parsed = SSOAuthenticationHandler._get_user_email_and_id_from_result(
result=result, generic_client_id="generic"
)
assert parsed.get("user_email") is not None
assert parsed.get("user_email").endswith("@sso.example.com")
assert parsed.get("user_id") == "lark-user-789"
def test_get_user_email_and_id_with_empty_email_generic_fake_disabled():
"""When SSO returns empty email and GENERIC_FAKE_EMAIL is false, email stays empty."""
result = {"id": "lark-user-789", "email": "", "display_name": "Test User"}
with patch.dict(os.environ, {}, clear=True):
os.environ.pop("GENERIC_FAKE_EMAIL", None)
parsed = SSOAuthenticationHandler._get_user_email_and_id_from_result(
result=result, generic_client_id="generic"
)
assert parsed.get("user_email") == ""
def test_build_sso_user_update_data_normalizes_email():
"""
Test that _build_sso_user_update_data normalizes email addresses to lowercase.
@ -761,6 +820,36 @@ def test_generic_response_convertor_normalizes_email():
assert result.display_name == "Test User"
def test_generic_response_convertor_empty_email_fake_enabled():
"""When SSO returns empty email and GENERIC_FAKE_EMAIL is true, use derived fake email."""
from litellm.proxy.management_endpoints.ui_sso import (
_derive_fake_email_from_sso_id,
generic_response_convertor,
)
mock_response = {
"preferred_username": "lark-user-456",
"email": "",
"sub": "Lark User",
"first_name": "Lark",
"last_name": "User",
"provider": "lark",
}
mock_jwt_handler = MagicMock(spec=JWTHandler)
mock_jwt_handler.get_team_ids_from_jwt.return_value = []
with patch.dict(os.environ, {"GENERIC_FAKE_EMAIL": "true"}):
result = generic_response_convertor(
response=mock_response,
jwt_handler=mock_jwt_handler,
sso_jwt_handler=None,
role_mappings=None,
)
assert result.email == _derive_fake_email_from_sso_id("lark-user-456")
assert result.id == "lark-user-456"
@pytest.mark.asyncio
async def test_upsert_sso_user_updates_role_for_existing_user():
"""