test(e2e): add SCIM v2 provisioning journey suite

Walk the IdP-driven provisioning lifecycle against a live proxy with a scoped
SCIM token (a virtual key limited to /scim/*): POST /scim/v2/Users provisions a
real internal user (verified via /user/info), a SCIM PATCH active=false blocks
that user's key at auth (proven by chat serving before the flip and 401 after),
DELETE deprovisions it (404 + gone from /user/list), and POST /scim/v2/Groups
provisions a real team (verified via /team/info). Plus the permission boundary:
a key not scoped to /scim/* is refused.

Adds SCIMProvisioningClient + scim_client fixture, SCIM v2 wire models, and five
mgmt coverage-registry cells. SCIM is enterprise-gated, so this needs a licensed
proxy.

Part of LIT-4639.

tests/e2e/coverage_registry/mgmt.yaml

@@ -75,3 +75,9 @@
 - {id: mgmt.sso_settings.update.admin_only, module: mgmt, tier: P0, surface: api, assertions: [admin_only], source: "route_checks.py:326", rationale: "Non-admin key cannot read or write the SSO provider config"}
 - {id: mgmt.sso.readiness.reports_provider, module: mgmt, tier: P1, surface: api, assertions: [reports_provider], source: "ui_sso.py:2303", rationale: "/sso/readiness reports the configured provider healthy once all its env vars are set"}
 - {id: mgmt.sso.readiness.reports_missing_vars, module: mgmt, tier: P1, surface: api, assertions: [reports_missing_vars], source: "ui_sso.py:2374", rationale: "/sso/readiness 503s and lists the missing env vars when a provider is only partially configured"}
+# SCIM v2 provisioning (enterprise). IdP-driven user/group lifecycle; grounded in management_endpoints/scim/scim_v2.py.
+- {id: mgmt.scim.user_provision.persists, module: mgmt, tier: P1, surface: api, assertions: [persists], source: "scim_v2.py:1111", rationale: "POST /scim/v2/Users provisions a real internal user, visible via /user/info with the SCIM email"}
+- {id: mgmt.scim.user_deactivate.blocks_key, module: mgmt, tier: P0, surface: api, assertions: [blocks_key], source: "scim_v2.py:1541", rationale: "SCIM PATCH active=false blocks the deactivated user's key at auth (cross-feature enforcement)"}
+- {id: mgmt.scim.user_deprovision.persists, module: mgmt, tier: P1, surface: api, assertions: [persists], source: "scim_v2.py:1268", rationale: "DELETE /scim/v2/Users removes the internal user (404 after, gone from /user/list)"}
+- {id: mgmt.scim.group_provision.persists, module: mgmt, tier: P1, surface: api, assertions: [persists], source: "scim_v2.py:1713", rationale: "POST /scim/v2/Groups provisions a real team, visible via /team/info with the SCIM displayName"}
+- {id: mgmt.scim.user_provision.denied_without_permission, module: mgmt, tier: P1, surface: api, assertions: [denied_without_permission], source: "scim_v2.py:143", rationale: "A key not scoped to /scim/* cannot provision via SCIM"}

tests/e2e/management/conftest.py

@@ -9,6 +9,7 @@ import pytest
 
 from management_client import ManagementClient, build_client
 from proxy_client import ProxyClient
+from scim_provisioning_client import SCIMProvisioningClient, build_scim_client
 from sso_management_client import SSOManagementClient, build_sso_client
 
 
@@ -27,3 +28,8 @@ def client(proxy: ProxyClient) -> ManagementClient:
 @pytest.fixture(scope="session")
 def sso_client(proxy: ProxyClient) -> SSOManagementClient:
     return build_sso_client(proxy)
+
+
+@pytest.fixture(scope="session")
+def scim_client(proxy: ProxyClient) -> SCIMProvisioningClient:
+    return build_scim_client(proxy)

tests/e2e/management/scim_provisioning_client.py

@@ -0,0 +1,100 @@
+"""Client for the SCIM v2 provisioning e2e suite: the shared ProxyClient plus the
+/scim/v2 Users and Groups operations an identity provider drives when it
+provisions, updates, and deprovisions accounts against the gateway.
+
+SCIM calls authenticate with a scoped provisioning token (a virtual key whose
+allowed_routes is ["/scim/*"]), which is exactly what the IdP is configured with;
+the token is passed per call so a test can also drive the denial path with a key
+that lacks the permission. Verification read-backs (/user/info, /team/info) use the
+master key through ManagementClient, so the suite injects both clients.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from e2e_http import NoBody, Result, unwrap
+from models import (
+    SCIMGroupBody,
+    SCIMGroupResponse,
+    SCIMPatchOp,
+    SCIMUserBody,
+    SCIMUserResponse,
+)
+from proxy_client import ProxyClient
+
+
+@dataclass(frozen=True, slots=True)
+class SCIMProvisioningClient:
+    proxy: ProxyClient
+
+    def create_user(self, token: str, body: SCIMUserBody) -> SCIMUserResponse:
+        return unwrap(
+            self.proxy.transport.post(
+                "/scim/v2/Users",
+                headers=self.proxy.transport.bearer(token),
+                json=body,
+                response_type=SCIMUserResponse,
+            )
+        )
+
+    def create_user_result(self, token: str, body: SCIMUserBody) -> Result[SCIMUserResponse]:
+        """The raw outcome, for the access-control path where a key without /scim/*
+        permission must be refused."""
+        return self.proxy.transport.post(
+            "/scim/v2/Users",
+            headers=self.proxy.transport.bearer(token),
+            json=body,
+            response_type=SCIMUserResponse,
+        )
+
+    def get_user_result(self, token: str, user_id: str) -> Result[SCIMUserResponse]:
+        return self.proxy.transport.get(
+            f"/scim/v2/Users/{user_id}",
+            headers=self.proxy.transport.bearer(token),
+            params=NoBody(),
+            response_type=SCIMUserResponse,
+        )
+
+    def patch_user(self, token: str, user_id: str, op: SCIMPatchOp) -> None:
+        _ = unwrap(
+            self.proxy.transport.patch(
+                f"/scim/v2/Users/{user_id}",
+                headers=self.proxy.transport.bearer(token),
+                json=op,
+                response_type=SCIMUserResponse,
+            )
+        )
+
+    def delete_user(self, token: str, user_id: str) -> None:
+        """DELETE answers 204 with an empty body (which the typed layer can't parse),
+        so the call is fire-and-effect: tests prove deletion by the follow-up 404 and
+        the internal user being gone. Best-effort, so teardown re-deletes harmlessly."""
+        _ = self.proxy.transport.delete(
+            f"/scim/v2/Users/{user_id}",
+            headers=self.proxy.transport.bearer(token),
+            json=NoBody(),
+            response_type=NoBody,
+        )
+
+    def create_group(self, token: str, body: SCIMGroupBody) -> SCIMGroupResponse:
+        return unwrap(
+            self.proxy.transport.post(
+                "/scim/v2/Groups",
+                headers=self.proxy.transport.bearer(token),
+                json=body,
+                response_type=SCIMGroupResponse,
+            )
+        )
+
+    def delete_group(self, token: str, group_id: str) -> None:
+        _ = self.proxy.transport.delete(
+            f"/scim/v2/Groups/{group_id}",
+            headers=self.proxy.transport.bearer(token),
+            json=NoBody(),
+            response_type=NoBody,
+        )
+
+
+def build_scim_client(proxy: ProxyClient) -> SCIMProvisioningClient:
+    return SCIMProvisioningClient(proxy=proxy)

tests/e2e/management/test_scim_provisioning_e2e.py

@@ -0,0 +1,205 @@
+"""Live e2e: the SCIM v2 provisioning lifecycle an enterprise IdP (Okta, Entra,
+etc.) drives against the gateway - provision a user, watch it become a real
+internal user, deactivate it and see the account's access cut, then deprovision
+it; plus group-to-team provisioning and the permission boundary on the SCIM token.
+
+SCIM is enterprise-gated (the /scim/v2 router requires a premium license) and
+authenticated with a scoped provisioning token: a virtual key whose allowed_routes
+is ["/scim/*"], exactly what the IdP is given. Verification read-backs use the
+master key via ManagementClient, so each test injects both clients.
+
+Requires a licensed proxy (LITELLM_LICENSE) with STORE_MODEL_IN_DB=True and a DB.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+
+import pytest
+
+from e2e_config import unique_marker
+from e2e_http import Result, UnauthorizedError, UnknownApiError
+from lifecycle import ResourceManager
+from management_client import ManagementClient
+from models import (
+    KeyGenerateBody,
+    LiteLLMParamsBody,
+    SCIMEmail,
+    SCIMGroupBody,
+    SCIMName,
+    SCIMPatchOp,
+    SCIMPatchOperation,
+    SCIMUserBody,
+    SCIMUserResponse,
+)
+from scim_provisioning_client import SCIMProvisioningClient
+
+pytestmark = pytest.mark.e2e
+
+# SCIM emails are validated as real addresses, so a routable domain is required
+# (reserved TLDs such as .test / .example are rejected by the server).
+_EMAIL_DOMAIN = "litellm-e2e.com"
+
+
+def _poll[T](client: ManagementClient, attempt: Callable[[], T | None], failure: str) -> T:
+    deadline = time.monotonic() + client.proxy.poll_timeout
+    while time.monotonic() < deadline:
+        found = attempt()
+        if found is not None:
+            return found
+        time.sleep(client.proxy.poll_interval)
+    pytest.fail(failure)
+
+
+def _scim_token(client: ManagementClient, resources: ResourceManager) -> str:
+    """A scoped SCIM provisioning token: a virtual key restricted to the /scim/*
+    routes, the way an IdP is configured. Auto-deleted on teardown."""
+    token = client.proxy.generate_key(KeyGenerateBody(allowed_routes=["/scim/*"]))
+    resources.defer(lambda: client.proxy.delete_key(token))
+    return token
+
+
+def _provision_user(
+    scim_client: SCIMProvisioningClient, token: str, resources: ResourceManager, marker: str
+) -> tuple[SCIMUserResponse, str]:
+    email = f"scim-{marker}@{_EMAIL_DOMAIN}"
+    user = scim_client.create_user(
+        token,
+        SCIMUserBody(
+            userName=f"scim-user-{marker}",
+            emails=[SCIMEmail(value=email)],
+            name=SCIMName(givenName="E2E", familyName="Scim"),
+        ),
+    )
+    resources.defer(lambda: scim_client.delete_user(token, user.id))
+    return user, email
+
+
+class TestSCIMUserProvisioning:
+    @pytest.mark.covers("mgmt.scim.user_provision.persists")
+    def test_provision_creates_a_real_internal_user(
+        self, client: ManagementClient, scim_client: SCIMProvisioningClient, resources: ResourceManager
+    ) -> None:
+        marker = unique_marker()
+        token = _scim_token(client, resources)
+
+        user, email = _provision_user(scim_client, token, resources, marker)
+        assert user.id == f"scim-user-{marker}", (
+            f"SCIM create returned id {user.id!r}; userName should map to the litellm user_id 'scim-user-{marker}'"
+        )
+        assert user.active is True, "a freshly provisioned SCIM user should be active"
+
+        info = client.user_info(user.id).user_info
+        assert info.user_email == email, (
+            f"/user/info reports user_email {info.user_email!r} for the SCIM-provisioned user, expected {email!r}"
+        )
+
+    @pytest.mark.covers("mgmt.scim.user_deprovision.persists")
+    def test_deprovision_removes_the_internal_user(
+        self, client: ManagementClient, scim_client: SCIMProvisioningClient, resources: ResourceManager
+    ) -> None:
+        marker = unique_marker()
+        token = _scim_token(client, resources)
+        user, _ = _provision_user(scim_client, token, resources, marker)
+        assert client.user_count(user.id) == 1, f"SCIM-provisioned user {user.id} absent from /user/list before delete"
+
+        scim_client.delete_user(token, user.id)
+
+        after = scim_client.get_user_result(token, user.id)
+        match after:
+            case UnknownApiError(status_code=code):
+                assert code == 404, f"GET /scim/v2/Users/{user.id} after delete should be 404, got {code}"
+            case _:
+                pytest.fail(f"GET /scim/v2/Users/{user.id} after delete should be 404, got {after}")
+
+        _ = _poll(
+            client,
+            lambda: True if client.user_count(user.id) == 0 else None,
+            f"SCIM-deleted user {user.id} still present in /user/list after the deadline",
+        )
+
+
+class TestSCIMUserDeactivation:
+    @pytest.mark.covers("mgmt.scim.user_deactivate.blocks_key")
+    def test_deactivation_blocks_the_users_key_on_chat(
+        self, client: ManagementClient, scim_client: SCIMProvisioningClient, resources: ResourceManager
+    ) -> None:
+        """The cross-feature promise: an IdP setting active=false via SCIM PATCH cuts
+        the deactivated user's gateway access. Proven end to end - the user's key
+        serves chat before the flip and is rejected at auth after it."""
+        marker = unique_marker()
+        token = _scim_token(client, resources)
+        user, _ = _provision_user(scim_client, token, resources, marker)
+
+        model_name = f"scim-mock-{marker}"
+        model_id = client.proxy.create_model(
+            model_name, LiteLLMParamsBody(model="openai/gpt-4o-mini", mock_response="scim ok")
+        )
+        resources.defer(lambda: client.proxy.delete_model(model_id))
+
+        key = client.proxy.generate_key(KeyGenerateBody(user_id=user.id, models=[model_name]))
+        resources.defer(lambda: client.proxy.delete_key(key))
+
+        _ = _poll(
+            client,
+            lambda: True if client.chat_status(key, model_name, f"hi {unique_marker()}").ok else None,
+            "the provisioned user's key never served chat before deactivation",
+        )
+
+        scim_client.patch_user(
+            token, user.id, SCIMPatchOp(Operations=[SCIMPatchOperation(op="replace", path="active", value=False)])
+        )
+
+        _ = _poll(
+            client,
+            lambda: True if client.chat_status(key, model_name, f"hi {unique_marker()}").status_code == 401 else None,
+            "the SCIM-deactivated user's key was never rejected (401) on chat after the deadline",
+        )
+
+
+class TestSCIMGroupProvisioning:
+    @pytest.mark.covers("mgmt.scim.group_provision.persists")
+    def test_group_provision_creates_a_real_team(
+        self, client: ManagementClient, scim_client: SCIMProvisioningClient, resources: ResourceManager
+    ) -> None:
+        marker = unique_marker()
+        token = _scim_token(client, resources)
+        display_name = f"scim-team-{marker}"
+
+        group = scim_client.create_group(token, SCIMGroupBody(displayName=display_name))
+        resources.defer(lambda: scim_client.delete_group(token, group.id))
+
+        info = _poll(
+            client,
+            lambda: (lambda data: data if data.team_alias == display_name else None)(client.team_info(group.id)),
+            f"/team/info never reported team_alias {display_name!r} for the SCIM-provisioned group {group.id}",
+        )
+        assert info.team_alias == display_name
+
+
+class TestSCIMAccessControl:
+    @pytest.mark.covers("mgmt.scim.user_provision.denied_without_permission")
+    def test_token_without_scim_permission_is_denied(
+        self, client: ManagementClient, scim_client: SCIMProvisioningClient, resources: ResourceManager
+    ) -> None:
+        """A key not scoped to /scim/* cannot provision. The IdP's provisioning token
+        must carry the SCIM route permission; a general key is refused."""
+        marker = unique_marker()
+        key = client.proxy.generate_key(KeyGenerateBody(allowed_routes=["llm_api_routes"]))
+        resources.defer(lambda: client.proxy.delete_key(key))
+
+        result: Result[SCIMUserResponse] = scim_client.create_user_result(
+            key,
+            SCIMUserBody(
+                userName=f"scim-denied-{marker}",
+                emails=[SCIMEmail(value=f"scim-denied-{marker}@{_EMAIL_DOMAIN}")],
+            ),
+        )
+        match result:
+            case UnauthorizedError():
+                return
+            case UnknownApiError(status_code=403):
+                return
+            case _:
+                pytest.fail(f"POST /scim/v2/Users with a non-SCIM key must be denied (401 or 403), got {result}")

tests/e2e/models.py

@@ -896,3 +896,82 @@ class SSOReadinessResponse(BaseModel):
     status: str
     sso_configured: bool
     provider: str | None = None
+
+
+# ---------- SCIM v2 provisioning ----------
+
+SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
+SCIM_GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group"
+SCIM_PATCH_OP_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:PatchOp"
+
+
+# SCIM wire names are camelCase/PascalCase; the field names match the protocol
+# exactly so no aliasing is needed and the serialized body is spec-correct.
+
+
+class SCIMName(BaseModel):
+    givenName: str | None = None
+    familyName: str | None = None
+
+
+class SCIMEmail(BaseModel):
+    """A SCIM email. `value` is validated as a real email address by the server, so
+    it must use a routable domain (reserved TLDs like .test/.example are rejected)."""
+
+    value: str
+    primary: bool = True
+    type: str = "work"
+
+
+class SCIMUserBody(BaseModel):
+    """POST /scim/v2/Users body. `userName` maps to the litellm user_id; the first
+    email's `value` becomes the internal user's user_email."""
+
+    schemas: list[str] = [SCIM_USER_SCHEMA]
+    userName: str
+    name: SCIMName | None = None
+    displayName: str | None = None
+    emails: list[SCIMEmail] = []
+    active: bool = True
+
+
+class SCIMUserResponse(BaseModel):
+    """A SCIM User resource. `id` is the litellm user_id (what the IdP addresses the
+    user by); the server derives the response `userName` from the user_email, so tests
+    key off `id`, not the echoed userName. extra=ignore drops the fields not asserted."""
+
+    model_config = ConfigDict(extra="ignore")
+    id: str
+    userName: str | None = None
+    active: bool = True
+
+
+class SCIMPatchOperation(BaseModel):
+    op: str
+    path: str | None = None
+    value: bool | str | dict[str, object] | list[object] | None = None
+
+
+class SCIMPatchOp(BaseModel):
+    schemas: list[str] = [SCIM_PATCH_OP_SCHEMA]
+    Operations: list[SCIMPatchOperation]
+
+
+class SCIMGroupMember(BaseModel):
+    value: str
+    display: str | None = None
+
+
+class SCIMGroupBody(BaseModel):
+    schemas: list[str] = [SCIM_GROUP_SCHEMA]
+    displayName: str
+    members: list[SCIMGroupMember] = []
+
+
+class SCIMGroupResponse(BaseModel):
+    """A SCIM Group resource. `id` is the litellm team_id; `displayName` mirrors the
+    team_alias."""
+
+    model_config = ConfigDict(extra="ignore")
+    id: str
+    displayName: str | None = None