Merge f1ec7a5753 into b03e913ccf

litellm/proxy/management_endpoints/types.py

@@ -4,12 +4,42 @@ Types for the management endpoints
 Might include fastapi/proxy requirements.txt related imports
 """
 
+from collections.abc import Sequence
 from typing import Any, Final, cast
 
 from fastapi_sso.sso.base import OpenID
 
 from litellm.proxy._types import LitellmUserRoles
 
+ROLE_PRIVILEGE_ORDER: Final = (
+    LitellmUserRoles.PROXY_ADMIN,
+    LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
+    LitellmUserRoles.INTERNAL_USER,
+    LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
+)
+
+
+def highest_privilege_role(roles: Sequence[LitellmUserRoles]) -> LitellmUserRoles | None:
+    """
+    Pick the most privileged role a user was granted.
+
+    SSO providers emit multi-valued role/group claims in an arbitrary order, so resolving by
+    privilege keeps a user's role stable across logins. Roles outside ROLE_PRIVILEGE_ORDER
+    (org_admin, team, customer) are not comparable, so the first of those is kept.
+    """
+    granted: Final = frozenset(roles)
+    ranked: Final = next((role for role in ROLE_PRIVILEGE_ORDER if role in granted), None)
+    if ranked is not None:
+        return ranked
+    return roles[0] if roles else None
+
+
+def _lookup_role(role_str: object) -> LitellmUserRoles | None:
+    if not isinstance(role_str, str):
+        return None
+    result: Final = LitellmUserRoles._value2member_map_.get(role_str.lower())
+    return cast(LitellmUserRoles | None, result)
+
 
 def is_valid_litellm_user_role(role_str: str) -> bool:
     """
@@ -33,7 +63,8 @@ def get_litellm_user_role(role_str) -> LitellmUserRoles | None:
     Convert a string (or list of strings) to a LitellmUserRoles enum if valid (case-insensitive).
 
     Handles list inputs since some SSO providers (e.g., Keycloak) return roles
-    as arrays like ["proxy_admin"] instead of plain strings.
+    as arrays like ["proxy_admin"] instead of plain strings. A user granted several
+    roles gets the most privileged one, not whichever the provider happened to list first.
 
     Args:
         role_str: String or list to convert (e.g., "proxy_admin", ["proxy_admin"])
