fix(scim): preserve scim_active on PUT when client omits the field

A SCIM PUT may legally omit `active` (full-replace with the field
absent). Pydantic fills the SCIMUser.active default of True, so the PUT
handler was overwriting metadata.scim_active with True even when the
client never sent it — silently reactivating a previously SCIM-blocked
user and unblocking their keys.

Use model_fields_set to detect whether the client actually sent
`active`. If omitted, preserve the prior scim_active value and skip
the cascade to virtual keys.

Also drop comments added in this PR that just narrate what the code
does; keep only the docstrings and the SQL-NULL pitfall note that
explain non-obvious behaviour.

Co-authored-by: Mateo Wang <mateo-berri@users.noreply.github.com>
This commit is contained in:
Cursor Agent
2026-05-02 04:15:25 +00:00
co-authored by Mateo Wang
parent 5faf75dceb
commit 8cf3bf11bd
4 changed files with 146 additions and 40 deletions
-4
View File
@@ -1307,10 +1307,6 @@ 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)
@@ -45,9 +45,6 @@ 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)
@@ -363,9 +363,9 @@ async def _set_user_keys_blocked(user_id: str, blocked: bool) -> int:
prisma_client = await _get_prisma_client_or_raise_exception()
if blocked:
# Block keys that aren't already blocked. `blocked` is a nullable column
# with no default so existing rows typically hold NULL; treat NULL as
# "not blocked" so SQL equality on NULL doesn't silently skip them.
# `blocked` is a nullable column with no default, so existing rows
# typically hold NULL; treat NULL as "not blocked" since SQL equality
# on NULL would otherwise silently skip them.
candidates = await prisma_client.db.litellm_verificationtoken.find_many(
where={
"user_id": user_id,
@@ -374,8 +374,6 @@ async def _set_user_keys_blocked(user_id: str, blocked: bool) -> int:
)
affected_keys = candidates
else:
# Only unblock keys that SCIM previously blocked. An admin-managed
# block has no `scim_blocked` marker and must not be reversed here.
candidates = await prisma_client.db.litellm_verificationtoken.find_many(
where={"user_id": user_id, "blocked": True},
)
@@ -384,9 +382,6 @@ async def _set_user_keys_blocked(user_id: str, blocked: bool) -> int:
if not affected_keys:
return 0
# Per-key updates: we need to add/remove the SCIM-block marker in JSON
# metadata, which `update_many` can't express. Cardinality is bounded by
# the number of keys a single user owns.
for key_row in affected_keys:
current_metadata: Dict[str, Any] = (
dict(key_row.metadata) if isinstance(key_row.metadata, dict) else {}
@@ -1048,47 +1043,49 @@ async def update_user(
prev_active = _scim_active_value(existing_user.metadata)
# Extract data from SCIM user
user_data = _extract_scim_user_data(user)
# Build metadata with SCIM data
# SCIM PUT may legally omit `active` (full-replace with the field absent).
# Pydantic fills the model default, so distinguish "client sent active"
# from "client omitted it" via model_fields_set, and preserve the prior
# SCIM active state when omitted — otherwise a vanilla PUT to a
# deactivated user would silently re-enable them and unblock their keys.
client_set_active = "active" in user.model_fields_set
scim_active_for_metadata = (
user_data["active"] if client_set_active else prev_active
)
metadata = _build_scim_metadata(
user_data["given_name"], user_data["family_name"], user_data["active"]
user_data["given_name"],
user_data["family_name"],
scim_active_for_metadata,
)
# Handle team membership changes
await _handle_team_membership_changes(
user_id=user_id,
existing_teams=existing_user.teams or [],
new_teams=user_data["teams"],
)
# Update user with all new data (full replacement)
update_data = {
"user_email": user_data["user_email"],
"user_alias": user_data["user_alias"],
"sso_user_id": user_data["sso_user_id"],
"teams": user_data["teams"],
"metadata": metadata,
"metadata": safe_dumps(metadata),
}
# Serialize metadata to JSON string for Prisma to avoid GraphQL parsing issues
if "metadata" in update_data and isinstance(update_data["metadata"], dict):
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
update_data["metadata"] = safe_dumps(update_data["metadata"])
updated_user = await prisma_client.db.litellm_usertable.update(
where={"user_id": user_id},
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)
if client_set_active:
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(
@@ -1136,10 +1133,6 @@ 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)
await _delete_rows_referencing_user(prisma_client, user_id=user_id)
@@ -1406,9 +1399,6 @@ 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
):
@@ -14,6 +14,7 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
_set_user_keys_blocked,
delete_user,
patch_user,
update_user,
)
from litellm.types.proxy.management_endpoints.scim_v2 import (
SCIMPatchOp,
@@ -418,3 +419,125 @@ async def test_scim_patch_user_no_active_change_does_not_touch_keys():
mock_db.litellm_verificationtoken.find_many.assert_not_called()
mock_db.litellm_verificationtoken.update_many.assert_not_called()
def _build_put_user_payload(user_id: str, **overrides) -> dict:
payload = {
"schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
"id": user_id,
"userName": user_id,
"name": {"givenName": "Y", "familyName": "X"},
"emails": [{"value": "x@example.com", "primary": True}],
}
payload.update(overrides)
return payload
@pytest.mark.asyncio
async def test_scim_put_user_omitting_active_preserves_deactivated_state():
"""PUT without `active` must not silently reactivate a SCIM-deactivated user
nor unblock their SCIM-blocked keys."""
user_id = "scim-user"
deactivated = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": False},
)
keys = [
_build_token_row(
"hash-keep-blocked", user_id, blocked=True, metadata={"scim_blocked": True}
)
]
mock_client, mock_db = _build_prisma_with_keys(
keys, mock_user=deactivated, updated_user=deactivated
)
put_user = SCIMUser.model_validate(_build_put_user_payload(user_id))
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 update_user(user_id=user_id, user=put_user)
mock_db.litellm_verificationtoken.update.assert_not_called()
mock_db.litellm_verificationtoken.update_many.assert_not_called()
mock_db.litellm_usertable.update.assert_awaited_once()
update_kwargs = mock_db.litellm_usertable.update.await_args.kwargs
assert '"scim_active": false' in update_kwargs["data"]["metadata"]
@pytest.mark.asyncio
async def test_scim_put_user_explicit_active_false_blocks_keys():
"""PUT explicitly setting active=False on an active user must cascade to keys."""
user_id = "scim-user"
active = LiteLLM_UserTable(
user_id=user_id,
user_email="x@example.com",
user_alias=None,
teams=[],
metadata={"scim_active": True},
)
deactivated = 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-block-me", user_id, blocked=False)]
mock_client, mock_db = _build_prisma_with_keys(
keys, mock_user=active, updated_user=deactivated
)
put_user = SCIMUser.model_validate(_build_put_user_payload(user_id, active=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 update_user(user_id=user_id, user=put_user)
mock_db.litellm_verificationtoken.update.assert_awaited_once()
update_kwargs = mock_db.litellm_verificationtoken.update.await_args.kwargs
assert update_kwargs["where"] == {"token": "hash-block-me"}
assert update_kwargs["data"]["blocked"] is True
assert '"scim_blocked": true' in update_kwargs["data"]["metadata"]