fix(auth): enforce SCIM status for admin JWTs and refresh SCIM caches

litellm/proxy/auth/handle_jwt.py

@@ -2463,7 +2463,7 @@ class JWTAuthManager:
                 proxy_logging_obj=proxy_logging_obj,
                 team_id_upsert=team_id_upsert,
             )
-            if provisioning is None:
+            if provisioning is None or prisma_client is not None:
                 identity: Final = await JWTAuthManager._resolve_claim_identity(
                     jwt_valid_token, handler, prisma_client, user_api_key_cache, parent_otel_span, proxy_logging_obj
                 )

litellm/proxy/auth/user_api_key_auth.py

@@ -1714,6 +1714,16 @@ async def _user_api_key_auth_builder(
                     jwt_claims = result.get("jwt_claims", None)
                     agent_id: Final[str | None] = result.get("agent_id")
 
+                    if (
+                        user_object is not None
+                        and isinstance(user_object.metadata, dict)
+                        and user_object.metadata.get("scim_active") is False
+                    ):
+                        raise HTTPException(
+                            status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail=f"User={user_id} has been deactivated via SCIM. Keys owned by this user cannot be used.",
+                        )
+
                     if is_proxy_admin:
                         # Proxy admins authenticate via auth_builder (full
                         # access), not via a mapped virtual key. If
@@ -1730,16 +1740,6 @@ async def _user_api_key_auth_builder(
                             )
                         return JWTAuthManager.user_api_key_auth_from_result(result, parent_otel_span)
 
-                    if (
-                        user_object is not None
-                        and isinstance(user_object.metadata, dict)
-                        and user_object.metadata.get("scim_active") is False
-                    ):
-                        raise HTTPException(
-                            status_code=status.HTTP_401_UNAUTHORIZED,
-                            detail=f"User={user_id} has been deactivated via SCIM. Keys owned by this user cannot be used.",
-                        )
-
                     valid_token = JWTAuthManager.user_api_key_auth_from_result(result, parent_otel_span)
 
                     # AUTO_REGISTER deferred from _resolve_jwt_to_virtual_key.

litellm/proxy/management_endpoints/scim/scim_v2.py

@@ -46,6 +46,7 @@ from litellm.proxy._types import (
 )
 from litellm.proxy.auth.auth_checks import _delete_cache_key_object
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import evict_and_broadcast
 from litellm.proxy.common_utils.http_parsing_utils import _safe_get_request_headers
 from litellm.proxy.management_endpoints.internal_user_endpoints import new_user
 from litellm.proxy.management_endpoints.scim.scim_transformations import (
@@ -1804,6 +1805,9 @@ async def update_user(
             where={"user_id": user_id},
             data=update_data,
         )
+        from litellm.proxy.proxy_server import user_api_key_cache
+
+        await evict_and_broadcast(cache_keys=(user_id,), user_api_key_cache=user_api_key_cache)
 
         if client_set_active:
             new_active: Final = _scim_active_value(metadata)
@@ -2375,6 +2379,9 @@ async def patch_user(
             where={"user_id": user_id},
             data=update_data,
         )
+        from litellm.proxy.proxy_server import user_api_key_cache
+
+        await evict_and_broadcast(cache_keys=(user_id,), user_api_key_cache=user_api_key_cache)
 
         if new_active is not None and new_active != (True if prev_active is None else prev_active):
             await _set_user_keys_blocked(user_id=user_id, blocked=not new_active)

tests/test_litellm/proxy/auth/test_handle_jwt.py

@@ -7144,3 +7144,46 @@ async def test_admin_jwt_team_header_only_provisions_during_admission(monkeypatc
     else:
         create_team.assert_not_awaited()
         assert result["team_id"] is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("existing_user", [False, True])
+@pytest.mark.parametrize("warm_cache", [False, True])
+async def test_scope_admin_admission_resolves_existing_user_without_provisioning(
+    monkeypatch: pytest.MonkeyPatch, existing_user: bool, warm_cache: bool
+) -> None:
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+
+    private_key, jwk = _get_rsa_key_and_jwk("admin-status")
+    cache: Final = UserApiKeyCache()
+    cache.set_cache("litellm_jwt_auth_keys_https://admin.example/jwks", [jwk])
+    user_id: Final = f"admin-status-{existing_user}-{warm_cache}"
+    user: Final = LiteLLM_UserTable(user_id=user_id, metadata={"scim_active": False}, organization_memberships=[])
+    if existing_user and warm_cache:
+        cache.set_cache(user_id, user)
+    database: Final = MagicMock()
+    users: Final = database.db.litellm_usertable
+    users.find_unique = AsyncMock(return_value=user if existing_user else None)
+    users.find_first = AsyncMock(return_value=None)
+    users.create = AsyncMock()
+    handler: Final = JWTHandler()
+    handler.update_environment(
+        prisma_client=database,
+        user_api_key_cache=cache,
+        litellm_jwtauth=LiteLLM_JWTAuth(user_id_jwt_field="sub", user_id_upsert=True),
+    )
+    monkeypatch.setenv("JWT_PUBLIC_KEY_URL", "https://admin.example/jwks")
+    monkeypatch.setenv("JWT_ISSUER", "https://admin.example")
+    monkeypatch.setenv("JWT_AUDIENCE", "gateway")
+    token: Final = _encode_rsa_jwt(
+        private_key, "https://admin.example", "gateway", "admin-status",
+        {"sub": user_id, "scope": "litellm_proxy_admin"},
+    )
+    result: Final = await JWTAuthManager.auth_builder(
+        api_key=token, jwt_handler=handler, prisma_client=database, user_api_key_cache=cache,
+        parent_otel_span=None, proxy_logging_obj=MagicMock(), request_data={}, general_settings={}, route="/user/info",
+    )
+    assert result["is_proxy_admin"] is True
+    assert result["user_id"] == user_id
+    assert result["user_object"] == (user if existing_user else None)
+    users.create.assert_not_awaited()

tests/test_litellm/proxy/auth/test_user_api_key_auth.py

@@ -7393,18 +7393,21 @@ class TestJWTAuthUserEmail:
         assert result.user_email == "resolved@example.com"
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("route", ["/mcp-rest/tools/list", "/mcp-rest/tools/call", "/v1/chat/completions"])
+    @pytest.mark.parametrize("route", ["/mcp-rest/tools/list", "/mcp-rest/tools/call", "/v1/chat/completions", "/user/info"])
     @pytest.mark.parametrize("active", [False, True, None, "false", 0])
-    async def test_jwt_auth_rejects_deactivated_user(self, route: str, active: bool | str | int | None) -> None:
+    @pytest.mark.parametrize("is_admin", [False, True])
+    async def test_jwt_auth_rejects_deactivated_user(
+        self, route: str, active: bool | str | int | None, is_admin: bool
+    ) -> None:
         from typing import Final
 
         jwt_token: Final = "eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ1c2VyMSJ9.signature"
         result: Final = {
-            "is_proxy_admin": False,
+            "is_proxy_admin": is_admin,
             "team_object": None,
             "user_object": LiteLLM_UserTable(
                 user_id="jwt-human-user",
-                user_role=LitellmUserRoles.INTERNAL_USER.value,
+                user_role=LitellmUserRoles.PROXY_ADMIN.value if is_admin else LitellmUserRoles.INTERNAL_USER.value,
                 metadata={} if active is None else {"scim_active": active},
             ),
             "end_user_object": None,

tests/test_litellm/proxy/management_endpoints/scim/test_scim_key_deactivation.py

@@ -541,3 +541,63 @@ async def test_scim_put_user_explicit_active_false_blocks_keys():
     assert update_kwargs["where"] == {"token": "hash-block-me"}
     assert update_kwargs["data"]["blocked"] is True
     assert '"scim_blocked": true' in update_kwargs["data"]["metadata"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["PUT", "PATCH"])
+@pytest.mark.parametrize("active", [False, True])
+@pytest.mark.parametrize("failure", [None, "write", "keys"])
+@pytest.mark.parametrize("status_change", [False, True])
+async def test_scim_status_write_refreshes_user_cache(
+    method: str, active: bool, failure: str | None, status_change: bool
+) -> None:
+    import json
+    from typing import Final
+
+    from litellm.proxy._types import ProxyException
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+
+    user_id: Final = "scim-cache-user"
+    saved: Final = LiteLLM_UserTable(
+        user_id=user_id, user_email="x@example.com", teams=[], metadata={"scim_active": not active if status_change else active},
+    )
+    updated: Final = LiteLLM_UserTable(
+        user_id=user_id, user_email="x@example.com", teams=[], metadata={"scim_active": active},
+    )
+    client, db = _build_prisma_with_keys([], mock_user=saved.model_copy(deep=True), updated_user=updated)
+    if failure == "write":
+        db.litellm_usertable.update.side_effect = RuntimeError("status write failed")
+    if failure == "keys":
+        db.litellm_verificationtoken.find_many.side_effect = RuntimeError("key update failed")
+    cache: Final = UserApiKeyCache()
+    await cache.async_set_cache(key=user_id, value=saved, model_type=LiteLLM_UserTable)
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", client),  # test-quality-ok: substitute the database dependency
+        patch("litellm.proxy.proxy_server.user_api_key_cache", cache),  # test-quality-ok: exercise a real isolated cache
+        patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),  # test-quality-ok: isolate the logging dependency
+        patch(  # test-quality-ok: observe the Redis publication boundary
+            "litellm.proxy.common_utils.auth_cache_invalidation_pubsub.publish_auth_cache_invalidation",
+            new_callable=AsyncMock,
+        ) as broadcast,
+    ):
+        request: Final = (
+            update_user(user_id=user_id, user=SCIMUser.model_validate(_build_put_user_payload(user_id, active=active)))
+            if method == "PUT" else
+            patch_user(user_id=user_id, patch_ops=SCIMPatchOp(
+                Operations=[SCIMPatchOperation(op="replace", path="active", value=active)]
+            ))
+        )
+        if failure == "write" or (failure == "keys" and status_change):
+            with pytest.raises(ProxyException, match="status write failed" if failure == "write" else "key update failed"):
+                await request
+        else:
+            response: Final = await request
+            assert response.active is active
+        assert json.loads(db.litellm_usertable.update.await_args.kwargs["data"]["metadata"])["scim_active"] is active
+        cached: Final = await cache.async_get_cache(key=user_id, model_type=LiteLLM_UserTable)
+        if failure == "write":
+            assert cached == saved
+            broadcast.assert_not_awaited()
+        else:
+            assert cached is None
+            broadcast.assert_awaited_once_with(cache_key=user_id)