scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-17 23:51:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 67

test(e2e): add SSO management journey suite (Entra + Okta)

Browse source 
Cover the browser-free half of enterprise SSO onboarding against a live
proxy: an admin writes a provider config via PATCH /update/sso_settings,
reads it back via GET /get/sso_settings (non-secret fields persist, the
OAuth client secret comes back masked), and GET /sso/readiness reflects the
configured provider - healthy when complete, 503 listing the missing env
vars when partial. Microsoft Entra ID and Okta are separate classes, plus a
non-admin access-control case.

Adds a dedicated SSOManagementClient (kept off ManagementClient), an
sso_client fixture, composed provider config models (EntraSSOConfig |
OktaSSOConfig off a shared base), and five mgmt coverage-registry cells.

Part of LIT-4639.
This commit is contained in:
mubashir1osmani 2026-07-20 19:39:22 -07:00
parent 731efafbca
commit 78213773e6
5 changed files with 460 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
tests/e2e/coverage_registry/mgmt.yaml
View file

@ -69,3 +69,9 @@
- {id: mgmt.workflow.list.happy_path, module: mgmt, tier: P2, surface: api, assertions: [happy_path], source: "workflow_management_endpoints.py", rationale: "Workflow tracking (smoke)"}
- {id: mgmt.credential_migration.check.happy_path, module: mgmt, tier: P2, surface: api, assertions: [happy_path], source: "key_management_endpoints.py:4252", rationale: "Encryption migration (smoke)"}
- {id: mgmt.credential.new.serves_request, module: mgmt, tier: P1, surface: api, assertions: [serves_request], source: "credential_endpoints/endpoints.py:42", rationale: "Stored credential resolves into a deployment and serves a live /messages request"}
# SSO management (Entra ID + Okta). Browser-free config surface; grounded in ui_crud_endpoints/proxy_setting_endpoints.py + ui_sso.py.
- {id: mgmt.sso_settings.update.persists, module: mgmt, tier: P1, surface: api, assertions: [persists], source: "proxy_setting_endpoints.py:810", rationale: "SSO provider config survives the update->get round-trip via the LiteLLM_SSOConfig table"}
- {id: mgmt.sso_settings.secret_masked, module: mgmt, tier: P0, surface: api, assertions: [secret_masked], source: "proxy_setting_endpoints.py:789", rationale: "OAuth client secret is masked on read; the plaintext is never returned to the UI"}
- {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"}

6
tests/e2e/management/conftest.py
View file

@ -9,6 +9,7 @@ import pytest
from management_client import ManagementClient, build_client
from proxy_client import ProxyClient
from sso_management_client import SSOManagementClient, build_sso_client
def pytest_configure(config: pytest.Config) -> None:
@ -21,3 +22,8 @@ def pytest_configure(config: pytest.Config) -> None:
@pytest.fixture(scope="session")
def client(proxy: ProxyClient) -> ManagementClient:
return build_client(proxy)
@pytest.fixture(scope="session")
def sso_client(proxy: ProxyClient) -> SSOManagementClient:
return build_sso_client(proxy)

99
tests/e2e/management/sso_management_client.py Normal file
View file

@ -0,0 +1,99 @@
"""Client for the SSO-management e2e suite: the shared ProxyClient plus the
enterprise SSO configuration surface an admin drives without a browser OAuth
dance - the /update/sso_settings write, the /get/sso_settings read-back (with
its masked secrets), and the /sso/readiness health check that reflects the
configured provider.
Kept separate from ManagementClient so the key/team/user routes and the SSO
routes stay independent; both hold the same shared ProxyClient, so the resources
fixture tears down through it either way.
"""
from __future__ import annotations
from dataclasses import dataclass
from e2e_http import NoBody, ProbeResult, Result, unwrap
from models import (
SSOConfigBody,
SSOConfigClear,
SSOReadinessResponse,
SSOSettingsResponse,
SSOSettingsValues,
)
from proxy_client import ProxyClient
@dataclass(frozen=True, slots=True)
class SSOManagementClient:
proxy: ProxyClient
def update_sso_settings(self, body: SSOConfigBody) -> None:
_ = unwrap(
self.proxy.transport.patch(
"/update/sso_settings",
headers=self.proxy.transport.master,
json=body,
response_type=NoBody,
)
)
def reset_sso_settings(self) -> None:
"""Clear every SSO env var + the stored config back to the unconfigured
baseline. Best-effort (teardown), so a failure is swallowed rather than
raised."""
_ = self.proxy.transport.patch(
"/update/sso_settings",
headers=self.proxy.transport.master,
json=SSOConfigClear(),
response_type=NoBody,
)
def get_sso_settings(self) -> SSOSettingsValues:
return unwrap(
self.proxy.transport.get(
"/get/sso_settings",
headers=self.proxy.transport.master,
params=NoBody(),
response_type=SSOSettingsResponse,
)
).values
def get_sso_settings_as(self, key: str) -> Result[SSOSettingsResponse]:
"""GET /get/sso_settings under an arbitrary key, so the access-control test
can assert a non-admin key is refused (a 403 comes back as UnknownApiError)."""
return self.proxy.transport.get(
"/get/sso_settings",
headers=self.proxy.transport.bearer(key),
params=NoBody(),
response_type=SSOSettingsResponse,
)
def update_sso_settings_as(self, key: str, body: SSOConfigBody) -> Result[NoBody]:
"""PATCH /update/sso_settings under an arbitrary key, for the same
access-control assertion on the write path."""
return self.proxy.transport.patch(
"/update/sso_settings",
headers=self.proxy.transport.bearer(key),
json=body,
response_type=NoBody,
)
def sso_readiness(self) -> SSOReadinessResponse:
return unwrap(
self.proxy.transport.get(
"/sso/readiness",
headers=self.proxy.transport.master,
params=NoBody(),
response_type=SSOReadinessResponse,
)
)
def sso_readiness_probe(self) -> ProbeResult:
"""Raw /sso/readiness outcome, for the partial-config path where the proxy
answers 503 (a non-2xx that would fail the typed read-back)."""
return self.proxy.transport.probe("/sso/readiness", params=NoBody())
def build_sso_client(proxy: ProxyClient) -> SSOManagementClient:
return SSOManagementClient(proxy=proxy)

270
tests/e2e/management/test_sso_management_e2e.py Normal file
View file

@ -0,0 +1,270 @@
"""Live e2e: the enterprise SSO configuration surface an admin drives when
onboarding an identity provider, for Microsoft Entra ID (Azure AD) and Okta.
This is the browser-free half of SSO: the OAuth login dance needs a real IdP, but
the configuration lifecycle an admin performs first - write the provider settings,
read them back, and confirm the proxy reports the provider ready - runs fully
against a live proxy. Each provider is its own class so the file reads as a spec
for how that provider is configured in production.
The SSO config is a singleton (one LiteLLM_SSOConfig row plus live os.environ), not
a per-test resource. A single PATCH /update/sso_settings is a full replace: the
endpoint clears every env var it isn't given, so configuring one provider isolates
it from the others. Every test that writes config defers a reset to the
unconfigured baseline, so the classes don't leak into each other or into sibling
suites. This assumes serial execution against one proxy process (the local proof
path); the settings are process-global, so a parallel run against a shared proxy
would race on them.
Requires STORE_MODEL_IN_DB=True on the proxy (the /update/sso_settings precondition)
and the master key. Not premium-gated: the SSO settings routes are admin-auth only,
unlike /sso/key/generate and the SCIM surface.
"""
from __future__ import annotations
import pytest
from e2e_http import NoBody, Result, UnauthorizedError, UnknownApiError
from e2e_config import unique_marker
from lifecycle import ResourceManager
from models import EntraSSOConfig, OktaSSOConfig, SSOSettingsResponse
from sso_management_client import SSOManagementClient
pytestmark = pytest.mark.e2e
def _configure(sso_client: SSOManagementClient, resources: ResourceManager, body: EntraSSOConfig | OktaSSOConfig) -> None:
"""Apply an SSO provider config and queue the reset first, so teardown clears
the singleton even if the write half-applies or a later assertion fails."""
resources.defer(sso_client.reset_sso_settings)
sso_client.update_sso_settings(body)
def _assert_denied(route: str, result: Result[SSOSettingsResponse] | Result[NoBody]) -> None:
"""A non-admin key must be refused. LiteLLM denies an admin-only route by role
with a 401 (unauthorized), and by an allowed_routes mismatch with a 403; either
is a valid denial of the SSO settings surface, so accept both and reject any
success or other outcome."""
match result:
case UnauthorizedError():
return
case UnknownApiError(status_code=403):
return
case _:
pytest.fail(f"non-admin call to {route} must be denied (401 or 403), got {result}")
class TestEntraIdSSOManagement:
@pytest.mark.covers("mgmt.sso_settings.update.persists")
def test_update_persists_entra_provider_config(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
client_id = f"entra-client-{marker}"
tenant = f"entra-tenant-{marker}"
_configure(
sso_client,
resources,
EntraSSOConfig(
microsoft_client_id=client_id,
microsoft_client_secret=f"entra-secret-{marker}-do-not-leak",
microsoft_tenant=tenant,
),
)
values = sso_client.get_sso_settings()
assert values.microsoft_client_id == client_id, (
f"/get/sso_settings reports microsoft_client_id {values.microsoft_client_id!r}, configured {client_id!r}"
)
assert values.microsoft_tenant == tenant, (
f"/get/sso_settings reports microsoft_tenant {values.microsoft_tenant!r}, configured {tenant!r}"
)
@pytest.mark.covers("mgmt.sso_settings.secret_masked")
def test_client_secret_masked_on_read(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
secret = f"entra-secret-{marker}-DO-NOT-LEAK-plaintext"
_configure(
sso_client,
resources,
EntraSSOConfig(
microsoft_client_id=f"entra-client-{marker}",
microsoft_client_secret=secret,
microsoft_tenant=f"entra-tenant-{marker}",
),
)
returned = sso_client.get_sso_settings().microsoft_client_secret
assert returned is not None, "/get/sso_settings omitted microsoft_client_secret after it was configured"
assert returned != secret, "/get/sso_settings returned the OAuth client secret verbatim (no masking)"
assert secret not in returned, f"/get/sso_settings leaked the full client secret inside {returned!r}"
assert marker not in returned, f"/get/sso_settings leaked the secret's distinctive middle inside {returned!r}"
assert "*" in returned, f"/get/sso_settings did not mask the client secret; got {returned!r}"
@pytest.mark.covers("mgmt.sso.readiness.reports_provider")
def test_readiness_reports_microsoft_when_fully_configured(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
_configure(
sso_client,
resources,
EntraSSOConfig(
microsoft_client_id=f"entra-client-{marker}",
microsoft_client_secret=f"entra-secret-{marker}",
microsoft_tenant=f"entra-tenant-{marker}",
),
)
readiness = sso_client.sso_readiness()
assert readiness.sso_configured is True, "/sso/readiness reports sso_configured false after Entra was configured"
assert readiness.provider == "microsoft", (
f"/sso/readiness reports provider {readiness.provider!r}, expected 'microsoft'"
)
assert readiness.status == "healthy", (
f"/sso/readiness reports status {readiness.status!r} for a fully configured Entra provider"
)
@pytest.mark.covers("mgmt.sso.readiness.reports_missing_vars")
def test_readiness_reports_missing_vars_when_partial(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
_configure(sso_client, resources, EntraSSOConfig(microsoft_client_id=f"entra-client-{marker}"))
probe = sso_client.sso_readiness_probe()
assert probe.status_code == 503, (
f"/sso/readiness must 503 when Entra is configured without its secret/tenant, got "
f"{probe.status_code}: {probe.body[:300]}"
)
assert "MICROSOFT_CLIENT_SECRET" in probe.body, (
f"/sso/readiness 503 must name MICROSOFT_CLIENT_SECRET as missing, got: {probe.body[:300]}"
)
assert "MICROSOFT_TENANT" in probe.body, (
f"/sso/readiness 503 must name MICROSOFT_TENANT as missing, got: {probe.body[:300]}"
)
class TestOktaSSOManagement:
@pytest.mark.covers("mgmt.sso_settings.update.persists")
def test_update_persists_okta_provider_config(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
client_id = f"okta-client-{marker}"
authorization_endpoint = f"https://e2e-{marker}.okta.com/oauth2/v1/authorize"
_configure(
sso_client,
resources,
OktaSSOConfig(
generic_client_id=client_id,
generic_client_secret=f"okta-secret-{marker}-do-not-leak",
generic_authorization_endpoint=authorization_endpoint,
generic_token_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/token",
generic_userinfo_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/userinfo",
),
)
values = sso_client.get_sso_settings()
assert values.generic_client_id == client_id, (
f"/get/sso_settings reports generic_client_id {values.generic_client_id!r}, configured {client_id!r}"
)
assert values.generic_authorization_endpoint == authorization_endpoint, (
f"/get/sso_settings reports generic_authorization_endpoint {values.generic_authorization_endpoint!r}, "
f"configured {authorization_endpoint!r}"
)
@pytest.mark.covers("mgmt.sso_settings.secret_masked")
def test_client_secret_masked_on_read(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
secret = f"okta-secret-{marker}-DO-NOT-LEAK-plaintext"
_configure(
sso_client,
resources,
OktaSSOConfig(
generic_client_id=f"okta-client-{marker}",
generic_client_secret=secret,
generic_authorization_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/authorize",
generic_token_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/token",
generic_userinfo_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/userinfo",
),
)
returned = sso_client.get_sso_settings().generic_client_secret
assert returned is not None, "/get/sso_settings omitted generic_client_secret after it was configured"
assert returned != secret, "/get/sso_settings returned the OAuth client secret verbatim (no masking)"
assert secret not in returned, f"/get/sso_settings leaked the full client secret inside {returned!r}"
assert marker not in returned, f"/get/sso_settings leaked the secret's distinctive middle inside {returned!r}"
assert "*" in returned, f"/get/sso_settings did not mask the client secret; got {returned!r}"
@pytest.mark.covers("mgmt.sso.readiness.reports_provider")
def test_readiness_reports_generic_when_fully_configured(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
_configure(
sso_client,
resources,
OktaSSOConfig(
generic_client_id=f"okta-client-{marker}",
generic_client_secret=f"okta-secret-{marker}",
generic_authorization_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/authorize",
generic_token_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/token",
generic_userinfo_endpoint=f"https://e2e-{marker}.okta.com/oauth2/v1/userinfo",
),
)
readiness = sso_client.sso_readiness()
assert readiness.sso_configured is True, "/sso/readiness reports sso_configured false after Okta was configured"
assert readiness.provider == "generic", (
f"/sso/readiness reports provider {readiness.provider!r}, expected 'generic' (Okta is the generic provider)"
)
assert readiness.status == "healthy", (
f"/sso/readiness reports status {readiness.status!r} for a fully configured Okta provider"
)
@pytest.mark.covers("mgmt.sso.readiness.reports_missing_vars")
def test_readiness_reports_missing_vars_when_partial(
self, sso_client: SSOManagementClient, resources: ResourceManager
) -> None:
marker = unique_marker()
_configure(sso_client, resources, OktaSSOConfig(generic_client_id=f"okta-client-{marker}"))
probe = sso_client.sso_readiness_probe()
assert probe.status_code == 503, (
f"/sso/readiness must 503 when Okta is configured without its secret/endpoints, got "
f"{probe.status_code}: {probe.body[:300]}"
)
for expected in ("GENERIC_CLIENT_SECRET", "GENERIC_AUTHORIZATION_ENDPOINT", "GENERIC_TOKEN_ENDPOINT", "GENERIC_USERINFO_ENDPOINT"):
assert expected in probe.body, (
f"/sso/readiness 503 must name {expected} as missing, got: {probe.body[:300]}"
)
class TestSSOSettingsAccessControl:
@pytest.mark.covers("mgmt.sso_settings.update.admin_only")
def test_non_admin_key_forbidden_from_sso_settings(
self, sso_client: SSOManagementClient, resources: ResourceManager, scoped_key: str
) -> None:
"""A non-admin virtual key can neither read nor write the SSO provider config
(an admin-only surface), and a rejected write persists nothing. The reset is
deferred defensively: if the write were wrongly accepted, teardown still
clears it."""
resources.defer(sso_client.reset_sso_settings)
_assert_denied("/get/sso_settings", sso_client.get_sso_settings_as(scoped_key))
forbidden_client_id = f"should-not-persist-{unique_marker()}"
_assert_denied(
"/update/sso_settings",
sso_client.update_sso_settings_as(scoped_key, EntraSSOConfig(microsoft_client_id=forbidden_client_id)),
)
assert sso_client.get_sso_settings().microsoft_client_id != forbidden_client_id, (
"a non-admin PATCH that was supposed to be forbidden still persisted microsoft_client_id"
)

79
tests/e2e/models.py
View file

@ -817,3 +817,82 @@ class TagListResponse(RootModel[list[TagListEntry]]):
"""GET /tag/list answers with a bare array of tag configs (the stored tags plus
any dynamically-seen spend tags), not an object wrapping them. Read the rows off
.root."""
# ---------- SSO management ----------
class SSOConfigBase(BaseModel):
"""Fields every SSO provider config shares on PATCH /update/sso_settings.
Serialized exclude_none, so a provider body carries only the fields it sets;
the endpoint reads every unset field as null and clears its env var, making a
single PATCH a full replace of the proxy's SSO state."""
proxy_base_url: str | None = None
user_email: str | None = None
class EntraSSOConfig(SSOConfigBase):
"""Microsoft Entra ID (Azure AD) provider config. `microsoft_client_id` is what
makes Entra the active provider; the secret and tenant are the other two vars
/sso/readiness requires."""
microsoft_client_id: str | None = None
microsoft_client_secret: str | None = None
microsoft_tenant: str | None = None
class OktaSSOConfig(SSOConfigBase):
"""Okta (generic OIDC) provider config. Okta is wired as the "generic" provider,
so `generic_client_id` activates it and readiness requires the secret plus the
three OIDC endpoint URLs."""
generic_client_id: str | None = None
generic_client_secret: str | None = None
generic_authorization_endpoint: str | None = None
generic_token_endpoint: str | None = None
generic_userinfo_endpoint: str | None = None
type SSOConfigBody = EntraSSOConfig | OktaSSOConfig
class SSOConfigClear(BaseModel):
"""Empty PATCH /update/sso_settings body: serializes to {}, which the endpoint
reads as every field unset and so clears all SSO env vars back to the
unconfigured baseline. Used at teardown to isolate one provider test from the
next (the SSO config is a singleton row + live os.environ, not a per-test
resource)."""
class SSOSettingsValues(BaseModel):
"""The `values` block of GET /get/sso_settings: the stored provider config with
the OAuth client secrets masked. extra=ignore drops the fields a test doesn't
read (google_*, role/team mappings, ui_access_mode)."""
model_config = ConfigDict(extra="ignore")
microsoft_client_id: str | None = None
microsoft_client_secret: str | None = None
microsoft_tenant: str | None = None
generic_client_id: str | None = None
generic_client_secret: str | None = None
generic_authorization_endpoint: str | None = None
generic_token_endpoint: str | None = None
generic_userinfo_endpoint: str | None = None
class SSOSettingsResponse(BaseModel):
"""GET /get/sso_settings answer: `{values, field_schema}`. Only `values` is read."""
values: SSOSettingsValues
class SSOReadinessResponse(BaseModel):
"""GET /sso/readiness (the 200 path). `sso_configured` flips true once a provider
client id is set; `provider` names the active provider once every required var is
present. The 503 "partial config" path is read as a raw ProbeResult instead, so
its `missing_environment_variables` list is asserted off the body."""
status: str
sso_configured: bool
provider: str | None = None