mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
Merge 68c47418b5 into 2ce4e3f8a9
This commit is contained in:
commit
965d94f2c3
3 changed files with 77 additions and 22 deletions
|
|
@ -4,12 +4,43 @@ Types for the management endpoints
|
|||
Might include fastapi/proxy requirements.txt related imports
|
||||
"""
|
||||
|
||||
from collections.abc import Iterable, Sequence
|
||||
from typing import Any, Final, cast
|
||||
|
||||
from fastapi_sso.sso.base import OpenID
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles
|
||||
|
||||
# Ordered highest to lowest privilege
|
||||
LITELLM_USER_ROLE_HIERARCHY: Final = (
|
||||
LitellmUserRoles.PROXY_ADMIN,
|
||||
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
|
||||
LitellmUserRoles.INTERNAL_USER,
|
||||
LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
|
||||
)
|
||||
|
||||
|
||||
def highest_privilege_role(roles: Iterable[LitellmUserRoles]) -> LitellmUserRoles | None:
|
||||
"""
|
||||
Pick the most privileged role a user was granted.
|
||||
|
||||
SSO providers emit multi-valued role claims in an arbitrary order, so resolving by
|
||||
privilege keeps a user's role stable across logins. Roles the hierarchy does not rank
|
||||
(org_admin, team, customer) resolve by name to stay deterministic.
|
||||
"""
|
||||
granted: Final = frozenset(roles)
|
||||
if not granted:
|
||||
return None
|
||||
ranked: Final = next((role for role in LITELLM_USER_ROLE_HIERARCHY if role in granted), None)
|
||||
return ranked if ranked is not None else min(granted, key=lambda role: role.value)
|
||||
|
||||
|
||||
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 +64,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 +75,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(
|
||||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -111,8 +111,10 @@ 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 (
|
||||
LITELLM_USER_ROLE_HIERARCHY,
|
||||
CustomOpenID,
|
||||
get_litellm_user_role,
|
||||
highest_privilege_role,
|
||||
is_valid_litellm_user_role,
|
||||
)
|
||||
from litellm.proxy.utils import (
|
||||
|
|
@ -808,15 +810,6 @@ def normalize_email(email: str | None) -> str | None:
|
|||
return email.lower() if isinstance(email, str) else email
|
||||
|
||||
|
||||
# Ordered highest to lowest privilege
|
||||
LITELLM_USER_ROLE_HIERARCHY: Final = (
|
||||
LitellmUserRoles.PROXY_ADMIN,
|
||||
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
|
||||
LitellmUserRoles.INTERNAL_USER,
|
||||
LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
|
||||
)
|
||||
|
||||
|
||||
def determine_role_from_groups(
|
||||
user_groups: list[str],
|
||||
role_mappings: "RoleMappings",
|
||||
|
|
@ -4287,14 +4280,9 @@ class MicrosoftSSOHandler:
|
|||
listed first. Roles the hierarchy does not rank (org_admin, team, customer)
|
||||
resolve by name to stay deterministic
|
||||
"""
|
||||
resolved: Final = frozenset(
|
||||
return highest_privilege_role(
|
||||
role for role in (get_litellm_user_role(role_str) for role_str in app_roles or ()) if role is not None
|
||||
)
|
||||
if not resolved:
|
||||
return None
|
||||
|
||||
ranked: Final = next((role for role in LITELLM_USER_ROLE_HIERARCHY if role in resolved), None)
|
||||
return ranked if ranked is not None else min(resolved, key=lambda role: role.value)
|
||||
|
||||
@staticmethod
|
||||
def get_app_roles_from_id_token(id_token: str | None) -> list[str]:
|
||||
|
|
|
|||
|
|
@ -4535,6 +4535,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"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue