scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-14 09:03:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 214

fix(proxy): propagate SSO user admission rejections

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
shivam 2026-11-06 00:26:13 +00:00
parent c17181e0f1
commit a885e9a7b8
2 changed files with 46 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -2938,7 +2938,7 @@ async def insert_sso_user(
user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN),
)
except ProxyException as error:
if premium_user is True:
if premium_user is True and error.code != "403":
raise
raise _SSOUserAdmissionError(
message=error.message, type=error.type, param=error.param, code=error.code

45
tests/unit/proxy/management_endpoints/test_ui_sso.py
View file

@ -25,6 +25,7 @@ from litellm.proxy._types import (
ProxyException,
)
from litellm.proxy.auth.handle_jwt import JWTHandler
from litellm.proxy.auth.litellm_license import LicenseCheck
from litellm.proxy.management_endpoints.sso import CustomMicrosoftSSO
from litellm.proxy.management_endpoints.types import CustomOpenID
from litellm.proxy.management_endpoints.ui_sso import (
@ -1458,6 +1459,11 @@ class _UserTable:
async def find_unique(self, where: dict[str, str]) -> LiteLLM_UserTable | None:
return self.rows.get(where["user_id"])
async def count(self, where: Mapping[str, object] | None = None) -> int:
if where is None:
return len(self.rows)
return sum((row.metadata or {}).get("scim_active") is False for row in self.rows.values())
def _users_db(table: _UserTable) -> SimpleNamespace:
database: Final = SimpleNamespace(litellm_usertable=table)
@ -1502,6 +1508,45 @@ async def test_first_sign_in_does_not_sign_in_to_another_account_holding_the_ema
await _first_sign_in(_UserTable(other))
@pytest.mark.parametrize("premium", [False, True], ids=["free", "premium"])
@pytest.mark.asyncio
async def test_sso_upsert_propagates_license_rejection_without_creating_the_user(premium: bool) -> None:
license_check: Final = LicenseCheck()
license_check.airgapped_license_data = {"max_users": 1}
occupants: Final = tuple(LiteLLM_UserTable(user_id=f"licensed-seat-{index}") for index in range(2))
table: Final = _UserTable(*occupants)
prisma: Final = _users_db(table)
with (
patch("litellm.proxy.proxy_server.prisma_client", prisma),
patch("litellm.proxy.proxy_server._license_check", license_check),
patch("litellm.proxy.proxy_server.premium_user", premium),
pytest.raises(ProxyException) as rejected,
):
await SSOAuthenticationHandler.upsert_sso_user(
result=CustomOpenID(id="sso-subject", email="person@example.com", team_ids=[]),
user_info=None,
user_email="person@example.com",
user_defined_values={
"user_id": "sso-subject",
"user_email": "person@example.com",
"user_role": None,
"max_budget": None,
"budget_duration": None,
"models": [],
},
prisma_client=prisma,
)
assert rejected.value.to_dict() == {
"message": "User count limit reached for your LiteLLM License. Please contact your administrator",
"type": "auth_error",
"param": None,
"code": "403",
}
assert table.rows == {row.user_id: row for row in occupants}
@pytest.mark.asyncio
async def test_first_sign_in_joins_default_teams_the_concurrent_first_sign_in_has_not_joined_yet(monkeypatch):
monkeypatch.setattr(litellm, "default_internal_user_params", {"teams": ["team-joined", "team-pending"]})