mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
test(integration): cover scim endpoint contracts (#44656)
* test(integration): cover scim endpoint contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): assert scim admin key access and required active Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): cover scim cold-cache reactivation, entra membership ops, username filter and external id Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): assert full SCIM responses and denied writes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): pin full SCIM resources and group filters Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): deny non-admin SCIM user deletes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: mateo <mateo@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
8a2f706ab7
commit
11e91f464a
3 changed files with 1922 additions and 14 deletions
|
|
@ -1,24 +1,127 @@
|
|||
import os
|
||||
import uuid
|
||||
from contextlib import ExitStack
|
||||
from hashlib import sha256
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import psycopg
|
||||
import pytest
|
||||
from hypothesis import strategies as st
|
||||
from hypothesis.stateful import RuleBasedStateMachine, invariant, rule, run_state_machine_as_test
|
||||
from pydantic import JsonValue
|
||||
from pydantic import BaseModel, ConfigDict, JsonValue
|
||||
|
||||
from tests.integration._support.client import Gateway, eventually, object_value, team_admin_permissions
|
||||
from tests.integration._support.client import Gateway, eventually, object_value, string_value, team_admin_permissions
|
||||
from tests.integration._support.database import read_rows
|
||||
from tests.integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests
|
||||
|
||||
|
||||
SCIM_CORE_USER_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:User"
|
||||
|
||||
|
||||
class ScimNameResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
givenName: str | None = None
|
||||
familyName: str | None = None
|
||||
formatted: str | None = None
|
||||
middleName: str | None = None
|
||||
honorificPrefix: str | None = None
|
||||
honorificSuffix: str | None = None
|
||||
|
||||
|
||||
class ScimMetaResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
resourceType: str
|
||||
created: str
|
||||
lastModified: str
|
||||
|
||||
|
||||
class ScimEmailResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
value: str
|
||||
type: str | None = None
|
||||
primary: bool | None = None
|
||||
|
||||
|
||||
class ScimGroupRefResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
value: str
|
||||
display: str | None = None
|
||||
type: str | None = None
|
||||
|
||||
|
||||
class ScimMultiValueResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
value: str | None = None
|
||||
display: str | None = None
|
||||
type: str | None = None
|
||||
primary: bool | None = None
|
||||
|
||||
|
||||
class ScimUserResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
schemas: list[str]
|
||||
id: str
|
||||
externalId: str | None = None
|
||||
meta: ScimMetaResponse
|
||||
userName: str
|
||||
displayName: str
|
||||
active: bool
|
||||
name: ScimNameResponse
|
||||
emails: list[ScimEmailResponse]
|
||||
groups: list[ScimGroupRefResponse]
|
||||
entitlements: list[ScimMultiValueResponse] | None = None
|
||||
roles: list[ScimMultiValueResponse] | None = None
|
||||
|
||||
|
||||
def _assert_scim_user_resource(
|
||||
response: httpx.Response,
|
||||
user_id: str,
|
||||
active: bool,
|
||||
expected_created: str | None,
|
||||
user_name: str,
|
||||
name: ScimNameResponse,
|
||||
emails: list[ScimEmailResponse],
|
||||
) -> ScimUserResponse:
|
||||
actual: Final = ScimUserResponse.model_validate(response.json())
|
||||
created: Final = actual.meta.created if expected_created is None else expected_created
|
||||
assert isinstance(created, str), response.text
|
||||
assert isinstance(actual.meta.lastModified, str), response.text
|
||||
expected: Final = ScimUserResponse(
|
||||
schemas=[SCIM_CORE_USER_SCHEMA],
|
||||
id=user_id,
|
||||
externalId=None,
|
||||
meta=ScimMetaResponse(
|
||||
resourceType="User",
|
||||
created=created,
|
||||
lastModified=actual.meta.lastModified,
|
||||
),
|
||||
userName=user_name,
|
||||
displayName=user_name,
|
||||
active=active,
|
||||
name=name,
|
||||
emails=emails,
|
||||
groups=[],
|
||||
entitlements=None,
|
||||
roles=None,
|
||||
)
|
||||
assert actual == expected, response.text
|
||||
return actual
|
||||
|
||||
|
||||
def assert_serving(gateway: Gateway, model: str, key: str, status: int, error_type: str = "auth_error") -> None:
|
||||
response: Final = eventually(
|
||||
lambda: gateway.request(
|
||||
"POST", "/v1/chat/completions",
|
||||
{"model": model, "messages": [{"role": "user", "content": "warmed policy control"}]}, key=key,
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{"model": model, "messages": [{"role": "user", "content": "warmed policy control"}]},
|
||||
key=key,
|
||||
),
|
||||
lambda value: value.status_code == status,
|
||||
seconds=3,
|
||||
|
|
@ -117,9 +220,12 @@ def test_scim_deactivation_blocks_null_and_false_keys_but_preserves_other_owners
|
|||
assert_serving(gateway, model, token, 200)
|
||||
for active in (False, True):
|
||||
response: Final = gateway.request(
|
||||
"PATCH", f"/scim/v2/Users/{user}",
|
||||
{"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": [{"op": "replace", "path": "active", "value": active}]},
|
||||
"PATCH",
|
||||
f"/scim/v2/Users/{user}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": [{"op": "replace", "path": "active", "value": active}],
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
for token in (null_key, false_key):
|
||||
|
|
@ -135,6 +241,435 @@ def test_scim_deactivation_blocks_null_and_false_keys_but_preserves_other_owners
|
|||
assert_serving(gateway, model, token, 200)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"deactivate_operations,reactivate_operations",
|
||||
[
|
||||
pytest.param(
|
||||
[{"op": "replace", "value": {"active": False}}],
|
||||
[{"op": "replace", "value": {"active": True}}],
|
||||
id="okta_pathless",
|
||||
),
|
||||
pytest.param(
|
||||
[{"op": "Replace", "path": "active", "value": "False"}],
|
||||
[{"op": "Replace", "path": "active", "value": "True"}],
|
||||
id="entra_string_boolean",
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"warm_before_deactivation",
|
||||
[True, False],
|
||||
ids=["warmed", "first_used_while_deactivated"],
|
||||
)
|
||||
def test_scim_idp_deactivation_spelling_blocks_and_restores_the_users_key(
|
||||
gateway: Gateway,
|
||||
peer: Gateway,
|
||||
deactivate_operations: list[dict[str, JsonValue]],
|
||||
reactivate_operations: list[dict[str, JsonValue]],
|
||||
warm_before_deactivation: bool,
|
||||
) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
model: Final = scenario.model()
|
||||
user: Final = scenario.user(user_role="internal_user")
|
||||
key: Final = scenario.key(user_id=user, models=[model])
|
||||
original_name: Final = ScimNameResponse(
|
||||
givenName="Unknown User",
|
||||
familyName="Unknown Family Name",
|
||||
)
|
||||
initial_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert initial_response.status_code == 200, initial_response.text
|
||||
initial_user: Final = _assert_scim_user_resource(
|
||||
initial_response,
|
||||
user,
|
||||
True,
|
||||
None,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
created: Final = initial_user.meta.created
|
||||
if warm_before_deactivation:
|
||||
for worker in (gateway, peer):
|
||||
assert_serving(worker, model, key, 200)
|
||||
|
||||
deactivated: Final = gateway.request(
|
||||
"PATCH",
|
||||
f"/scim/v2/Users/{user}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": deactivate_operations,
|
||||
},
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert deactivated.status_code == 200, deactivated.text
|
||||
_assert_scim_user_resource(
|
||||
deactivated,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
deactivated_readback_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert deactivated_readback_response.status_code == 200, deactivated_readback_response.text
|
||||
_assert_scim_user_resource(
|
||||
deactivated_readback_response,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
user_rows: Final = read_rows(
|
||||
'SELECT metadata FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user,),
|
||||
)
|
||||
user_metadata: Final = object_value(user_rows[0]["metadata"])
|
||||
assert user_metadata.get("scim_active") is False and "active" not in user_metadata, user_metadata
|
||||
key_rows: Final = read_rows(
|
||||
'SELECT blocked, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
|
||||
(sha256(key.encode()).hexdigest(),),
|
||||
)
|
||||
assert key_rows[0]["blocked"] is True, key_rows
|
||||
assert object_value(key_rows[0]["metadata"]).get("scim_blocked") is True, key_rows
|
||||
for worker in (gateway, peer):
|
||||
assert_serving(worker, model, key, 401)
|
||||
|
||||
reactivated: Final = gateway.request(
|
||||
"PATCH",
|
||||
f"/scim/v2/Users/{user}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": reactivate_operations,
|
||||
},
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert reactivated.status_code == 200, reactivated.text
|
||||
_assert_scim_user_resource(
|
||||
reactivated,
|
||||
user,
|
||||
True,
|
||||
created,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
reactivated_readback_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert reactivated_readback_response.status_code == 200, reactivated_readback_response.text
|
||||
_assert_scim_user_resource(
|
||||
reactivated_readback_response,
|
||||
user,
|
||||
True,
|
||||
created,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
reactivated_user_rows: Final = read_rows(
|
||||
'SELECT metadata FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user,),
|
||||
)
|
||||
reactivated_metadata: Final = object_value(reactivated_user_rows[0]["metadata"])
|
||||
assert reactivated_metadata.get("scim_active") is True, reactivated_user_rows
|
||||
assert "active" not in reactivated_metadata, reactivated_user_rows
|
||||
restored_key_rows: Final = read_rows(
|
||||
'SELECT blocked, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
|
||||
(sha256(key.encode()).hexdigest(),),
|
||||
)
|
||||
assert restored_key_rows[0]["blocked"] is False, restored_key_rows
|
||||
assert "scim_blocked" not in object_value(restored_key_rows[0]["metadata"]), restored_key_rows
|
||||
for worker in (gateway, peer):
|
||||
assert_serving(worker, model, key, 200)
|
||||
|
||||
|
||||
def test_scim_put_without_active_keeps_a_deactivated_user_blocked(gateway: Gateway, peer: Gateway) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
model: Final = scenario.model()
|
||||
user: Final = scenario.user(user_role="internal_user")
|
||||
key: Final = scenario.key(user_id=user, models=[model])
|
||||
original_name: Final = ScimNameResponse(
|
||||
givenName="Unknown User",
|
||||
familyName="Unknown Family Name",
|
||||
)
|
||||
initial_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert initial_response.status_code == 200, initial_response.text
|
||||
initial_user: Final = _assert_scim_user_resource(
|
||||
initial_response,
|
||||
user,
|
||||
True,
|
||||
None,
|
||||
"Unknown Display Name",
|
||||
original_name,
|
||||
[],
|
||||
)
|
||||
created: Final = initial_user.meta.created
|
||||
email: Final = f"{user}@example.com"
|
||||
resource: Final[dict[str, JsonValue]] = {
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": user,
|
||||
"name": {"givenName": "Inactive", "familyName": "User"},
|
||||
"emails": [{"value": email, "primary": True, "type": "work"}],
|
||||
"active": False,
|
||||
}
|
||||
deactivated: Final = gateway.request(
|
||||
"PUT",
|
||||
f"/scim/v2/Users/{user}",
|
||||
resource,
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert deactivated.status_code == 200, deactivated.text
|
||||
deactivated_name: Final = ScimNameResponse(givenName="Inactive", familyName="User")
|
||||
deactivated_emails: Final = [ScimEmailResponse(value=email, type=None, primary=True)]
|
||||
_assert_scim_user_resource(
|
||||
deactivated,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
email,
|
||||
deactivated_name,
|
||||
deactivated_emails,
|
||||
)
|
||||
deactivated_readback_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert deactivated_readback_response.status_code == 200, deactivated_readback_response.text
|
||||
_assert_scim_user_resource(
|
||||
deactivated_readback_response,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
email,
|
||||
deactivated_name,
|
||||
deactivated_emails,
|
||||
)
|
||||
blocked_key_rows: Final = read_rows(
|
||||
'SELECT blocked, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
|
||||
(sha256(key.encode()).hexdigest(),),
|
||||
)
|
||||
assert blocked_key_rows[0]["blocked"] is True, blocked_key_rows
|
||||
assert object_value(blocked_key_rows[0]["metadata"]).get("scim_blocked") is True, blocked_key_rows
|
||||
for worker in (gateway, peer):
|
||||
assert_serving(worker, model, key, 401)
|
||||
|
||||
without_active: Final[dict[str, JsonValue]] = {
|
||||
field: value for field, value in resource.items() if field != "active"
|
||||
}
|
||||
without_active["name"] = {"givenName": "Inactive", "familyName": "Updated"}
|
||||
updated: Final = gateway.request(
|
||||
"PUT",
|
||||
f"/scim/v2/Users/{user}",
|
||||
without_active,
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert updated.status_code == 200, updated.text
|
||||
updated_name: Final = ScimNameResponse(givenName="Inactive", familyName="Updated")
|
||||
_assert_scim_user_resource(
|
||||
updated,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
email,
|
||||
updated_name,
|
||||
deactivated_emails,
|
||||
)
|
||||
updated_readback_response: Final = gateway.request("GET", f"/scim/v2/Users/{user}")
|
||||
assert updated_readback_response.status_code == 200, updated_readback_response.text
|
||||
_assert_scim_user_resource(
|
||||
updated_readback_response,
|
||||
user,
|
||||
False,
|
||||
created,
|
||||
email,
|
||||
updated_name,
|
||||
deactivated_emails,
|
||||
)
|
||||
still_blocked_rows: Final = read_rows(
|
||||
'SELECT blocked, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
|
||||
(sha256(key.encode()).hexdigest(),),
|
||||
)
|
||||
assert still_blocked_rows[0]["blocked"] is True, still_blocked_rows
|
||||
assert object_value(still_blocked_rows[0]["metadata"]).get("scim_blocked") is True, still_blocked_rows
|
||||
for worker in (gateway, peer):
|
||||
assert_serving(worker, model, key, 401)
|
||||
|
||||
|
||||
def _expected_scim_write_denial(route: str, user_id: str, user_role: str) -> dict[str, JsonValue]:
|
||||
masked_user_id: Final = f"{user_id[:6]}{'*' * (len(user_id) - 8)}{user_id[-2:]}"
|
||||
return {
|
||||
"error": {
|
||||
"message": (
|
||||
"Authentication Error, Only proxy admin can be used to generate, delete, update info for new "
|
||||
f"keys/users/teams. Route={route}. Your role={user_role}. Your user_id={masked_user_id}"
|
||||
),
|
||||
"type": "auth_error",
|
||||
"param": "None",
|
||||
"code": "401",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _delete_user_if_present(gateway: Gateway, user_id: str) -> None:
|
||||
if read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)):
|
||||
response: Final = gateway.request("POST", "/user/delete", {"user_ids": [user_id]})
|
||||
assert response.status_code == 200 and response.json() == 1, response.text
|
||||
assert read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)) == []
|
||||
|
||||
|
||||
def _delete_group_if_present(gateway: Gateway, group_id: str) -> None:
|
||||
if read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (group_id,)):
|
||||
response: Final = gateway.request(
|
||||
"DELETE",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert response.status_code == 204, response.text
|
||||
assert read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (group_id,)) == []
|
||||
|
||||
|
||||
def _delete_team_if_present(gateway: Gateway, team_id: str) -> None:
|
||||
if read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team_id,)):
|
||||
gateway.post("/team/delete", {"team_ids": [team_id]})
|
||||
assert read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team_id,)) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"write",
|
||||
[
|
||||
"post_user",
|
||||
"put_user",
|
||||
"patch_user",
|
||||
"delete_user",
|
||||
"post_group",
|
||||
"put_group",
|
||||
"patch_group",
|
||||
"delete_group",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("team_admin", [False, True], ids=["internal_user", "team_admin"])
|
||||
def test_non_admin_keys_cannot_write_scim_users_or_groups(gateway: Gateway, team_admin: bool, write: str) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
caller: Final = scenario.user(user_role="internal_user")
|
||||
created_team: Final = gateway.post(
|
||||
"/team/new",
|
||||
{
|
||||
"team_alias": f"integration-{uuid.uuid4().hex}",
|
||||
"members_with_roles": [{"user_id": caller, "role": "admin"}] if team_admin else [],
|
||||
},
|
||||
)
|
||||
team: Final = string_value(created_team["team_id"])
|
||||
scenario.cleanups.callback(_delete_team_if_present, gateway, team)
|
||||
key: Final = scenario.key(user_id=caller, team_id=team if team_admin else None)
|
||||
user_before: Final = read_rows(
|
||||
'SELECT user_role, metadata, teams FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(caller,),
|
||||
)
|
||||
caller_role: Final = string_value(user_before[0]["user_role"])
|
||||
team_before: Final = read_rows(
|
||||
'SELECT team_alias, members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s',
|
||||
(team,),
|
||||
)
|
||||
new_user: Final = f"scim-denied-{uuid.uuid4().hex}"
|
||||
scenario.cleanups.callback(_delete_user_if_present, gateway, new_user)
|
||||
new_group: Final = str(uuid.uuid4())
|
||||
scenario.cleanups.callback(_delete_group_if_present, gateway, new_group)
|
||||
request_details: Final = {
|
||||
"post_user": (
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": new_user,
|
||||
"name": {"givenName": "Denied", "familyName": "Create"},
|
||||
"emails": [{"value": f"{new_user}@example.com", "primary": True, "type": "work"}],
|
||||
"active": True,
|
||||
"roles": [{"value": "proxy_admin"}],
|
||||
},
|
||||
),
|
||||
"put_user": (
|
||||
"PUT",
|
||||
f"/scim/v2/Users/{caller}",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": caller,
|
||||
"name": {"givenName": "Denied", "familyName": "Update"},
|
||||
"emails": [{"value": f"{caller}@example.com", "primary": True, "type": "work"}],
|
||||
"active": False,
|
||||
"roles": [{"value": "proxy_admin"}],
|
||||
},
|
||||
),
|
||||
"patch_user": (
|
||||
"PATCH",
|
||||
f"/scim/v2/Users/{caller}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": [
|
||||
{"op": "replace", "path": "roles", "value": [{"value": "proxy_admin"}]},
|
||||
{"op": "replace", "path": "active", "value": False},
|
||||
],
|
||||
},
|
||||
),
|
||||
"delete_user": ("DELETE", f"/scim/v2/Users/{caller}", None),
|
||||
"post_group": (
|
||||
"POST",
|
||||
"/scim/v2/Groups",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
"id": new_group,
|
||||
"displayName": f"denied-{new_group}",
|
||||
"members": [{"value": caller}],
|
||||
},
|
||||
),
|
||||
"put_group": (
|
||||
"PUT",
|
||||
f"/scim/v2/Groups/{team}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
"displayName": f"updated-{team}",
|
||||
"members": [{"value": caller}],
|
||||
},
|
||||
),
|
||||
"patch_group": (
|
||||
"PATCH",
|
||||
f"/scim/v2/Groups/{team}",
|
||||
{
|
||||
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
"Operations": [{"op": "add", "path": "members", "value": [{"value": caller}]}],
|
||||
},
|
||||
),
|
||||
"delete_group": ("DELETE", f"/scim/v2/Groups/{team}", None),
|
||||
}
|
||||
method, route, body = request_details[write]
|
||||
response: Final = gateway.request(
|
||||
method,
|
||||
route,
|
||||
body,
|
||||
key=key,
|
||||
headers={"Content-Type": "application/scim+json"},
|
||||
)
|
||||
assert response.status_code == 401, response.text
|
||||
assert object_value(response.json()) == _expected_scim_write_denial(route, caller, caller_role), response.text
|
||||
user_after: Final = read_rows(
|
||||
'SELECT user_role, metadata, teams FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(caller,),
|
||||
)
|
||||
team_after: Final = read_rows(
|
||||
'SELECT team_alias, members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s',
|
||||
(team,),
|
||||
)
|
||||
new_user_rows: Final = read_rows(
|
||||
'SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(new_user,),
|
||||
)
|
||||
new_group_rows: Final = read_rows(
|
||||
'SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s',
|
||||
(new_group,),
|
||||
)
|
||||
assert user_after == user_before, (user_after, response.text)
|
||||
assert team_after == team_before, (team_after, response.text)
|
||||
assert new_user_rows == [], (new_user_rows, response.text)
|
||||
assert new_group_rows == [], (new_group_rows, response.text)
|
||||
|
||||
|
||||
@pytest.mark.covers("mgmt.team.member_update.demoted_role_cannot_write")
|
||||
def test_warmed_team_role_demotion_prevents_later_management_writes(gateway: Gateway) -> None:
|
||||
with gateway.scenario() as scenario, team_admin_permissions(gateway, ["tpm_limit"]):
|
||||
|
|
@ -160,22 +695,24 @@ def test_warmed_team_role_demotion_prevents_later_management_writes(gateway: Gat
|
|||
"POST", "/team/update", {"team_id": control_team, "tpm_limit": 7000}, key=caller
|
||||
)
|
||||
assert unrelated.status_code == 403, unrelated.text
|
||||
assert read_rows(
|
||||
'SELECT tpm_limit FROM "LiteLLM_TeamTable" WHERE team_id = %s', (control_team,)
|
||||
) == unrelated_before
|
||||
assert (
|
||||
read_rows('SELECT tpm_limit FROM "LiteLLM_TeamTable" WHERE team_id = %s', (control_team,))
|
||||
== unrelated_before
|
||||
)
|
||||
gateway.post("/team/member_update", {"team_id": team, "user_id": user, "role": "user"})
|
||||
for target in (team, control_team):
|
||||
before: Final = read_rows('SELECT tpm_limit FROM "LiteLLM_TeamTable" WHERE team_id = %s', (target,))
|
||||
denied: Final = gateway.request(
|
||||
"POST", "/team/update", {"team_id": target, "tpm_limit": 9000}, key=caller
|
||||
)
|
||||
denied: Final = gateway.request("POST", "/team/update", {"team_id": target, "tpm_limit": 9000}, key=caller)
|
||||
assert denied.status_code == 403, denied.text
|
||||
after: Final = read_rows('SELECT tpm_limit FROM "LiteLLM_TeamTable" WHERE team_id = %s', (target,))
|
||||
assert after == before
|
||||
roster: Final = read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,))
|
||||
members: Final = roster[0]["members_with_roles"]
|
||||
assert isinstance(members, list)
|
||||
assert next(object_value(member)["role"] for member in members if object_value(member)["user_id"] == user) == "user"
|
||||
assert (
|
||||
next(object_value(member)["role"] for member in members if object_value(member)["user_id"] == user)
|
||||
== "user"
|
||||
)
|
||||
assert_serving(gateway, model, caller, 200)
|
||||
|
||||
|
||||
|
|
|
|||
776
tests/integration/management/test_scim_admin_group.py
Normal file
776
tests/integration/management/test_scim_admin_group.py
Normal file
|
|
@ -0,0 +1,776 @@
|
|||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final, Literal, Mapping
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import yaml
|
||||
from integration._support.client import Gateway, Scenario, eventually, object_value, string_value
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.process import owned_proxy_process
|
||||
from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
|
||||
|
||||
SCIM_HEADERS: Final = {"Content-Type": "application/scim+json"}
|
||||
SCIM_CORE_USER_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:User"
|
||||
SCIM_CORE_GROUP_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:Group"
|
||||
SCIM_PATCH_SCHEMA: Final = "urn:ietf:params:scim:api:messages:2.0:PatchOp"
|
||||
SCIM_LIST_RESPONSE_SCHEMA: Final = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
|
||||
SCIM_CONFIG: Final = TypeAdapter(dict[str, JsonValue])
|
||||
|
||||
|
||||
class ScimGroupMeta(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
resourceType: str
|
||||
created: str
|
||||
lastModified: str
|
||||
|
||||
|
||||
class ScimGroupMember(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
value: str
|
||||
display: str
|
||||
type: str
|
||||
|
||||
|
||||
class ScimGroupResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
schemas: list[str]
|
||||
id: str
|
||||
externalId: str | None = None
|
||||
meta: ScimGroupMeta
|
||||
displayName: str
|
||||
members: list[ScimGroupMember]
|
||||
|
||||
|
||||
class ScimGroupListResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
schemas: list[str]
|
||||
totalResults: int
|
||||
startIndex: int
|
||||
itemsPerPage: int
|
||||
Resources: list[ScimGroupResponse]
|
||||
|
||||
|
||||
class ScimUserResponse(BaseModel):
|
||||
schemas: list[str]
|
||||
id: str
|
||||
userName: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ScimAdminGroupFixture:
|
||||
gateway: Gateway
|
||||
scenario: Scenario
|
||||
group_id: str
|
||||
group_name: str
|
||||
created_group: ScimGroupResponse
|
||||
user_id: str
|
||||
member_display: str
|
||||
user_key: str
|
||||
other_user: str
|
||||
default_role: str
|
||||
|
||||
|
||||
def _owned_scim_config(path: Path, settings: Mapping[str, JsonValue]) -> Path:
|
||||
config: Final = object_value(
|
||||
SCIM_CONFIG.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
|
||||
)
|
||||
litellm_settings: Final = object_value(config["litellm_settings"])
|
||||
path.write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
**config,
|
||||
"litellm_settings": {**litellm_settings, **settings},
|
||||
}
|
||||
)
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
def _delete_user_if_present(gateway: Gateway, user_id: str) -> None:
|
||||
if read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)):
|
||||
deleted: Final = gateway.request("POST", "/user/delete", {"user_ids": [user_id]})
|
||||
assert deleted.status_code == 200 and deleted.json() == 1, deleted.text
|
||||
assert read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)) == []
|
||||
|
||||
|
||||
def _delete_group_if_present(gateway: Gateway, group_id: str) -> None:
|
||||
if read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (group_id,)):
|
||||
deleted: Final = gateway.request(
|
||||
"DELETE",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
assert deleted.status_code == 204, deleted.text
|
||||
assert read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (group_id,)) == []
|
||||
|
||||
|
||||
def _user_role(gateway: Gateway, user_id: str) -> str:
|
||||
info: Final = object_value(gateway.get("/user/info", {"user_id": user_id})["user_info"])
|
||||
return string_value(info["user_role"])
|
||||
|
||||
|
||||
def _database_user_role(user_id: str) -> str:
|
||||
rows: Final = read_rows('SELECT user_role FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,))
|
||||
return string_value(rows[0]["user_role"])
|
||||
|
||||
|
||||
def _expected_scim_write_denial(route: str, user_id: str, user_role: str) -> dict[str, JsonValue]:
|
||||
masked_user_id: Final = f"{user_id[:6]}{'*' * (len(user_id) - 8)}{user_id[-2:]}"
|
||||
return {
|
||||
"error": {
|
||||
"message": (
|
||||
"Authentication Error, Only proxy admin can be used to generate, delete, update info for new "
|
||||
f"keys/users/teams. Route={route}. Your role={user_role}. Your user_id={masked_user_id}"
|
||||
),
|
||||
"type": "auth_error",
|
||||
"param": "None",
|
||||
"code": "401",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _assert_scim_group_resource(
|
||||
response: httpx.Response,
|
||||
group_id: str,
|
||||
display_name: str,
|
||||
members: list[ScimGroupMember],
|
||||
expected_created: str | None,
|
||||
) -> ScimGroupResponse:
|
||||
actual: Final = ScimGroupResponse.model_validate(response.json())
|
||||
created: Final = actual.meta.created if expected_created is None else expected_created
|
||||
assert isinstance(created, str), response.text
|
||||
assert isinstance(actual.meta.lastModified, str), response.text
|
||||
expected: Final = ScimGroupResponse(
|
||||
schemas=[SCIM_CORE_GROUP_SCHEMA],
|
||||
id=group_id,
|
||||
externalId=None,
|
||||
meta=ScimGroupMeta(
|
||||
resourceType="Group",
|
||||
created=created,
|
||||
lastModified=actual.meta.lastModified,
|
||||
),
|
||||
displayName=display_name,
|
||||
members=members,
|
||||
)
|
||||
assert actual == expected, response.text
|
||||
return actual
|
||||
|
||||
|
||||
def _assert_scim_group_resource_except_members(
|
||||
response: httpx.Response,
|
||||
group_id: str,
|
||||
display_name: str,
|
||||
) -> ScimGroupResponse:
|
||||
actual: Final = ScimGroupResponse.model_validate(response.json())
|
||||
expected: Final = ScimGroupResponse(
|
||||
schemas=[SCIM_CORE_GROUP_SCHEMA],
|
||||
id=group_id,
|
||||
externalId=None,
|
||||
meta=ScimGroupMeta(
|
||||
resourceType="Group",
|
||||
created=actual.meta.created,
|
||||
lastModified=actual.meta.lastModified,
|
||||
),
|
||||
displayName=display_name,
|
||||
members=actual.members,
|
||||
)
|
||||
assert actual.model_dump(exclude={"members"}) == expected.model_dump(exclude={"members"}), response.text
|
||||
return actual
|
||||
|
||||
|
||||
def _group_filter_response(gateway: Gateway, group_name: str) -> httpx.Response:
|
||||
return gateway.request(
|
||||
"GET",
|
||||
"/scim/v2/Groups",
|
||||
params={"filter": f'displayName eq "{group_name}"', "startIndex": "1", "count": "100"},
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
|
||||
|
||||
def _assert_scim_group_list_response(
|
||||
response: httpx.Response,
|
||||
group_id: str | None,
|
||||
group_name: str,
|
||||
members: list[ScimGroupMember],
|
||||
) -> ScimGroupListResponse:
|
||||
actual: Final = ScimGroupListResponse.model_validate(response.json())
|
||||
expected_results: Final = 0 if group_id is None else 1
|
||||
assert len(actual.Resources) == expected_results, response.text
|
||||
expected_resources: Final = (
|
||||
[]
|
||||
if group_id is None
|
||||
else [
|
||||
ScimGroupResponse(
|
||||
schemas=[SCIM_CORE_GROUP_SCHEMA],
|
||||
id=group_id,
|
||||
externalId=None,
|
||||
meta=ScimGroupMeta(
|
||||
resourceType="Group",
|
||||
created=actual.Resources[0].meta.created,
|
||||
lastModified=actual.Resources[0].meta.lastModified,
|
||||
),
|
||||
displayName=group_name,
|
||||
members=members,
|
||||
)
|
||||
]
|
||||
)
|
||||
expected: Final = ScimGroupListResponse(
|
||||
schemas=[SCIM_LIST_RESPONSE_SCHEMA],
|
||||
totalResults=expected_results,
|
||||
startIndex=1,
|
||||
itemsPerPage=expected_results,
|
||||
Resources=expected_resources,
|
||||
)
|
||||
assert actual == expected, response.text
|
||||
return actual
|
||||
|
||||
|
||||
def _group_membership_operations(
|
||||
user_id: str,
|
||||
spelling: Literal["okta", "entra"],
|
||||
) -> tuple[dict[str, JsonValue], dict[str, JsonValue]]:
|
||||
if spelling == "okta":
|
||||
return (
|
||||
{"op": "add", "path": "members", "value": [{"value": user_id}]},
|
||||
{"op": "remove", "path": f'members[value eq "{user_id}"]'},
|
||||
)
|
||||
return (
|
||||
{"op": "Add", "path": "members", "value": [{"value": user_id}]},
|
||||
{"op": "Remove", "path": "members", "value": [{"value": user_id}]},
|
||||
)
|
||||
|
||||
|
||||
def _change_group_membership(
|
||||
gateway: Gateway,
|
||||
group_id: str,
|
||||
group_name: str,
|
||||
user_id: str,
|
||||
member_display: str,
|
||||
action: Literal["add", "remove"],
|
||||
method: Literal["PATCH", "PUT"],
|
||||
spelling: Literal["okta", "entra"] = "okta",
|
||||
) -> httpx.Response:
|
||||
if method == "PUT":
|
||||
members: Final = [{"value": user_id, "display": member_display}] if action == "add" else []
|
||||
return gateway.request(
|
||||
"PUT",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
{
|
||||
"schemas": [SCIM_CORE_GROUP_SCHEMA],
|
||||
"id": group_id,
|
||||
"displayName": group_name,
|
||||
"members": members,
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
add_operation, remove_operation = _group_membership_operations(user_id, spelling)
|
||||
operation: Final = add_operation if action == "add" else remove_operation
|
||||
return gateway.request(
|
||||
"PATCH",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
{
|
||||
"schemas": [SCIM_PATCH_SCHEMA],
|
||||
"Operations": [operation],
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
|
||||
|
||||
def _generate_key_for_user(scenario: Scenario, user_key: str, target_user: str) -> httpx.Response:
|
||||
response: Final = scenario.gateway.request(
|
||||
"POST",
|
||||
"/key/generate",
|
||||
{"user_id": target_user},
|
||||
key=user_key,
|
||||
)
|
||||
if response.status_code == 200:
|
||||
generated_key: Final = string_value(object_value(response.json())["key"])
|
||||
scenario.cleanups.callback(scenario.delete_key, generated_key)
|
||||
return response
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _scim_admin_group_fixture(
|
||||
gateway: Gateway,
|
||||
tmp_path: Path,
|
||||
group_name: str,
|
||||
) -> Iterator[ScimAdminGroupFixture]:
|
||||
config: Final = _owned_scim_config(tmp_path / "scim-admin-group.yaml", {"scim_admin_group": group_name})
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config) as owned, owned.gateway.scenario() as scenario:
|
||||
group_id: Final = str(uuid.uuid4())
|
||||
unrelated_group_alias: Final = f"{group_name}-unrelated"
|
||||
unrelated_group_id: Final = scenario.team(team_alias=unrelated_group_alias)
|
||||
assert read_rows(
|
||||
'SELECT team_alias FROM "LiteLLM_TeamTable" WHERE team_id = %s',
|
||||
(unrelated_group_id,),
|
||||
) == [{"team_alias": unrelated_group_alias}]
|
||||
before_filter: Final = _group_filter_response(owned.gateway, group_name)
|
||||
assert before_filter.status_code == 200, before_filter.text
|
||||
_assert_scim_group_list_response(before_filter, None, group_name, [])
|
||||
group_response: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Groups",
|
||||
{
|
||||
"schemas": [SCIM_CORE_GROUP_SCHEMA],
|
||||
"id": group_id,
|
||||
"displayName": group_name,
|
||||
"members": [],
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
assert group_response.status_code == 201, group_response.text
|
||||
created_group: Final = _assert_scim_group_resource(
|
||||
group_response,
|
||||
group_id,
|
||||
group_name,
|
||||
[],
|
||||
None,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_group_if_present, owned.gateway, group_id)
|
||||
after_create_filter: Final = _group_filter_response(owned.gateway, group_name)
|
||||
assert after_create_filter.status_code == 200, after_create_filter.text
|
||||
_assert_scim_group_list_response(after_create_filter, group_id, group_name, [])
|
||||
user: Final = f"scim-admin-user-{uuid.uuid4().hex}"
|
||||
created_user: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": user,
|
||||
"name": {"givenName": "Group", "familyName": "Member"},
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, owned.gateway, user)
|
||||
assert created_user.status_code == 201, created_user.text
|
||||
created_user_response: Final = ScimUserResponse.model_validate(created_user.json())
|
||||
assert created_user_response.id == user, created_user.text
|
||||
default_role: Final = _database_user_role(user)
|
||||
assert _user_role(owned.gateway, user) == default_role
|
||||
member_rows: Final = read_rows(
|
||||
'SELECT user_email FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user,),
|
||||
)
|
||||
member_email: Final = member_rows[0]["user_email"]
|
||||
member_display: Final = member_email if isinstance(member_email, str) and member_email else user
|
||||
user_key: Final = scenario.key(user_id=user)
|
||||
other_user: Final = scenario.user(user_role="internal_user")
|
||||
yield ScimAdminGroupFixture(
|
||||
gateway=owned.gateway,
|
||||
scenario=scenario,
|
||||
group_id=group_id,
|
||||
group_name=group_name,
|
||||
created_group=created_group,
|
||||
user_id=user,
|
||||
member_display=member_display,
|
||||
user_key=user_key,
|
||||
other_user=other_user,
|
||||
default_role=default_role,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(300)
|
||||
@pytest.mark.parametrize("spelling", ["okta", "entra"], ids=["okta", "entra"])
|
||||
def test_scim_admin_group_promotes_and_demotes_group_members(
|
||||
gateway: Gateway,
|
||||
tmp_path: Path,
|
||||
spelling: Literal["okta", "entra"],
|
||||
) -> None:
|
||||
group_name: Final = f"scim-admin-{uuid.uuid4().hex}"
|
||||
with _scim_admin_group_fixture(gateway, tmp_path, group_name) as fixture:
|
||||
owned: Final = fixture
|
||||
scenario: Final = fixture.scenario
|
||||
group_id: Final = fixture.group_id
|
||||
created_group: Final = fixture.created_group
|
||||
user: Final = fixture.user_id
|
||||
member_display: Final = fixture.member_display
|
||||
user_key: Final = fixture.user_key
|
||||
other_user: Final = fixture.other_user
|
||||
default_role: Final = fixture.default_role
|
||||
initial_denial: Final = _generate_key_for_user(scenario, user_key, other_user)
|
||||
assert initial_denial.status_code == 401, initial_denial.text
|
||||
assert object_value(initial_denial.json()) == _expected_scim_write_denial(
|
||||
"/key/generate",
|
||||
user,
|
||||
default_role,
|
||||
), initial_denial.text
|
||||
|
||||
expected_member: Final = ScimGroupMember(value=user, display=member_display, type="User")
|
||||
added: Final = _change_group_membership(
|
||||
owned.gateway,
|
||||
group_id,
|
||||
group_name,
|
||||
user,
|
||||
member_display,
|
||||
"add",
|
||||
"PATCH",
|
||||
spelling,
|
||||
)
|
||||
assert added.status_code == 200, added.text
|
||||
after_add_filter: Final = _group_filter_response(owned.gateway, group_name)
|
||||
assert after_add_filter.status_code == 200, after_add_filter.text
|
||||
_assert_scim_group_list_response(after_add_filter, group_id, group_name, [expected_member])
|
||||
added_group: Final = _assert_scim_group_resource(
|
||||
added,
|
||||
group_id,
|
||||
group_name,
|
||||
[expected_member],
|
||||
created_group.meta.created,
|
||||
)
|
||||
assert added_group.members == [expected_member], added.text
|
||||
added_readback: Final = owned.gateway.request(
|
||||
"GET",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
assert added_readback.status_code == 200, added_readback.text
|
||||
_assert_scim_group_resource(
|
||||
added_readback,
|
||||
group_id,
|
||||
group_name,
|
||||
[expected_member],
|
||||
created_group.meta.created,
|
||||
)
|
||||
assert eventually(lambda: _user_role(owned.gateway, user), lambda role: role == "proxy_admin") == "proxy_admin"
|
||||
assert eventually(lambda: _database_user_role(user), lambda role: role == "proxy_admin") == "proxy_admin"
|
||||
allowed: Final = eventually(
|
||||
lambda: _generate_key_for_user(scenario, user_key, other_user),
|
||||
lambda response: response.status_code == 200,
|
||||
)
|
||||
assert allowed.status_code == 200, allowed.text
|
||||
|
||||
removed: Final = _change_group_membership(
|
||||
owned.gateway,
|
||||
group_id,
|
||||
group_name,
|
||||
user,
|
||||
member_display,
|
||||
"remove",
|
||||
"PATCH",
|
||||
spelling,
|
||||
)
|
||||
assert removed.status_code == 200, removed.text
|
||||
_assert_scim_group_resource(
|
||||
removed,
|
||||
group_id,
|
||||
group_name,
|
||||
[],
|
||||
created_group.meta.created,
|
||||
)
|
||||
removed_readback: Final = owned.gateway.request(
|
||||
"GET",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
assert removed_readback.status_code == 200, removed_readback.text
|
||||
_assert_scim_group_resource(
|
||||
removed_readback,
|
||||
group_id,
|
||||
group_name,
|
||||
[],
|
||||
created_group.meta.created,
|
||||
)
|
||||
assert eventually(lambda: _user_role(owned.gateway, user), lambda role: role == default_role) == default_role
|
||||
assert eventually(lambda: _database_user_role(user), lambda role: role == default_role) == default_role
|
||||
refused_again: Final = eventually(
|
||||
lambda: _generate_key_for_user(scenario, user_key, other_user),
|
||||
lambda response: response.status_code == 401,
|
||||
)
|
||||
assert refused_again.status_code == 401, refused_again.text
|
||||
assert object_value(refused_again.json()) == _expected_scim_write_denial(
|
||||
"/key/generate",
|
||||
user,
|
||||
default_role,
|
||||
), refused_again.text
|
||||
|
||||
second_user: Final = f"scim-admin-user-{uuid.uuid4().hex}"
|
||||
created_second_user: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": second_user,
|
||||
"name": {"givenName": "Display", "familyName": "Member"},
|
||||
"groups": [{"value": group_id, "display": group_name}],
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, owned.gateway, second_user)
|
||||
assert created_second_user.status_code == 201, created_second_user.text
|
||||
created_second_user_response: Final = ScimUserResponse.model_validate(created_second_user.json())
|
||||
assert created_second_user_response.id == second_user, created_second_user.text
|
||||
second_key: Final = scenario.key(user_id=second_user)
|
||||
assert eventually(lambda: _user_role(owned.gateway, second_user), lambda role: role == "proxy_admin") == (
|
||||
"proxy_admin"
|
||||
)
|
||||
assert eventually(lambda: _database_user_role(second_user), lambda role: role == "proxy_admin") == (
|
||||
"proxy_admin"
|
||||
)
|
||||
second_allowed: Final = eventually(
|
||||
lambda: _generate_key_for_user(scenario, second_key, other_user),
|
||||
lambda response: response.status_code == 200,
|
||||
)
|
||||
assert second_allowed.status_code == 200, second_allowed.text
|
||||
|
||||
deleted_group: Final = owned.gateway.request(
|
||||
"DELETE",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
assert deleted_group.status_code == 204, deleted_group.text
|
||||
assert eventually(lambda: _user_role(owned.gateway, second_user), lambda role: role == default_role) == (
|
||||
default_role
|
||||
)
|
||||
assert eventually(lambda: _database_user_role(second_user), lambda role: role == default_role) == default_role
|
||||
|
||||
|
||||
@pytest.mark.timeout(300)
|
||||
def test_scim_admin_group_put_members_promotes_and_demotes(gateway: Gateway, tmp_path: Path) -> None:
|
||||
group_name: Final = f"scim-admin-put-{uuid.uuid4().hex}"
|
||||
with _scim_admin_group_fixture(gateway, tmp_path, group_name) as fixture:
|
||||
expected_member: Final = ScimGroupMember(
|
||||
value=fixture.user_id,
|
||||
display=fixture.member_display,
|
||||
type="User",
|
||||
)
|
||||
initial_denial: Final = _generate_key_for_user(
|
||||
fixture.scenario,
|
||||
fixture.user_key,
|
||||
fixture.other_user,
|
||||
)
|
||||
assert initial_denial.status_code == 401, initial_denial.text
|
||||
assert object_value(initial_denial.json()) == _expected_scim_write_denial(
|
||||
"/key/generate",
|
||||
fixture.user_id,
|
||||
fixture.default_role,
|
||||
), initial_denial.text
|
||||
|
||||
added: Final = _change_group_membership(
|
||||
fixture.gateway,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
fixture.user_id,
|
||||
fixture.member_display,
|
||||
"add",
|
||||
"PUT",
|
||||
)
|
||||
assert added.status_code == 200, added.text
|
||||
_assert_scim_group_resource_except_members(
|
||||
added,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
)
|
||||
added_readback: Final = fixture.gateway.request(
|
||||
"GET",
|
||||
f"/scim/v2/Groups/{fixture.group_id}",
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
assert added_readback.status_code == 200, added_readback.text
|
||||
_assert_scim_group_resource(
|
||||
added_readback,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
[expected_member],
|
||||
fixture.created_group.meta.created,
|
||||
)
|
||||
added_filter: Final = _group_filter_response(fixture.gateway, fixture.group_name)
|
||||
assert added_filter.status_code == 200, added_filter.text
|
||||
_assert_scim_group_list_response(
|
||||
added_filter,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
[expected_member],
|
||||
)
|
||||
assert eventually(
|
||||
lambda: _user_role(fixture.gateway, fixture.user_id),
|
||||
lambda role: role == "proxy_admin",
|
||||
) == "proxy_admin"
|
||||
assert eventually(
|
||||
lambda: _database_user_role(fixture.user_id),
|
||||
lambda role: role == "proxy_admin",
|
||||
) == "proxy_admin"
|
||||
allowed: Final = eventually(
|
||||
lambda: _generate_key_for_user(fixture.scenario, fixture.user_key, fixture.other_user),
|
||||
lambda response: response.status_code == 200,
|
||||
)
|
||||
assert allowed.status_code == 200, allowed.text
|
||||
|
||||
removed: Final = _change_group_membership(
|
||||
fixture.gateway,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
fixture.user_id,
|
||||
fixture.member_display,
|
||||
"remove",
|
||||
"PUT",
|
||||
)
|
||||
assert removed.status_code == 200, removed.text
|
||||
_assert_scim_group_resource_except_members(
|
||||
removed,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
)
|
||||
removed_readback: Final = fixture.gateway.request(
|
||||
"GET",
|
||||
f"/scim/v2/Groups/{fixture.group_id}",
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
assert removed_readback.status_code == 200, removed_readback.text
|
||||
_assert_scim_group_resource(
|
||||
removed_readback,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
[],
|
||||
fixture.created_group.meta.created,
|
||||
)
|
||||
removed_filter: Final = _group_filter_response(fixture.gateway, fixture.group_name)
|
||||
assert removed_filter.status_code == 200, removed_filter.text
|
||||
_assert_scim_group_list_response(
|
||||
removed_filter,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
[],
|
||||
)
|
||||
assert eventually(
|
||||
lambda: _user_role(fixture.gateway, fixture.user_id),
|
||||
lambda role: role == fixture.default_role,
|
||||
) == fixture.default_role
|
||||
assert eventually(
|
||||
lambda: _database_user_role(fixture.user_id),
|
||||
lambda role: role == fixture.default_role,
|
||||
) == fixture.default_role
|
||||
refused: Final = eventually(
|
||||
lambda: _generate_key_for_user(fixture.scenario, fixture.user_key, fixture.other_user),
|
||||
lambda response: response.status_code == 401,
|
||||
)
|
||||
assert refused.status_code == 401, refused.text
|
||||
assert object_value(refused.json()) == _expected_scim_write_denial(
|
||||
"/key/generate",
|
||||
fixture.user_id,
|
||||
fixture.default_role,
|
||||
), refused.text
|
||||
|
||||
|
||||
@pytest.mark.timeout(300)
|
||||
def test_scim_group_put_returns_the_updated_members(gateway: Gateway, tmp_path: Path) -> None:
|
||||
pytest.skip(
|
||||
"BUG: PUT /scim/v2/Groups/{id} returns the members list from before the update, so the 200 body omits "
|
||||
"the members it just added"
|
||||
)
|
||||
group_name: Final = f"scim-admin-put-response-{uuid.uuid4().hex}"
|
||||
with _scim_admin_group_fixture(gateway, tmp_path, group_name) as fixture:
|
||||
expected_member: Final = ScimGroupMember(
|
||||
value=fixture.user_id,
|
||||
display=fixture.member_display,
|
||||
type="User",
|
||||
)
|
||||
added: Final = _change_group_membership(
|
||||
fixture.gateway,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
fixture.user_id,
|
||||
fixture.member_display,
|
||||
"add",
|
||||
"PUT",
|
||||
)
|
||||
assert added.status_code == 200, added.text
|
||||
_assert_scim_group_resource(
|
||||
added,
|
||||
fixture.group_id,
|
||||
fixture.group_name,
|
||||
[expected_member],
|
||||
fixture.created_group.meta.created,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(300)
|
||||
def test_scim_admin_group_delete_revokes_the_members_admin_key(gateway: Gateway, tmp_path: Path) -> None:
|
||||
pytest.skip(
|
||||
"BUG: DELETE /scim/v2/Groups/{id} demotes the admin group's members in the DB but their existing keys keep proxy_admin access on /key/generate"
|
||||
)
|
||||
group_name: Final = f"scim-admin-delete-{uuid.uuid4().hex}"
|
||||
config: Final = _owned_scim_config(tmp_path / "scim-admin-group-delete.yaml", {"scim_admin_group": group_name})
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config) as owned, owned.gateway.scenario() as scenario:
|
||||
group_id: Final = str(uuid.uuid4())
|
||||
group_response: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Groups",
|
||||
{
|
||||
"schemas": [SCIM_CORE_GROUP_SCHEMA],
|
||||
"id": group_id,
|
||||
"displayName": group_name,
|
||||
"members": [],
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_group_if_present, owned.gateway, group_id)
|
||||
assert group_response.status_code == 201, group_response.text
|
||||
created_group: Final = ScimGroupResponse.model_validate(group_response.json())
|
||||
assert created_group.id == group_id and created_group.displayName == group_name, group_response.text
|
||||
|
||||
other_user: Final = f"scim-admin-target-{uuid.uuid4().hex}"
|
||||
created_other_user: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": other_user,
|
||||
"name": {"givenName": "Key", "familyName": "Target"},
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, owned.gateway, other_user)
|
||||
assert created_other_user.status_code == 201, created_other_user.text
|
||||
other_user_response: Final = ScimUserResponse.model_validate(created_other_user.json())
|
||||
assert other_user_response.id == other_user, created_other_user.text
|
||||
default_role: Final = _database_user_role(other_user)
|
||||
|
||||
user: Final = f"scim-admin-delete-user-{uuid.uuid4().hex}"
|
||||
created_user: Final = owned.gateway.request(
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": user,
|
||||
"name": {"givenName": "Delete", "familyName": "Member"},
|
||||
"groups": [{"value": group_id, "display": group_name}],
|
||||
},
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, owned.gateway, user)
|
||||
assert created_user.status_code == 201, created_user.text
|
||||
created_user_response: Final = ScimUserResponse.model_validate(created_user.json())
|
||||
assert created_user_response.id == user, created_user.text
|
||||
user_key: Final = scenario.key(user_id=user)
|
||||
assert eventually(lambda: _user_role(owned.gateway, user), lambda role: role == "proxy_admin") == "proxy_admin"
|
||||
assert eventually(lambda: _database_user_role(user), lambda role: role == "proxy_admin") == "proxy_admin"
|
||||
allowed: Final = eventually(
|
||||
lambda: _generate_key_for_user(scenario, user_key, other_user),
|
||||
lambda response: response.status_code == 200,
|
||||
)
|
||||
assert allowed.status_code == 200, allowed.text
|
||||
|
||||
deleted_group: Final = owned.gateway.request(
|
||||
"DELETE",
|
||||
f"/scim/v2/Groups/{group_id}",
|
||||
headers=SCIM_HEADERS,
|
||||
)
|
||||
assert deleted_group.status_code == 204, deleted_group.text
|
||||
assert eventually(lambda: _user_role(owned.gateway, user), lambda role: role == default_role) == default_role
|
||||
assert eventually(lambda: _database_user_role(user), lambda role: role == default_role) == default_role
|
||||
revoked: Final = eventually(
|
||||
lambda: _generate_key_for_user(scenario, user_key, other_user),
|
||||
lambda response: response.status_code == 401,
|
||||
return_last_on_timeout=True,
|
||||
)
|
||||
assert revoked.status_code == 401, (
|
||||
revoked.text if revoked.status_code != 200 else "Key generation returned 200; response body redacted"
|
||||
)
|
||||
595
tests/integration/management/test_scim_user_lifecycle.py
Normal file
595
tests/integration/management/test_scim_user_lifecycle.py
Normal file
|
|
@ -0,0 +1,595 @@
|
|||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from hashlib import sha256
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from integration._support.client import Gateway, Scenario, object_value, string_value
|
||||
from integration._support.database import read_rows
|
||||
from pydantic import BaseModel, ConfigDict, Field, JsonValue
|
||||
|
||||
SCIM_HEADERS: Final = {"Content-Type": "application/scim+json"}
|
||||
SCIM_CORE_USER_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:User"
|
||||
SCIM_ENTERPRISE_USER_SCHEMA: Final = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
|
||||
SCIM_LIST_RESPONSE_SCHEMA: Final = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
|
||||
|
||||
|
||||
class ScimResponseModel(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid", populate_by_name=True)
|
||||
|
||||
|
||||
class ScimName(ScimResponseModel):
|
||||
familyName: str | None = None
|
||||
givenName: str | None = None
|
||||
formatted: str | None = None
|
||||
middleName: str | None = None
|
||||
honorificPrefix: str | None = None
|
||||
honorificSuffix: str | None = None
|
||||
|
||||
|
||||
class ScimEmail(ScimResponseModel):
|
||||
value: str
|
||||
type: str | None = None
|
||||
primary: bool | None = None
|
||||
|
||||
|
||||
class ScimGroupRef(ScimResponseModel):
|
||||
value: str
|
||||
display: str | None = None
|
||||
type: str | None = None
|
||||
|
||||
|
||||
class ScimUserMeta(ScimResponseModel):
|
||||
resourceType: str
|
||||
created: str | None = None
|
||||
lastModified: str | None = None
|
||||
|
||||
|
||||
class ScimManager(ScimResponseModel):
|
||||
value: str | None = None
|
||||
displayName: str | None = None
|
||||
ref: str | None = Field(default=None, alias="$ref")
|
||||
|
||||
|
||||
class ScimEnterpriseUser(ScimResponseModel):
|
||||
employeeNumber: str | None = None
|
||||
costCenter: str | None = None
|
||||
organization: str | None = None
|
||||
division: str | None = None
|
||||
department: str | None = None
|
||||
manager: ScimManager | None = None
|
||||
|
||||
|
||||
class ScimMultiValuedAttribute(ScimResponseModel):
|
||||
value: str
|
||||
display: str | None = None
|
||||
type: str | None = None
|
||||
primary: bool | None = None
|
||||
|
||||
|
||||
class ScimUserResponse(ScimResponseModel):
|
||||
schemas: list[str]
|
||||
id: str
|
||||
externalId: str | None = None
|
||||
userName: str
|
||||
displayName: str | None = None
|
||||
name: ScimName | None = None
|
||||
emails: list[ScimEmail] | None = None
|
||||
groups: list[ScimGroupRef] | None = None
|
||||
entitlements: list[ScimMultiValuedAttribute] | None = None
|
||||
roles: list[ScimMultiValuedAttribute] | None = None
|
||||
active: bool
|
||||
meta: ScimUserMeta
|
||||
enterprise: ScimEnterpriseUser | None = Field(default=None, alias=SCIM_ENTERPRISE_USER_SCHEMA)
|
||||
|
||||
|
||||
class ScimListResponse(ScimResponseModel):
|
||||
schemas: list[str]
|
||||
totalResults: int
|
||||
startIndex: int
|
||||
itemsPerPage: int
|
||||
Resources: list[ScimUserResponse]
|
||||
|
||||
|
||||
def _utc_datetime(value: str) -> datetime:
|
||||
parsed: Final = datetime.fromisoformat(value)
|
||||
return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def _scim_request(
|
||||
gateway: Gateway,
|
||||
method: str,
|
||||
path: str,
|
||||
body: dict[str, JsonValue] | None = None,
|
||||
) -> httpx.Response:
|
||||
return gateway.request(method, path, body, headers=SCIM_HEADERS)
|
||||
|
||||
|
||||
def _delete_user_if_present(gateway: Gateway, user_id: str) -> None:
|
||||
if read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)):
|
||||
deleted: Final = gateway.request("POST", "/user/delete", {"user_ids": [user_id]})
|
||||
assert deleted.status_code == 200 and deleted.json() == 1, deleted.text
|
||||
assert read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,)) == []
|
||||
|
||||
|
||||
def _team_members(gateway: Gateway, team_id: str) -> list[dict[str, JsonValue]]:
|
||||
team_info: Final = object_value(gateway.get("/team/info", {"team_id": team_id})["team_info"])
|
||||
members: Final = team_info.get("members_with_roles") or []
|
||||
assert isinstance(members, list), members
|
||||
return [object_value(member) for member in members]
|
||||
|
||||
|
||||
def _has_team_member(gateway: Gateway, team_id: str, user_id: str) -> bool:
|
||||
return any(
|
||||
member.get("user_id") == user_id and member.get("role") == "user" for member in _team_members(gateway, team_id)
|
||||
)
|
||||
|
||||
|
||||
def _delete_key_if_present(gateway: Gateway, key: str) -> None:
|
||||
digest: Final = sha256(key.encode()).hexdigest()
|
||||
if read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', (digest,)):
|
||||
deleted: Final = gateway.request("POST", "/key/delete", {"keys": [key]})
|
||||
assert deleted.status_code == 200, deleted.text
|
||||
assert read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', (digest,)) == []
|
||||
|
||||
|
||||
def _key_with_cleanup(scenario: Scenario, **fields: JsonValue) -> str:
|
||||
created: Final = scenario.gateway.post("/key/generate", fields)
|
||||
key: Final = string_value(created["key"])
|
||||
scenario.cleanups.callback(_delete_key_if_present, scenario.gateway, key)
|
||||
return key
|
||||
|
||||
|
||||
def _assert_serving(gateway: Gateway, model: str, key: str) -> None:
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{"model": model, "messages": [{"role": "user", "content": "SCIM lifecycle integration"}]},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["usage"]["total_tokens"] == 40, response.text
|
||||
assert response.json()["choices"][0]["message"]["content"] == (
|
||||
"Hello! This is a mock response from the fake OpenAI endpoint."
|
||||
), response.text
|
||||
|
||||
|
||||
def test_okta_create_provisions_the_user_into_its_group_and_rejects_a_duplicate(gateway: Gateway) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
model: Final = scenario.model()
|
||||
team: Final = scenario.team(models=[model])
|
||||
team_info: Final = object_value(gateway.get("/team/info", {"team_id": team})["team_info"])
|
||||
team_alias: Final = string_value(team_info["team_alias"])
|
||||
suffix: Final = uuid.uuid4().hex
|
||||
user_name: Final = f"scim-{suffix}@example.com"
|
||||
user_email: Final = user_name
|
||||
external_id: Final = f"external-{suffix}"
|
||||
manager_id: Final = f"manager-{suffix}"
|
||||
schemas: Final = [SCIM_CORE_USER_SCHEMA, SCIM_ENTERPRISE_USER_SCHEMA]
|
||||
enterprise: Final = {
|
||||
"employeeNumber": f"employee-{suffix}",
|
||||
"department": "Research",
|
||||
"manager": {"value": manager_id},
|
||||
}
|
||||
requested: Final[dict[str, JsonValue]] = {
|
||||
"schemas": schemas,
|
||||
"userName": user_name,
|
||||
"name": {"givenName": "SCIM", "familyName": "Provisioned"},
|
||||
"emails": [{"primary": True, "type": "work", "value": user_email}],
|
||||
"externalId": external_id,
|
||||
"active": True,
|
||||
"groups": [{"value": team}],
|
||||
SCIM_ENTERPRISE_USER_SCHEMA: enterprise,
|
||||
}
|
||||
created: Final = _scim_request(gateway, "POST", "/scim/v2/Users", requested)
|
||||
scenario.cleanups.callback(_delete_user_if_present, gateway, user_name)
|
||||
assert created.status_code == 201, created.text
|
||||
expected_user: Final = ScimUserResponse(
|
||||
schemas=schemas,
|
||||
id=user_name,
|
||||
externalId=None,
|
||||
userName=user_name,
|
||||
displayName=user_name,
|
||||
active=True,
|
||||
meta=ScimUserMeta(resourceType="User"),
|
||||
name=ScimName(givenName="SCIM", familyName="Provisioned"),
|
||||
emails=[ScimEmail(value=user_email, type=None, primary=True)],
|
||||
groups=[ScimGroupRef(value=team, display=team_alias, type="direct")],
|
||||
enterprise=ScimEnterpriseUser(
|
||||
employeeNumber=f"employee-{suffix}",
|
||||
department="Research",
|
||||
manager=ScimManager(value=manager_id),
|
||||
),
|
||||
)
|
||||
created_user: Final = ScimUserResponse.model_validate(created.json())
|
||||
create_exclude: Final = {"externalId": True, "meta": {"created": True, "lastModified": True}}
|
||||
assert created_user.model_dump(by_alias=True, exclude=create_exclude) == expected_user.model_dump(
|
||||
by_alias=True,
|
||||
exclude=create_exclude,
|
||||
), created.text
|
||||
readback_response: Final = _scim_request(gateway, "GET", f"/scim/v2/Users/{user_name}")
|
||||
assert readback_response.status_code == 200, readback_response.text
|
||||
readback_user: Final = ScimUserResponse.model_validate(readback_response.json())
|
||||
metadata_rows: Final = read_rows(
|
||||
'SELECT created_at::text AS created_at, updated_at::text AS updated_at '
|
||||
'FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user_name,),
|
||||
)
|
||||
assert len(metadata_rows) == 1, (metadata_rows, readback_response.text)
|
||||
created_at: Final = string_value(metadata_rows[0]["created_at"])
|
||||
updated_at: Final = string_value(metadata_rows[0]["updated_at"])
|
||||
created_at_utc: Final = _utc_datetime(created_at)
|
||||
updated_at_utc: Final = _utc_datetime(updated_at)
|
||||
assert readback_user.meta.created is not None, readback_response.text
|
||||
assert readback_user.meta.lastModified is not None, readback_response.text
|
||||
assert _utc_datetime(readback_user.meta.created) == created_at_utc, (
|
||||
created_at,
|
||||
readback_user.meta.created,
|
||||
readback_response.text,
|
||||
)
|
||||
assert _utc_datetime(readback_user.meta.lastModified) == updated_at_utc, (
|
||||
updated_at,
|
||||
readback_user.meta.lastModified,
|
||||
readback_response.text,
|
||||
)
|
||||
expected_readback_user: Final = expected_user.model_copy(
|
||||
update={
|
||||
"meta": ScimUserMeta(
|
||||
resourceType="User",
|
||||
created=created_at_utc.isoformat(),
|
||||
lastModified=updated_at_utc.isoformat(),
|
||||
)
|
||||
}
|
||||
)
|
||||
assert readback_user.model_dump(by_alias=True, exclude={"externalId": True}) == (
|
||||
expected_readback_user.model_dump(by_alias=True, exclude={"externalId": True})
|
||||
), readback_response.text
|
||||
|
||||
user_rows: Final = read_rows(
|
||||
'SELECT user_id, user_email, user_role, teams, metadata FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user_name,),
|
||||
)
|
||||
assert len(user_rows) == 1, user_rows
|
||||
assert user_rows[0]["user_id"] == user_name, user_rows
|
||||
assert user_rows[0]["user_email"] == user_email, user_rows
|
||||
assert user_rows[0]["user_role"] == "internal_user_viewer", user_rows
|
||||
assert user_rows[0]["teams"] == [team], user_rows
|
||||
metadata: Final = object_value(user_rows[0]["metadata"])
|
||||
assert metadata.get("scim_metadata") == {
|
||||
"givenName": "SCIM",
|
||||
"familyName": "Provisioned",
|
||||
}, metadata
|
||||
assert metadata.get("scim_enterprise") == enterprise, metadata
|
||||
assert _has_team_member(gateway, team, user_name), _team_members(gateway, team)
|
||||
key: Final = scenario.key(user_id=user_name, team_id=team, models=[model])
|
||||
_assert_serving(gateway, model, key)
|
||||
|
||||
duplicate: Final = _scim_request(gateway, "POST", "/scim/v2/Users", requested)
|
||||
assert duplicate.status_code == 409, duplicate.text
|
||||
assert object_value(duplicate.json()) == {
|
||||
"detail": {"error": f"User already exists with username: {user_name}"}
|
||||
}, duplicate.text
|
||||
assert read_rows(
|
||||
'SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user_name,),
|
||||
) == [{"user_id": user_name}]
|
||||
|
||||
|
||||
def test_scim_create_adopts_an_existing_user_by_email_and_keeps_its_key_and_teams(gateway: Gateway) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
suffix: Final = uuid.uuid4().hex
|
||||
user_email: Final = f"existing-{suffix}@example.com"
|
||||
existing_user: Final = scenario.user(user_email=user_email, user_role="internal_user")
|
||||
model: Final = scenario.model()
|
||||
team: Final = scenario.team(models=[model])
|
||||
team_info: Final = object_value(gateway.get("/team/info", {"team_id": team})["team_info"])
|
||||
team_alias: Final = string_value(team_info["team_alias"])
|
||||
added_member: Final = gateway.request(
|
||||
"POST",
|
||||
"/team/member_add",
|
||||
{"team_id": team, "member": {"role": "user", "user_id": existing_user}},
|
||||
)
|
||||
assert added_member.status_code == 200, added_member.text
|
||||
key: Final = _key_with_cleanup(scenario, user_id=existing_user, team_id=team, models=[model])
|
||||
_assert_serving(gateway, model, key)
|
||||
team_before: Final = _team_members(gateway, team)
|
||||
user_before: Final = read_rows(
|
||||
'SELECT user_id, user_email, user_role, teams FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(existing_user,),
|
||||
)
|
||||
idp_user_name: Final = f"idp-login-{suffix}"
|
||||
created: Final = _scim_request(
|
||||
gateway,
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": idp_user_name,
|
||||
"emails": [{"value": user_email, "primary": True, "type": "work"}],
|
||||
},
|
||||
)
|
||||
assert created.status_code == 201, created.text
|
||||
created_user: Final = ScimUserResponse.model_validate(created.json())
|
||||
assert created_user.id == existing_user, created.text
|
||||
readback: Final = _scim_request(gateway, "GET", f"/scim/v2/Users/{existing_user}")
|
||||
assert readback.status_code == 200, readback.text
|
||||
readback_user: Final = ScimUserResponse.model_validate(readback.json())
|
||||
assert readback_user.id == existing_user, readback.text
|
||||
assert readback_user.groups == [
|
||||
ScimGroupRef(value=team, display=team_alias, type="direct")
|
||||
], readback.text
|
||||
assert len(read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_email = %s', (user_email,))) == 1
|
||||
assert (
|
||||
read_rows(
|
||||
'SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(idp_user_name,),
|
||||
)
|
||||
== []
|
||||
)
|
||||
assert (
|
||||
read_rows(
|
||||
'SELECT user_id, user_email, user_role, teams FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(existing_user,),
|
||||
)
|
||||
== user_before
|
||||
)
|
||||
assert _team_members(gateway, team) == team_before
|
||||
_assert_serving(gateway, model, key)
|
||||
|
||||
|
||||
def test_profile_put_without_groups_keeps_memberships_and_a_new_groups_list_moves_them(gateway: Gateway) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
model_a: Final = scenario.model()
|
||||
model_b: Final = scenario.model()
|
||||
team_a: Final = scenario.team(models=[model_a])
|
||||
team_b: Final = scenario.team(models=[model_b])
|
||||
team_a_info: Final = object_value(gateway.get("/team/info", {"team_id": team_a})["team_info"])
|
||||
team_b_info: Final = object_value(gateway.get("/team/info", {"team_id": team_b})["team_info"])
|
||||
alias_a: Final = string_value(team_a_info["team_alias"])
|
||||
alias_b: Final = string_value(team_b_info["team_alias"])
|
||||
suffix: Final = uuid.uuid4().hex
|
||||
user_name: Final = f"scim-move-{suffix}@example.com"
|
||||
external_id: Final = f"entra-object-{suffix}"
|
||||
profile: Final[dict[str, JsonValue]] = {
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": user_name,
|
||||
"name": {"givenName": "Group", "familyName": "Member"},
|
||||
"emails": [{"value": user_name, "primary": True, "type": "work"}],
|
||||
"externalId": external_id,
|
||||
"active": True,
|
||||
}
|
||||
created: Final = _scim_request(
|
||||
gateway,
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{**profile, "groups": [{"value": team_a}]},
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, gateway, user_name)
|
||||
assert created.status_code == 201, created.text
|
||||
created_user: Final = ScimUserResponse.model_validate(created.json())
|
||||
assert created_user.groups == [
|
||||
ScimGroupRef(value=team_a, display=alias_a, type="direct")
|
||||
], created.text
|
||||
key_a: Final = _key_with_cleanup(scenario, user_id=user_name, team_id=team_a, models=[model_a])
|
||||
_assert_serving(gateway, model_a, key_a)
|
||||
|
||||
without_groups: Final = _scim_request(gateway, "PUT", f"/scim/v2/Users/{user_name}", profile)
|
||||
assert without_groups.status_code == 200, without_groups.text
|
||||
assert read_rows(
|
||||
'SELECT sso_user_id FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user_name,),
|
||||
) == [{"sso_user_id": external_id}], without_groups.text
|
||||
assert object_value(gateway.get("/user/info", {"user_id": user_name})["user_info"])["sso_user_id"] == (
|
||||
external_id
|
||||
), without_groups.text
|
||||
without_groups_user: Final = ScimUserResponse.model_validate(without_groups.json())
|
||||
assert without_groups_user.groups == [
|
||||
ScimGroupRef(value=team_a, display=alias_a, type="direct")
|
||||
], without_groups.text
|
||||
without_groups_readback: Final = _scim_request(
|
||||
gateway,
|
||||
"GET",
|
||||
f"/scim/v2/Users/{user_name}",
|
||||
)
|
||||
assert without_groups_readback.status_code == 200, without_groups_readback.text
|
||||
without_groups_readback_user: Final = ScimUserResponse.model_validate(without_groups_readback.json())
|
||||
assert without_groups_readback_user.groups == [
|
||||
ScimGroupRef(value=team_a, display=alias_a, type="direct")
|
||||
], (
|
||||
without_groups_readback.text
|
||||
)
|
||||
assert _has_team_member(gateway, team_a, user_name), _team_members(gateway, team_a)
|
||||
_assert_serving(gateway, model_a, key_a)
|
||||
|
||||
empty_groups: Final = _scim_request(
|
||||
gateway,
|
||||
"PUT",
|
||||
f"/scim/v2/Users/{user_name}",
|
||||
{**profile, "groups": []},
|
||||
)
|
||||
assert empty_groups.status_code == 200, empty_groups.text
|
||||
empty_groups_user: Final = ScimUserResponse.model_validate(empty_groups.json())
|
||||
assert empty_groups_user.groups == [
|
||||
ScimGroupRef(value=team_a, display=alias_a, type="direct")
|
||||
], empty_groups.text
|
||||
empty_groups_readback: Final = _scim_request(gateway, "GET", f"/scim/v2/Users/{user_name}")
|
||||
assert empty_groups_readback.status_code == 200, empty_groups_readback.text
|
||||
empty_groups_readback_user: Final = ScimUserResponse.model_validate(empty_groups_readback.json())
|
||||
assert empty_groups_readback_user.groups == [
|
||||
ScimGroupRef(value=team_a, display=alias_a, type="direct")
|
||||
], (
|
||||
empty_groups_readback.text
|
||||
)
|
||||
assert _has_team_member(gateway, team_a, user_name), _team_members(gateway, team_a)
|
||||
_assert_serving(gateway, model_a, key_a)
|
||||
|
||||
moved: Final = _scim_request(
|
||||
gateway,
|
||||
"PUT",
|
||||
f"/scim/v2/Users/{user_name}",
|
||||
{**profile, "groups": [{"value": team_b}]},
|
||||
)
|
||||
assert moved.status_code == 200, moved.text
|
||||
moved_user: Final = ScimUserResponse.model_validate(moved.json())
|
||||
assert moved_user.groups == [
|
||||
ScimGroupRef(value=team_b, display=alias_b, type="direct")
|
||||
], moved.text
|
||||
moved_readback: Final = _scim_request(gateway, "GET", f"/scim/v2/Users/{user_name}")
|
||||
assert moved_readback.status_code == 200, moved_readback.text
|
||||
moved_readback_user: Final = ScimUserResponse.model_validate(moved_readback.json())
|
||||
assert moved_readback_user.groups == [
|
||||
ScimGroupRef(value=team_b, display=alias_b, type="direct")
|
||||
], moved_readback.text
|
||||
assert not _has_team_member(gateway, team_a, user_name), _team_members(gateway, team_a)
|
||||
assert _has_team_member(gateway, team_b, user_name), _team_members(gateway, team_b)
|
||||
refused: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{"model": model_a, "messages": [{"role": "user", "content": "SCIM moved membership"}]},
|
||||
key=key_a,
|
||||
)
|
||||
expected_refusal: Final = {
|
||||
"error": {
|
||||
"message": (
|
||||
"Authentication Error, Invalid proxy server token passed. "
|
||||
f"Received API Key = sk-...{key_a[-4:]}, Key Hash (Token) ={sha256(key_a.encode()).hexdigest()}. "
|
||||
"Unable to find token in cache or `LiteLLM_VerificationTokenTable`"
|
||||
),
|
||||
"type": "token_not_found_in_db",
|
||||
"param": "key",
|
||||
"code": "401",
|
||||
}
|
||||
}
|
||||
assert refused.status_code == 401, refused.text
|
||||
assert object_value(refused.json()) == expected_refusal, refused.text
|
||||
key_b: Final = _key_with_cleanup(scenario, user_id=user_name, team_id=team_b, models=[model_b])
|
||||
_assert_serving(gateway, model_b, key_b)
|
||||
|
||||
|
||||
def test_scim_create_persists_external_id_as_the_sso_user_id(gateway: Gateway) -> None:
|
||||
pytest.skip(
|
||||
"BUG: POST /scim/v2/Users drops externalId, so the 201 resource and GET readback return externalId null and "
|
||||
"the user's sso_user_id stays NULL"
|
||||
)
|
||||
with gateway.scenario() as scenario:
|
||||
suffix: Final = uuid.uuid4().hex
|
||||
user_name: Final = f"scim-ext-{suffix}@example.com"
|
||||
external_id: Final = f"okta-{suffix}"
|
||||
created: Final = _scim_request(
|
||||
gateway,
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": user_name,
|
||||
"emails": [{"value": user_name, "primary": True, "type": "work"}],
|
||||
"externalId": external_id,
|
||||
"active": True,
|
||||
},
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, gateway, user_name)
|
||||
assert created.status_code == 201, created.text
|
||||
created_user: Final = ScimUserResponse.model_validate(created.json())
|
||||
assert created_user.externalId == external_id, created.text
|
||||
readback: Final = _scim_request(gateway, "GET", f"/scim/v2/Users/{user_name}")
|
||||
assert readback.status_code == 200, readback.text
|
||||
readback_user: Final = ScimUserResponse.model_validate(readback.json())
|
||||
assert readback_user.externalId == external_id, readback.text
|
||||
assert read_rows(
|
||||
'SELECT sso_user_id FROM "LiteLLM_UserTable" WHERE user_id = %s',
|
||||
(user_name,),
|
||||
) == [{"sso_user_id": external_id}], created.text
|
||||
assert object_value(gateway.get("/user/info", {"user_id": user_name})["user_info"])["sso_user_id"] == (
|
||||
external_id
|
||||
), created.text
|
||||
|
||||
|
||||
def test_okta_username_filter_finds_users_by_email_and_by_id(gateway: Gateway) -> None:
|
||||
with gateway.scenario() as scenario:
|
||||
suffix: Final = uuid.uuid4().hex
|
||||
email_a: Final = f"filter-existing-{suffix}@example.com"
|
||||
email_b: Final = f"filter-scim-{suffix}@example.com"
|
||||
existing: Final = scenario.user(user_email=email_a, user_role="internal_user")
|
||||
created: Final = _scim_request(
|
||||
gateway,
|
||||
"POST",
|
||||
"/scim/v2/Users",
|
||||
{
|
||||
"schemas": [SCIM_CORE_USER_SCHEMA],
|
||||
"userName": email_b,
|
||||
"emails": [{"value": email_b, "primary": True, "type": "work"}],
|
||||
},
|
||||
)
|
||||
scenario.cleanups.callback(_delete_user_if_present, gateway, email_b)
|
||||
assert created.status_code == 201, created.text
|
||||
|
||||
def lookup(value: str) -> tuple[ScimListResponse, str]:
|
||||
response: Final = gateway.request(
|
||||
"GET",
|
||||
"/scim/v2/Users",
|
||||
params={"filter": f'userName eq "{value}"', "startIndex": "1", "count": "100"},
|
||||
headers={"Accept": "application/scim+json"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
return ScimListResponse.model_validate(response.json()), response.text
|
||||
|
||||
existing_results, existing_text = lookup(email_a)
|
||||
assert (
|
||||
existing_results.schemas,
|
||||
existing_results.totalResults,
|
||||
existing_results.startIndex,
|
||||
existing_results.itemsPerPage,
|
||||
[(resource.id, resource.userName) for resource in existing_results.Resources],
|
||||
) == (
|
||||
[SCIM_LIST_RESPONSE_SCHEMA],
|
||||
1,
|
||||
1,
|
||||
1,
|
||||
[(existing, email_a)],
|
||||
), existing_text
|
||||
|
||||
existing_id_results, existing_id_text = lookup(existing)
|
||||
assert (
|
||||
existing_id_results.schemas,
|
||||
existing_id_results.totalResults,
|
||||
existing_id_results.startIndex,
|
||||
existing_id_results.itemsPerPage,
|
||||
[(resource.id, resource.userName) for resource in existing_id_results.Resources],
|
||||
) == (
|
||||
["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
|
||||
1,
|
||||
1,
|
||||
1,
|
||||
[(existing, email_a)],
|
||||
), existing_id_text
|
||||
|
||||
scim_results, scim_text = lookup(email_b)
|
||||
assert (
|
||||
scim_results.schemas,
|
||||
scim_results.totalResults,
|
||||
scim_results.startIndex,
|
||||
scim_results.itemsPerPage,
|
||||
[(resource.id, resource.userName) for resource in scim_results.Resources],
|
||||
) == (
|
||||
["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
|
||||
1,
|
||||
1,
|
||||
1,
|
||||
[(email_b, email_b)],
|
||||
), scim_text
|
||||
|
||||
absent_results, absent_text = lookup(f"absent-{suffix}@example.com")
|
||||
assert (
|
||||
absent_results.schemas,
|
||||
absent_results.totalResults,
|
||||
absent_results.startIndex,
|
||||
absent_results.itemsPerPage,
|
||||
absent_results.Resources,
|
||||
) == (
|
||||
["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
|
||||
0,
|
||||
1,
|
||||
0,
|
||||
[],
|
||||
), absent_text
|
||||
Loading…
Add table
Reference in a new issue