@@ -43,12 +74,11 @@ def get_litellm_user_role(role_str) -> LitellmUserRoles | None:
     """
     try:
         if isinstance(role_str, list):
-            if len(role_str) == 0:
-                return None
-            role_str = role_str[0]
-        # Use _value2member_map_ for O(1) lookup, case-insensitive
-        result: Final = LitellmUserRoles._value2member_map_.get(role_str.lower())
-        return cast(LitellmUserRoles | None, result)
+            entries: Final[Sequence[object]] = role_str
+            return highest_privilege_role(
+                tuple(role for role in (_lookup_role(entry) for entry in entries) if role is not None)
+            )
+        return _lookup_role(role_str)
     except Exception:
         return None
 

litellm/proxy/management_endpoints/ui_sso.py

@@ -111,6 +111,7 @@ from litellm.proxy.management_endpoints.sso_helper_utils import (
 )
 from litellm.proxy.management_endpoints.team_endpoints import new_team, team_member_add
 from litellm.proxy.management_endpoints.types import (
+    ROLE_PRIVILEGE_ORDER,
     CustomOpenID,
     get_litellm_user_role,
     is_valid_litellm_user_role,
@@ -832,19 +833,11 @@ def determine_role_from_groups(
         # No role mappings configured, return default_role
         return role_mappings.default_role
 
-    # Role hierarchy (highest to lowest)
-    role_hierarchy: Final = [
-        LitellmUserRoles.PROXY_ADMIN,
-        LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
-        LitellmUserRoles.INTERNAL_USER,
-        LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
-    ]
-
     # Convert user_groups to a set for efficient lookup
     user_groups_set: Final = set(user_groups) if isinstance(user_groups, list) else set()
 
     # Find the highest privilege role the user belongs to
-    for role in role_hierarchy:
+    for role in ROLE_PRIVILEGE_ORDER:
         if role in role_mappings.roles:
             role_groups = role_mappings.roles[role]
             if isinstance(role_groups, list) and user_groups_set.intersection(set(role_groups)):
@@ -4236,15 +4229,9 @@ class MicrosoftSSOHandler:
         verbose_proxy_logger.debug("Extracted app roles from id_token: %s", app_roles)
 
         # Combine groups and app roles
-        user_role: LitellmUserRoles | None = None
-        if app_roles:
-            # Check if any app role is a valid LitellmUserRoles
-            for role_str in app_roles:
-                role = get_litellm_user_role(role_str)
-                if role is not None:
-                    user_role = role
-                    verbose_proxy_logger.debug("Found valid LitellmUserRoles '%s' in app_roles", role.value)
-                    break
+        user_role: Final = get_litellm_user_role(app_roles) if app_roles else None
+        if user_role is not None:
+            verbose_proxy_logger.debug("Resolved role '%s' from app_roles %s", user_role.value, app_roles)
 
         verbose_proxy_logger.debug("Combined team_ids (groups + app roles): %s", user_team_ids)
 

tests/test_litellm/proxy/management_endpoints/test_entraid_app_roles.py

@@ -1,3 +1,4 @@
+import pytest
 import jwt
 
 from litellm.proxy.management_endpoints.ui_sso import MicrosoftSSOHandler
@@ -89,3 +90,26 @@ def test_defaults_to_internal_user_viewer_when_no_role():
     # Default role would be internal_user_viewer
     default_role = LitellmUserRoles.INTERNAL_USER_VIEW_ONLY
     assert default_role.value == "internal_user_viewer"
+
+@pytest.mark.parametrize(
+    "app_roles",
+    [
+        ["proxy_admin_viewer", "internal_user"],
+        ["internal_user", "proxy_admin_viewer"],
+    ],
+)
+def test_multiple_app_roles_resolve_to_highest_privilege(app_roles):
+    """A user granted several app roles keeps the most privileged one, whatever order the token lists them in."""
+    payload = {
+        "sub": "user_multi_role",
+        "email": "multi@company.com",
+        "app_roles": app_roles,
+        "aud": "litellm-app",
+        "iss": "https://login.microsoftonline.com/tenant-id/v2.0",
+        "exp": 9999999999,
+    }
+
+    token = jwt.encode(payload, "secret", algorithm="HS256")
+    roles = MicrosoftSSOHandler.get_app_roles_from_id_token(token)
+
+    assert get_litellm_user_role(roles) == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY

tests/test_litellm/proxy/management_endpoints/test_ui_sso.py

@@ -4523,6 +4523,42 @@ class TestGenericResponseConvertorUserRole:
             assert isinstance(result, CustomOpenID)
             assert result.user_role is None
 
+    @pytest.mark.parametrize(
+        "role_claim",
+        [
+            ["proxy_admin_viewer", "internal_user"],
+            ["internal_user", "proxy_admin_viewer"],
+            ["internal_user", "unknown_role", "proxy_admin_viewer"],
+        ],
+    )
+    def test_generic_response_convertor_multi_valued_role_claim_picks_highest_privilege(self, role_claim):
+        """
+        A user in several mapped groups keeps the most privileged role, regardless of the
+        order the SSO provider lists the roles in.
+        """
+        from litellm.proxy._types import LitellmUserRoles
+        from litellm.proxy.management_endpoints.ui_sso import generic_response_convertor
+
+        mock_jwt_handler = MagicMock(spec=JWTHandler)
+        mock_jwt_handler.get_team_ids_from_jwt.return_value = []
+
+        sso_response = {
+            "preferred_username": "testuser",
+            "email": "test@example.com",
+            "sub": "Test User",
+            "role": role_claim,
+        }
+
+        with patch.dict(os.environ, {"GENERIC_USER_ROLE_ATTRIBUTE": "role"}):
+            result = generic_response_convertor(
+                response=sso_response,
+                jwt_handler=mock_jwt_handler,
+                sso_jwt_handler=None,
+            )
+
+        assert isinstance(result, CustomOpenID)
+        assert result.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
+
 
 class TestGetGenericSSORedirectParams:
     """Test _get_generic_sso_redirect_params state parameter priority handling"""