fix(scim): block virtual keys when SCIM deprovisions/deactivates a user

Previously, deleting a user via SCIM (`DELETE /scim/v2/Users/{id}`) or
marking them inactive (`PATCH active=false` / `PUT active=false`) only
touched the user row. Their virtual keys kept working because:

- `litellm_verificationtoken` was never updated.
- The auth path's combined-view query on the key never joined to the
  user's active state.
- `get_user_object()` was wrapped in a silent `except` that set
  `user_obj=None` when the owning user record was gone, so requests
  proceeded normally.

Changes:

- Add `_set_user_keys_blocked(user_id, blocked)` in scim_v2.py that
  flips only mismatched rows via `update_many` and invalidates each
  affected token in the dual cache.
- Cascade SCIM lifecycle events to keys:
  - `delete_user`: block all of the user's keys before deleting the
    user row (preserves spend/audit while orphaning safely).
  - `patch_user` / `update_user`: on `scim_active` transitions,
    block (false) or unblock (true) the user's keys.
- Defense in depth in `user_api_key_auth`: reject the request when the
  loaded `user_obj` has `metadata.scim_active == False`, even if a
  cached key snuck past the per-key block.
- `transform_litellm_user_to_scim_user` now reflects the real
  `scim_active` value instead of always returning `active=True`.

Tests:
- New `test_scim_key_deactivation.py` covering DELETE, PATCH
  active=false, PATCH active=true, no-op patches, and the helper's
  cache-invalidation contract.
- New `test_scim_deactivated_user_key_is_rejected` exercising the
  auth-path defense.
- Existing PATCH tests updated with verificationtoken mocks for the
  new code path.
