From 6e419a5e7ab0f6e44f86cd1219d359c7974e705d Mon Sep 17 00:00:00 2001 From: naaa760 Date: Fri, 20 Mar 2026 17:18:29 +0530 Subject: [PATCH] feat(proxy): support deterministic fake SSO email when provider hides email --- litellm/proxy/management_endpoints/ui_sso.py | 90 ++++++++++++++++--- .../proxy/management_endpoints/test_ui_sso.py | 89 ++++++++++++++++++ 2 files changed, 165 insertions(+), 14 deletions(-) diff --git a/litellm/proxy/management_endpoints/ui_sso.py b/litellm/proxy/management_endpoints/ui_sso.py index daf1d6f1316..33b96147fd9 100644 --- a/litellm/proxy/management_endpoints/ui_sso.py +++ b/litellm/proxy/management_endpoints/ui_sso.py @@ -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, diff --git a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py b/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py index d43b2c4ba05..54b757677c0 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py +++ b/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py @@ -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(): """