From a885e9a7b88ccc1cc57c2aeeff9dddb521279bc5 Mon Sep 17 00:00:00 2001
From: shivam <shivam@berri.ai>
Date: Fri, 6 Nov 2026 00:26:13 +0000
Subject: [PATCH] fix(proxy): propagate SSO user admission rejections

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 litellm/proxy/management_endpoints/ui_sso.py  |  2 +-
 .../proxy/management_endpoints/test_ui_sso.py | 45 +++++++++++++++++++
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/litellm/proxy/management_endpoints/ui_sso.py b/litellm/proxy/management_endpoints/ui_sso.py
index 53079d84fdd..5c5bdebd7f7 100644
--- a/litellm/proxy/management_endpoints/ui_sso.py
+++ b/litellm/proxy/management_endpoints/ui_sso.py
@@ -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
diff --git a/tests/unit/proxy/management_endpoints/test_ui_sso.py b/tests/unit/proxy/management_endpoints/test_ui_sso.py
index ba78fefd653..c6ad3eb5a4f 100644
--- a/tests/unit/proxy/management_endpoints/test_ui_sso.py
+++ b/tests/unit/proxy/management_endpoints/test_ui_sso.py
@@ -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"]})