This commit is contained in:
Claude
2026-04-30 02:01:23 +00:00
parent dedaf74a5e
commit 236e896189
6 changed files with 512 additions and 7 deletions
+13
View File
@@ -1266,6 +1266,19 @@ async def _user_api_key_auth_builder( # noqa: PLR0915
)
user_obj = None
# Defense in depth for SCIM-deprovisioned users: even if a
# cached key snuck past the blocked-flag check, refuse the
# request when the owning user has been marked inactive by
# the SCIM provider.
if (
user_obj is not None
and isinstance(user_obj.metadata, dict)
and user_obj.metadata.get("scim_active") is False
):
raise Exception(
f"User={valid_token.user_id} has been deactivated via SCIM. Keys owned by this user cannot be used."
)
# Check 2a. Check if model has zero cost - if so, skip all budget checks
model = get_model_from_request(request_data, route)
skip_budget_checks = False
@@ -45,6 +45,13 @@ class ScimTransformations:
if user.user_email and "@" in user.user_email:
emails.append(SCIMUserEmail(value=user.user_email, primary=True))
# Reflect SCIM-provider-controlled active state. Default to True for
# users that have never had the flag set (e.g. created before this
# field existed, or created outside SCIM).
metadata = user.metadata or {}
scim_active = metadata.get("scim_active")
active = True if scim_active is None else bool(scim_active)
return SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id=user.user_id,
@@ -56,7 +63,7 @@ class ScimTransformations:
),
emails=emails,
groups=groups,
active=True,
active=active,
meta={
"resourceType": "User",
"created": user_created_at,
@@ -38,6 +38,7 @@ from litellm.proxy._types import (
TeamMemberDeleteRequest,
UserAPIKeyAuth,
)
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.management_endpoints.internal_user_endpoints import new_user
from litellm.proxy.management_endpoints.scim.scim_transformations import (
@@ -336,6 +337,58 @@ async def _handle_team_membership_changes(
)
async def _set_user_keys_blocked(user_id: str, blocked: bool) -> int:
"""
Block or unblock all virtual keys owned by a user and invalidate them in
the in-memory/redis caches so the change takes effect immediately.
Returns the number of keys whose state was flipped. Used by the SCIM
deprovisioning flow so a user's keys stop working the moment SCIM marks
the user inactive (or deletes them).
"""
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
prisma_client = await _get_prisma_client_or_raise_exception()
# Only flip keys whose current state differs — avoids touching keys that
# were already (un)blocked manually by an admin.
affected_keys = await prisma_client.db.litellm_verificationtoken.find_many(
where={"user_id": user_id, "blocked": not blocked},
)
if not affected_keys:
return 0
await prisma_client.db.litellm_verificationtoken.update_many(
where={"user_id": user_id, "blocked": not blocked},
data={"blocked": blocked},
)
for key_row in affected_keys:
await _delete_cache_key_object(
hashed_token=key_row.token,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
verbose_proxy_logger.info(
"SCIM: %s %d virtual key(s) for user_id=%s",
"blocked" if blocked else "unblocked",
len(affected_keys),
user_id,
)
return len(affected_keys)
def _scim_active_value(metadata: Optional[Dict[str, Any]]) -> Optional[bool]:
"""Read the SCIM active flag from a user's metadata dict, if present."""
if not metadata:
return None
value = metadata.get("scim_active")
if value is None:
return None
return bool(value)
async def _create_user_if_not_exists(
user_id: str, created_via: str = "scim_group"
) -> Optional[NewUserResponse]:
@@ -928,6 +981,8 @@ async def update_user(
prisma_client = await _get_prisma_client_or_raise_exception()
existing_user = await _check_user_exists(user_id)
prev_active = _scim_active_value(existing_user.metadata)
# Extract data from SCIM user
user_data = _extract_scim_user_data(user)
@@ -963,6 +1018,13 @@ async def update_user(
data=update_data,
)
# Cascade SCIM active transitions to virtual keys (mirrors PATCH).
new_active = _scim_active_value(metadata)
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)
# Convert back to SCIM format
scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
updated_user
@@ -1009,6 +1071,12 @@ async def delete_user(
where={"team_id": team.team_id}, data={"members": new_members}
)
# Block the user's virtual keys before deleting the user record.
# The user row going away leaves the keys orphaned; without this
# they'd keep working because the auth path silently tolerates a
# missing owner.
await _set_user_keys_blocked(user_id=user_id, blocked=True)
# Delete user
await prisma_client.db.litellm_usertable.delete(where={"user_id": user_id})
@@ -1242,11 +1310,15 @@ async def patch_user(
prisma_client = await _get_prisma_client_or_raise_exception()
existing_user = await _check_user_exists(user_id)
prev_active = _scim_active_value(existing_user.metadata)
update_data, final_team_set = _apply_patch_ops(
existing_user=existing_user,
patch_ops=patch_ops,
)
new_active = _scim_active_value(update_data.get("metadata"))
# Handle team membership changes
await _handle_team_membership_changes(
user_id=user_id,
@@ -1267,6 +1339,14 @@ async def patch_user(
data=update_data,
)
# Cascade SCIM active transitions to virtual keys. Treat "previously
# unset" as active=True so a first-time PATCH with active=false still
# blocks any pre-existing keys.
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)
scim_user = await ScimTransformations.transform_litellm_user_to_scim_user(
updated_user
)
@@ -798,6 +798,99 @@ async def test_proxy_admin_expired_key_from_cache():
setattr(_proxy_server_mod, attr, val)
@pytest.mark.asyncio
async def test_scim_deactivated_user_key_is_rejected():
"""A virtual key whose owning user has metadata.scim_active=False must be
rejected by the auth flow (defense in depth on top of key-level blocking).
"""
from fastapi import Request
from starlette.datastructures import URL
from litellm.proxy.auth.user_api_key_auth import _user_api_key_auth_builder
from litellm.proxy.proxy_server import hash_token
api_key = "sk-scim-deactivated-user-key"
hashed_key = hash_token(api_key)
valid_token = UserAPIKeyAuth(
api_key=api_key,
token=hashed_key,
user_id="scim-disabled-user",
)
deactivated_user = LiteLLM_UserTable(
user_id="scim-disabled-user",
metadata={"scim_active": False},
)
mock_cache = AsyncMock()
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.delete_cache = MagicMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.internal_usage_cache = MagicMock()
mock_proxy_logging_obj.internal_usage_cache.dual_cache = AsyncMock()
mock_proxy_logging_obj.internal_usage_cache.dual_cache.async_delete_cache = (
AsyncMock()
)
mock_proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None)
mock_prisma_client = MagicMock()
import litellm.proxy.proxy_server as _proxy_server_mod
_attrs_to_set = {
"prisma_client": mock_prisma_client,
"user_api_key_cache": mock_cache,
"proxy_logging_obj": mock_proxy_logging_obj,
"master_key": "sk-master-key",
"general_settings": {},
"llm_model_list": [],
"llm_router": None,
"open_telemetry_logger": None,
"model_max_budget_limiter": MagicMock(),
"user_custom_auth": None,
"jwt_handler": None,
"litellm_proxy_admin_name": "admin",
}
_original_values = {
attr: getattr(_proxy_server_mod, attr, None) for attr in _attrs_to_set
}
try:
for attr, val in _attrs_to_set.items():
setattr(_proxy_server_mod, attr, val)
request = Request(scope={"type": "http"})
request._url = URL(url="/chat/completions")
with (
patch(
"litellm.proxy.auth.user_api_key_auth.get_key_object",
new_callable=AsyncMock,
return_value=valid_token,
),
patch(
"litellm.proxy.auth.user_api_key_auth.get_user_object",
new_callable=AsyncMock,
return_value=deactivated_user,
),
):
with pytest.raises(ProxyException) as exc_info:
await _user_api_key_auth_builder(
request=request,
api_key=f"Bearer {api_key}",
azure_api_key_header="",
anthropic_api_key_header=None,
google_ai_studio_api_key_header=None,
azure_apim_header=None,
request_data={},
)
assert "deactivated via SCIM" in str(exc_info.value.message)
finally:
for attr, val in _original_values.items():
setattr(_proxy_server_mod, attr, val)
@pytest.mark.asyncio
async def test_return_user_api_key_auth_obj_user_spend_and_budget():
"""
@@ -1752,7 +1845,11 @@ async def test_team_metadata_refreshed_from_team_object_during_auth():
from starlette.datastructures import URL
from starlette.requests import Request
from litellm.proxy._types import LiteLLM_TeamTableCachedObj, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy._types import (
LiteLLM_TeamTableCachedObj,
LitellmUserRoles,
UserAPIKeyAuth,
)
from litellm.proxy.auth.user_api_key_auth import _user_api_key_auth_builder
api_key = "sk-test-team-metadata-refresh"
@@ -1833,16 +1930,17 @@ async def test_team_metadata_refreshed_from_team_object_during_auth():
request_data={},
)
assert result.team_metadata == {"guardrails": ["test-guardrail-333"]}, (
f"team_metadata was not updated from fresh team object. Got: {result.team_metadata}"
)
assert result.team_metadata == {
"guardrails": ["test-guardrail-333"]
}, f"team_metadata was not updated from fresh team object. Got: {result.team_metadata}"
finally:
for k, v in _originals.items():
setattr(_proxy_server_mod, k, v)
# ---------------------------------------------------------------------------
# _run_centralized_common_checks — centralized authz gate
# ---------------------------------------------------------------------------
@@ -0,0 +1,300 @@
"""Tests for SCIM-driven virtual key deactivation.
When a SCIM provider deprovisions a user (DELETE) or marks them inactive
(PATCH/PUT with active=False), virtual keys owned by that user must stop
working immediately. Reactivating (active=True) must un-block them.
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from litellm.proxy._types import LiteLLM_UserTable
from litellm.proxy.management_endpoints.scim.scim_v2 import (
_set_user_keys_blocked,
delete_user,
patch_user,
)
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIMPatchOp,
SCIMPatchOperation,
SCIMUser,
SCIMUserEmail,
SCIMUserName,
)
def _build_token_row(token: str, user_id: str, blocked: bool):
row = MagicMock()
row.token = token
row.user_id = user_id
row.blocked = blocked
return row
def _build_prisma_with_keys(user_keys, mock_user=None, updated_user=None):
mock_client = MagicMock()
mock_db = MagicMock()
mock_client.db = mock_db
if mock_user is not None:
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
if updated_user is not None:
mock_db.litellm_usertable.update = AsyncMock(return_value=updated_user)
mock_db.litellm_usertable.delete = AsyncMock(return_value=None)
mock_db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
mock_db.litellm_verificationtoken.find_many = AsyncMock(return_value=user_keys)
mock_db.litellm_verificationtoken.update_many = AsyncMock(return_value=None)
return mock_client, mock_db
@pytest.mark.asyncio
async def test_set_user_keys_blocked_flips_state_and_invalidates_cache():
"""_set_user_keys_blocked must update_many AND invalidate each token in the cache."""
keys = [
_build_token_row("hash-1", "user-x", blocked=False),
_build_token_row("hash-2", "user-x", blocked=False),
]
mock_client, mock_db = _build_prisma_with_keys(keys)
cache_deletions = []
async def fake_delete(hashed_token, user_api_key_cache, proxy_logging_obj):
cache_deletions.append(hashed_token)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(side_effect=fake_delete),
),
):
flipped = await _set_user_keys_blocked(user_id="user-x", blocked=True)
assert flipped == 2
mock_db.litellm_verificationtoken.update_many.assert_awaited_once_with(
where={"user_id": "user-x", "blocked": False},
data={"blocked": True},
)
assert sorted(cache_deletions) == ["hash-1", "hash-2"]
@pytest.mark.asyncio
async def test_set_user_keys_blocked_noop_when_no_matching_keys():
"""If no keys match the desired flip, neither update_many nor cache delete runs."""
mock_client, mock_db = _build_prisma_with_keys(user_keys=[])
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(),
) as mocked_delete,
):
flipped = await _set_user_keys_blocked(user_id="user-x", blocked=True)
assert flipped == 0
mock_db.litellm_verificationtoken.update_many.assert_not_called()
mocked_delete.assert_not_called()
@pytest.mark.asyncio
async def test_scim_delete_user_blocks_keys_before_deleting_user():
"""SCIM DELETE /Users/{id} must block the user's keys before removing the row."""
user_id = "user-to-delete"
mock_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={},
)
keys = [_build_token_row("hash-a", user_id, blocked=False)]
mock_client, mock_db = _build_prisma_with_keys(keys, mock_user=mock_user)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(),
),
):
response = await delete_user(user_id=user_id)
assert response.status_code == 204
mock_db.litellm_verificationtoken.update_many.assert_awaited_once_with(
where={"user_id": user_id, "blocked": False},
data={"blocked": True},
)
mock_db.litellm_usertable.delete.assert_awaited_once_with(
where={"user_id": user_id}
)
@pytest.mark.asyncio
async def test_scim_patch_user_active_false_blocks_keys():
user_id = "scim-user"
mock_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": True},
)
updated_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": False, "scim_metadata": {}},
)
keys = [_build_token_row("hash-z", user_id, blocked=False)]
mock_client, mock_db = _build_prisma_with_keys(
keys, mock_user=mock_user, updated_user=updated_user
)
patch_ops = SCIMPatchOp(
Operations=[SCIMPatchOperation(op="replace", path="active", value="False")]
)
mock_scim_user = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id=user_id,
userName=user_id,
name=SCIMUserName(familyName="X", givenName="Y"),
emails=[SCIMUserEmail(value="x@example.com")],
active=False,
)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user",
AsyncMock(return_value=mock_scim_user),
),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(),
),
):
await patch_user(user_id=user_id, patch_ops=patch_ops)
mock_db.litellm_verificationtoken.update_many.assert_awaited_once_with(
where={"user_id": user_id, "blocked": False},
data={"blocked": True},
)
@pytest.mark.asyncio
async def test_scim_patch_user_active_true_unblocks_keys():
user_id = "scim-user"
mock_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": False},
)
updated_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": True, "scim_metadata": {}},
)
keys = [_build_token_row("hash-r", user_id, blocked=True)]
mock_client, mock_db = _build_prisma_with_keys(
keys, mock_user=mock_user, updated_user=updated_user
)
patch_ops = SCIMPatchOp(
Operations=[SCIMPatchOperation(op="replace", path="active", value="True")]
)
mock_scim_user = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id=user_id,
userName=user_id,
name=SCIMUserName(familyName="X", givenName="Y"),
emails=[SCIMUserEmail(value="x@example.com")],
active=True,
)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user",
AsyncMock(return_value=mock_scim_user),
),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(),
),
):
await patch_user(user_id=user_id, patch_ops=patch_ops)
mock_db.litellm_verificationtoken.update_many.assert_awaited_once_with(
where={"user_id": user_id, "blocked": True},
data={"blocked": False},
)
@pytest.mark.asyncio
async def test_scim_patch_user_no_active_change_does_not_touch_keys():
"""A patch that doesn't flip active must not call update_many on tokens."""
user_id = "scim-user"
mock_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias="Old",
teams=[],
metadata={"scim_active": True},
)
updated_user = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias="New",
teams=[],
metadata={"scim_active": True, "scim_metadata": {}},
)
mock_client, mock_db = _build_prisma_with_keys(
user_keys=[], mock_user=mock_user, updated_user=updated_user
)
patch_ops = SCIMPatchOp(
Operations=[SCIMPatchOperation(op="replace", path="displayName", value="New")]
)
mock_scim_user = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
id=user_id,
userName=user_id,
name=SCIMUserName(familyName="X", givenName="Y"),
emails=[SCIMUserEmail(value="x@example.com")],
active=True,
)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_user_to_scim_user",
AsyncMock(return_value=mock_scim_user),
),
patch(
"litellm.proxy.management_endpoints.scim.scim_v2._delete_cache_key_object",
AsyncMock(),
),
):
await patch_user(user_id=user_id, patch_ops=patch_ops)
mock_db.litellm_verificationtoken.find_many.assert_not_called()
mock_db.litellm_verificationtoken.update_many.assert_not_called()
@@ -42,6 +42,9 @@ async def test_patch_user_updates_fields():
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
mock_db.litellm_usertable.update = AsyncMock(side_effect=mock_update)
mock_db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
# active=False triggers cascading key-block. No keys here, so return [].
mock_db.litellm_verificationtoken.find_many = AsyncMock(return_value=[])
mock_db.litellm_verificationtoken.update_many = AsyncMock(return_value=None)
# Mock the transformation function to return a proper SCIMUser
mock_scim_user = SCIMUser(
@@ -194,6 +197,8 @@ async def test_patch_user_deprovision_without_path():
mock_client.db = mock_db
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
mock_db.litellm_usertable.update = AsyncMock(side_effect=mock_update)
mock_db.litellm_verificationtoken.find_many = AsyncMock(return_value=[])
mock_db.litellm_verificationtoken.update_many = AsyncMock(return_value=None)
mock_scim_user = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
@@ -271,6 +276,8 @@ async def test_patch_user_multiple_fields_without_path():
mock_client.db = mock_db
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
mock_db.litellm_usertable.update = AsyncMock(side_effect=mock_update)
mock_db.litellm_verificationtoken.find_many = AsyncMock(return_value=[])
mock_db.litellm_verificationtoken.update_many = AsyncMock(return_value=None)
mock_scim_user = SCIMUser(
schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],