mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-05 07:37:50 +00:00
fix(proxy): keep stored SSO client secrets when settings are saved back masked
PATCH /update/sso_settings now merges into the effective SSO config: a field the request omits keeps its stored value, a client secret sent back as the mask /get/sso_settings returned keeps the real secret, and an explicit null or empty string still clears. The update response masks client secrets the way the read endpoint does. Adds the Keycloak-judged e2e the ticket describes and extends the mapped unit tests Co-authored-by: Harshit1259 <hrshingala825@gmail.com>
This commit is contained in:
parent
9a38ff4cb5
commit
b5080cd1d4
8 changed files with 411 additions and 22 deletions
|
|
@ -33,6 +33,7 @@ from litellm.proxy.config_resolvers.settings_store import ConfigOwnedKeyError
|
|||
from litellm.proxy.config_resolvers.sso import (
|
||||
SSO_FIELD_ENV_VARS,
|
||||
SSO_SECRET_FIELDS,
|
||||
ResolvedSSOConfig,
|
||||
resolve_sso_config,
|
||||
)
|
||||
from litellm.proxy.management_endpoints.team_admin_field_permissions import (
|
||||
|
|
@ -1230,6 +1231,49 @@ async def get_sso_settings():
|
|||
}
|
||||
|
||||
|
||||
_SSO_SETTINGS_VALUES: Final = TypeAdapter(dict[str, JsonValue])
|
||||
|
||||
|
||||
def _stored_sso_settings(record: _StoredSsoSettingsRow | None) -> Mapping[str, JsonValue] | None:
|
||||
if record is None or not record.sso_settings:
|
||||
return None
|
||||
stored: Final = record.sso_settings
|
||||
decoded: Final = _DECODED_JSON.validate_python(json.loads(stored)) if isinstance(stored, str) else stored
|
||||
try:
|
||||
return _SSO_SETTINGS_VALUES.validate_python(decoded)
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def _current_sso_value(current: ResolvedSSOConfig, field_name: str, value: JsonValue) -> JsonValue:
|
||||
return None if current.provenance.get(field_name) in ("default", "unset") else value
|
||||
|
||||
|
||||
def _merged_sso_field(field_name: str, incoming: JsonValue, current: JsonValue, submitted: bool) -> JsonValue:
|
||||
if not submitted:
|
||||
return current
|
||||
if field_name not in SSO_SECRET_FIELDS or not isinstance(current, str) or not current:
|
||||
return incoming
|
||||
if incoming == mask_sensitive_keys({field_name: current}, {field_name})[field_name]:
|
||||
return current
|
||||
return incoming
|
||||
|
||||
|
||||
def merge_sso_settings_update(sso_config: SSOConfig, current: ResolvedSSOConfig) -> Mapping[str, JsonValue]:
|
||||
"""Fields the request omits keep their effective value, and a client secret sent back
|
||||
as the mask /get/sso_settings returned keeps the real secret. An explicit null or empty
|
||||
string still clears a field."""
|
||||
incoming: Final = _SSO_SETTINGS_VALUES.validate_python(sso_config.model_dump(mode="json"))
|
||||
current_values: Final = {
|
||||
name: _current_sso_value(current, name, value)
|
||||
for name, value in _SSO_SETTINGS_VALUES.validate_python(current.config.model_dump(mode="json")).items()
|
||||
}
|
||||
return {
|
||||
name: _merged_sso_field(name, value, current_values[name], name in sso_config.model_fields_set)
|
||||
for name, value in incoming.items()
|
||||
}
|
||||
|
||||
|
||||
@router.patch(
|
||||
"/update/sso_settings",
|
||||
tags=["SSO Settings"],
|
||||
|
|
@ -1272,15 +1316,12 @@ async def update_sso_settings(
|
|||
existing_sso_record: Final = await _stored_sso_settings_db(SSOConfigRepository(prisma_client)).find_unique(
|
||||
where={"id": "sso_config"}
|
||||
)
|
||||
before_sso_data: dict[str, JsonValue] | None = None
|
||||
if existing_sso_record and existing_sso_record.sso_settings:
|
||||
stored = existing_sso_record.sso_settings
|
||||
if isinstance(stored, str):
|
||||
stored = _DECODED_JSON.validate_python(json.loads(stored))
|
||||
if isinstance(stored, dict):
|
||||
before_sso_data = ( # rebind-ok: pre-existing rebinding on a rename-only line
|
||||
proxy_config.decrypt_db_variables(stored)
|
||||
)
|
||||
stored_sso_settings: Final = _stored_sso_settings(existing_sso_record)
|
||||
before_sso_data: Final = (
|
||||
_SSO_SETTINGS_VALUES.validate_python(proxy_config.decrypt_db_variables(dict(stored_sso_settings)))
|
||||
if stored_sso_settings is not None
|
||||
else None
|
||||
)
|
||||
|
||||
# Load existing config
|
||||
config: Final = await proxy_config.get_config()
|
||||
|
|
@ -1294,17 +1335,17 @@ async def update_sso_settings(
|
|||
config["general_settings"] = {}
|
||||
|
||||
# Update environment variables in config and in memory
|
||||
sso_data: Final = sso_config.model_dump()
|
||||
sso_data: Final = merge_sso_settings_update(sso_config, resolve_sso_config(stored_sso_settings, os.environ))
|
||||
for field_name, value in sso_data.items():
|
||||
if field_name in SSO_FIELD_ENV_VARS:
|
||||
env_var_name = SSO_FIELD_ENV_VARS[field_name]
|
||||
if value:
|
||||
if isinstance(value, str) and value:
|
||||
os.environ[env_var_name] = value
|
||||
else:
|
||||
# Clear environment variable if value is null/empty
|
||||
os.environ.pop(env_var_name, None)
|
||||
|
||||
encrypted_sso_data: Final = proxy_config.encrypt_env_variables(environment_variables=sso_data)
|
||||
encrypted_sso_data: Final = proxy_config.encrypt_env_variables(environment_variables=dict(sso_data))
|
||||
|
||||
# Save to dedicated SSO table
|
||||
await _stored_sso_settings_db(SSOConfigRepository(prisma_client)).upsert(
|
||||
|
|
@ -1325,7 +1366,7 @@ async def update_sso_settings(
|
|||
param_name="sso_config",
|
||||
action="updated",
|
||||
before_value=before_sso_data,
|
||||
after_value=sso_data,
|
||||
after_value=dict(sso_data),
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
table_name=LitellmTableNames.SSO_CONFIG_TABLE_NAME,
|
||||
)
|
||||
|
|
@ -1368,7 +1409,7 @@ async def update_sso_settings(
|
|||
return {
|
||||
"message": "SSO settings updated successfully",
|
||||
"status": "success",
|
||||
"settings": sso_data,
|
||||
"settings": mask_sensitive_keys(sso_data, set(SSO_SECRET_FIELDS)),
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -105,6 +105,7 @@
|
|||
- {id: mgmt.user.jwt.database_roles, module: mgmt, tier: P0, surface: api, assertions: [database_roles], source: "auth/handle_jwt.py", rationale: "User-only JWT subjects retain their seeded database roles and memberships"}
|
||||
- {id: mgmt.key.jwt.viewer_denied, module: mgmt, tier: P0, surface: api, assertions: [viewer_denied], source: "auth/route_checks.py", rationale: "An admin viewer can read a key but cannot update it or change stored state"}
|
||||
- {id: mgmt.user.oidc.identity_mapping, module: mgmt, tier: P0, surface: api, assertions: [identity_mapping], source: "tests/e2e/idp.py", rationale: "IdP configuration canary only: confidential-client token and userinfo subjects match the seeded user; application SSO is separate"}
|
||||
- {id: mgmt.sso_settings.update.secret_preserved, module: mgmt, tier: P0, surface: api, assertions: [secret_preserved], source: "ui_crud_endpoints/proxy_setting_endpoints.py", fail_before_fix: proven, rationale: "Saving the SSO settings page back with the masked client secret /get/sso_settings showed keeps the stored secret, so the IdP still accepts the proxy's client credentials on the next login"}
|
||||
- {id: mgmt.team.jwt.tenant_isolation, module: mgmt, tier: P0, surface: api, assertions: [tenant_isolation], source: "auth/handle_jwt.py", rationale: "Isolated team actors read their own key and receive 403 for the other tenant key"}
|
||||
- {id: mgmt.team.jwt.multiple_memberships, module: mgmt, tier: P0, surface: api, assertions: [multiple_memberships], source: "auth/handle_jwt.py", rationale: "A multi-group actor has exactly the configured memberships without admin scope"}
|
||||
- {id: mgmt.user.jwt.cleanup, module: mgmt, tier: P0, surface: api, assertions: [cleanup], source: "management_endpoints/internal_user_endpoints.py", rationale: "Owned users teams organizations keys and IdP objects disappear after successful cleanup"}
|
||||
|
|
|
|||
|
|
@ -593,6 +593,27 @@ def post_json_external(
|
|||
)
|
||||
|
||||
|
||||
def put_json_external(
|
||||
url: str,
|
||||
*,
|
||||
headers: BaseModel,
|
||||
json: BaseModel,
|
||||
timeout: float = 30.0,
|
||||
) -> ExternalWrite:
|
||||
"""PUT an absolute URL outside the proxy under its own bearer, for an API whose
|
||||
update answers with a status and no body (Keycloak's realm event config)."""
|
||||
try:
|
||||
resp = requests.put(
|
||||
url,
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return ExternalWrite(status_code=-1, body=str(exc))
|
||||
return ExternalWrite(status_code=resp.status_code, body=resp.text)
|
||||
|
||||
|
||||
def send_text_external(
|
||||
method: Literal["GET", "POST", "PATCH"],
|
||||
url: str,
|
||||
|
|
|
|||
|
|
@ -29,10 +29,11 @@ from e2e_http import (
|
|||
get_external,
|
||||
post_form_external,
|
||||
post_json_external,
|
||||
put_json_external,
|
||||
unwrap,
|
||||
)
|
||||
from e2e_metadata import step
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic import BaseModel, Field, RootModel
|
||||
|
||||
KEYCLOAK_URL_ENV: Final = "E2E_KEYCLOAK_URL"
|
||||
KEYCLOAK_REALM_ENV: Final = "E2E_KEYCLOAK_REALM"
|
||||
|
|
@ -321,6 +322,25 @@ class Keycloak:
|
|||
assert configured.attributes.pkce == "S256"
|
||||
return client
|
||||
|
||||
@step("Turn on Keycloak's login event log for the realm")
|
||||
def enable_login_events(self) -> None:
|
||||
write: Final = put_json_external(
|
||||
self._admin_url("/events/config"), headers=self._admin_headers(), json=EventsConfigBody()
|
||||
)
|
||||
if write.status_code != 204:
|
||||
pytest.fail(f"Keycloak refused to enable login events: HTTP {write.status_code} {write.body[:300]}")
|
||||
|
||||
@step("Read the code-to-token errors Keycloak logged for the browser SSO client, newest first")
|
||||
def code_to_token_errors(self, client: BrowserClient) -> tuple[str, ...]:
|
||||
events: Final = unwrap(
|
||||
get_external(
|
||||
self._admin_url(f"/events?client={client.client_id}&type=CODE_TO_TOKEN_ERROR&max=100"),
|
||||
headers=self._admin_headers(),
|
||||
response_type=LoginEvents,
|
||||
)
|
||||
)
|
||||
return tuple(event.error for event in events.root)
|
||||
|
||||
@step("Get a Keycloak access token for {identity.username} through the browser SSO client")
|
||||
def browser_token(self, identity: Identity, client: BrowserClient) -> str:
|
||||
return self._token(
|
||||
|
|
@ -387,6 +407,19 @@ class UserInfo(BaseModel):
|
|||
email: str
|
||||
|
||||
|
||||
class EventsConfigBody(BaseModel):
|
||||
events_enabled: bool = Field(default=True, alias="eventsEnabled")
|
||||
|
||||
|
||||
class LoginEvent(BaseModel):
|
||||
type: str
|
||||
error: str = ""
|
||||
|
||||
|
||||
class LoginEvents(RootModel[tuple[LoginEvent, ...]]):
|
||||
pass
|
||||
|
||||
|
||||
class BrowserAttributes(BaseModel):
|
||||
pkce: str = Field(default="S256", alias="pkce.code.challenge.method")
|
||||
|
||||
|
|
|
|||
|
|
@ -59,6 +59,10 @@ from models import (
|
|||
OrgNewBody,
|
||||
OrgNewResponse,
|
||||
OrgUpdateBody,
|
||||
SSOCallbackParams,
|
||||
SSOSettingsResponse,
|
||||
SSOSettingsUpdateBody,
|
||||
SSOSettingsUpdateResponse,
|
||||
TagDeleteBody,
|
||||
TagListEntry,
|
||||
TagListResponse,
|
||||
|
|
@ -198,6 +202,32 @@ class ManagementClient:
|
|||
)
|
||||
)
|
||||
|
||||
@step("Read the SSO settings the dashboard shows from /get/sso_settings")
|
||||
def sso_settings(self) -> SSOSettingsResponse:
|
||||
return unwrap(
|
||||
self.proxy.transport.get(
|
||||
"/get/sso_settings",
|
||||
headers=self.proxy.management_headers(),
|
||||
params=NoBody(),
|
||||
response_type=SSOSettingsResponse,
|
||||
)
|
||||
)
|
||||
|
||||
@step("Save SSO settings with /update/sso_settings")
|
||||
def update_sso_settings(self, body: SSOSettingsUpdateBody) -> SSOSettingsUpdateResponse:
|
||||
return unwrap(
|
||||
self.proxy.transport.patch(
|
||||
"/update/sso_settings",
|
||||
headers=self.proxy.management_headers(),
|
||||
json=body,
|
||||
response_type=SSOSettingsUpdateResponse,
|
||||
)
|
||||
)
|
||||
|
||||
@step("Finish an SSO login at /sso/callback with the authorization code {code}")
|
||||
def sso_callback(self, code: str) -> ProbeResult:
|
||||
return self.proxy.transport.probe("/sso/callback", params=SSOCallbackParams(code=code, state="e2e"))
|
||||
|
||||
@step("Read the key's settings back from /key/info")
|
||||
def key_info_as(self, key: str, *, caller_key: str | None = None) -> Result[KeyInfoResponse]:
|
||||
return self.proxy.transport.get(
|
||||
|
|
@ -256,6 +286,7 @@ class ManagementClient:
|
|||
response_type=NoBody,
|
||||
)
|
||||
)
|
||||
|
||||
@step("Regenerate the virtual key with /key/regenerate")
|
||||
def regenerate_key(self, key: str, *, grace_period: str | None = None) -> str:
|
||||
return unwrap(
|
||||
|
|
|
|||
81
tests/e2e/management/test_sso_settings_update_e2e.py
Normal file
81
tests/e2e/management/test_sso_settings_update_e2e.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
"""Saving SSO settings back the way the dashboard does must keep the stored client secret."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from e2e_config import PROXY_BASE_URL, unique_marker
|
||||
from e2e_metadata import Domain, Route, Subject, meta
|
||||
from idp import Keycloak
|
||||
from lifecycle import ResourceManager
|
||||
from management.management_client import ManagementClient
|
||||
from models import CLEAR, SSORoleMappings, SSOSettingsUpdateBody
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
|
||||
def _cleared_sso_settings() -> SSOSettingsUpdateBody:
|
||||
return SSOSettingsUpdateBody(
|
||||
generic_client_id=CLEAR,
|
||||
generic_client_secret=CLEAR,
|
||||
generic_authorization_endpoint=CLEAR,
|
||||
generic_token_endpoint=CLEAR,
|
||||
generic_userinfo_endpoint=CLEAR,
|
||||
proxy_base_url=CLEAR,
|
||||
role_mappings=CLEAR,
|
||||
)
|
||||
|
||||
|
||||
class TestSsoSettingsUpdate:
|
||||
@pytest.mark.covers("mgmt.sso_settings.update.secret_preserved")
|
||||
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.ADMIN_UI))
|
||||
def test_saving_role_mappings_keeps_the_client_secret(
|
||||
self, client: ManagementClient, idp: Keycloak, resources: ResourceManager
|
||||
) -> None:
|
||||
"""The IdP is the judge: a bogus authorization code exchanged with the right client
|
||||
secret is logged as invalid_code, one exchanged with a wrong secret as
|
||||
invalid_client_credentials. The dashboard only ever holds the masked secret, so the
|
||||
second save sends the mask back."""
|
||||
browser: Final = idp.browser_client(callback_url=f"{PROXY_BASE_URL}/sso/callback", defer=resources.defer)
|
||||
idp.enable_login_events()
|
||||
discovery: Final = idp.discovery()
|
||||
resources.defer(lambda: client.update_sso_settings(_cleared_sso_settings()))
|
||||
client.update_sso_settings(
|
||||
SSOSettingsUpdateBody(
|
||||
generic_client_id=browser.client_id,
|
||||
generic_client_secret=browser.secret,
|
||||
generic_authorization_endpoint=discovery.authorization_endpoint,
|
||||
generic_token_endpoint=discovery.token_endpoint,
|
||||
generic_userinfo_endpoint=discovery.userinfo_endpoint,
|
||||
proxy_base_url=PROXY_BASE_URL,
|
||||
)
|
||||
)
|
||||
client.sso_callback(code=f"e2e-bogus-code-{unique_marker()}")
|
||||
assert idp.code_to_token_errors(browser) == ("invalid_code",), (
|
||||
"the freshly saved secret must authenticate the proxy at the IdP before the round trip proves anything"
|
||||
)
|
||||
|
||||
shown: Final = client.sso_settings().values
|
||||
assert shown.generic_client_secret != browser.secret, "/get/sso_settings must not show the secret in clear"
|
||||
saved: Final = client.update_sso_settings(
|
||||
SSOSettingsUpdateBody(
|
||||
generic_client_id=shown.generic_client_id,
|
||||
generic_client_secret=shown.generic_client_secret,
|
||||
generic_authorization_endpoint=shown.generic_authorization_endpoint,
|
||||
generic_token_endpoint=shown.generic_token_endpoint,
|
||||
generic_userinfo_endpoint=shown.generic_userinfo_endpoint,
|
||||
generic_scope=shown.generic_scope,
|
||||
proxy_base_url=shown.proxy_base_url,
|
||||
role_mappings=SSORoleMappings(
|
||||
provider="generic", group_claim="groups", roles={"proxy_admin": [f"e2e-admins-{unique_marker()}"]}
|
||||
),
|
||||
)
|
||||
)
|
||||
assert saved.settings.generic_client_secret != browser.secret
|
||||
assert saved.settings.role_mappings is not None
|
||||
|
||||
client.sso_callback(code=f"e2e-bogus-code-{unique_marker()}")
|
||||
assert idp.code_to_token_errors(browser) == ("invalid_code", "invalid_code"), (
|
||||
"the IdP rejected the proxy's client credentials: saving the settings page replaced the stored secret"
|
||||
)
|
||||
|
|
@ -1535,6 +1535,62 @@ class Cleared(BaseModel):
|
|||
CLEAR: Final = Cleared()
|
||||
|
||||
|
||||
class SSORoleMappings(BaseModel):
|
||||
provider: str
|
||||
group_claim: str
|
||||
roles: dict[str, list[str]]
|
||||
|
||||
|
||||
class SSOSettingsUpdateBody(BaseModel):
|
||||
"""PATCH /update/sso_settings is a merge patch like /key/update: a field left `None`
|
||||
is dropped from the body and keeps its stored value, `CLEAR` sends an explicit null
|
||||
that clears it. The dashboard sends the client secret back exactly as
|
||||
/get/sso_settings showed it, which is the masked form, and that must keep the stored
|
||||
secret rather than replace it."""
|
||||
|
||||
generic_client_id: str | Cleared | None = None
|
||||
generic_client_secret: str | Cleared | None = Field(default=None, repr=False)
|
||||
generic_authorization_endpoint: str | Cleared | None = None
|
||||
generic_token_endpoint: str | Cleared | None = None
|
||||
generic_userinfo_endpoint: str | Cleared | None = None
|
||||
generic_scope: str | Cleared | None = None
|
||||
proxy_base_url: str | Cleared | None = None
|
||||
role_mappings: SSORoleMappings | Cleared | None = None
|
||||
|
||||
|
||||
class SSOSettingsValues(BaseModel):
|
||||
"""The generic-provider slice of `values` in /get/sso_settings. Secrets come back masked."""
|
||||
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
generic_client_id: str | None = None
|
||||
generic_client_secret: str | None = Field(default=None, repr=False)
|
||||
generic_authorization_endpoint: str | None = None
|
||||
generic_token_endpoint: str | None = None
|
||||
generic_userinfo_endpoint: str | None = None
|
||||
generic_scope: str | None = None
|
||||
proxy_base_url: str | None = None
|
||||
role_mappings: SSORoleMappings | None = None
|
||||
|
||||
|
||||
class SSOSettingsResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
values: SSOSettingsValues
|
||||
|
||||
|
||||
class SSOSettingsUpdateResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
status: str
|
||||
settings: SSOSettingsValues
|
||||
|
||||
|
||||
class SSOCallbackParams(BaseModel):
|
||||
code: str
|
||||
state: str
|
||||
|
||||
|
||||
class KeyUpdateBody(BaseModel):
|
||||
"""POST /key/update is a merge patch: a field left `None` is dropped from the body and
|
||||
keeps its stored value, `CLEAR` sends an explicit null that clears it (`budget_duration`
|
||||
|
|
|
|||
|
|
@ -689,16 +689,13 @@ class TestProxySettingEndpoints:
|
|||
# Verify settings were updated
|
||||
settings = data["settings"]
|
||||
assert settings["google_client_id"] == new_sso_settings["google_client_id"]
|
||||
assert (
|
||||
settings["google_client_secret"] == new_sso_settings["google_client_secret"]
|
||||
)
|
||||
assert settings["google_client_secret"] != new_sso_settings["google_client_secret"]
|
||||
assert "*" in settings["google_client_secret"]
|
||||
assert (
|
||||
settings["microsoft_client_id"] == new_sso_settings["microsoft_client_id"]
|
||||
)
|
||||
assert (
|
||||
settings["microsoft_client_secret"]
|
||||
== new_sso_settings["microsoft_client_secret"]
|
||||
)
|
||||
assert settings["microsoft_client_secret"] != new_sso_settings["microsoft_client_secret"]
|
||||
assert "*" in settings["microsoft_client_secret"]
|
||||
assert settings["proxy_base_url"] == new_sso_settings["proxy_base_url"]
|
||||
assert settings["user_email"] == new_sso_settings["user_email"]
|
||||
|
||||
|
|
@ -1139,6 +1136,134 @@ class TestProxySettingEndpoints:
|
|||
assert create_sso_settings["ui_access_mode"] is None
|
||||
assert create_sso_settings["user_email"] is None
|
||||
|
||||
def _mock_sso_update_db(self, monkeypatch, stored_sso_settings):
|
||||
"""Point /update/sso_settings at a stored SSO row (or None) with encryption as identity."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
|
||||
monkeypatch.setenv("LITELLM_SALT_KEY", "test_salt_key")
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", True)
|
||||
mock_prisma = MagicMock()
|
||||
if stored_sso_settings is None:
|
||||
record = None
|
||||
else:
|
||||
record = MagicMock()
|
||||
record.sso_settings = stored_sso_settings
|
||||
mock_prisma.db.litellm_ssoconfig.find_unique = AsyncMock(return_value=record)
|
||||
mock_prisma.db.litellm_ssoconfig.upsert = AsyncMock()
|
||||
mock_prisma.writer_db.litellm_config.find_unique = AsyncMock(return_value=None)
|
||||
mock_prisma.db.litellm_config.update = AsyncMock()
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
|
||||
monkeypatch.setattr(proxy_config, "encrypt_env_variables", lambda environment_variables: environment_variables)
|
||||
identity = lambda value, key=None, exception_type="error", return_original_value=False: value # noqa: E731
|
||||
monkeypatch.setattr("litellm.proxy.config_resolvers.sso.decrypt_value_helper", identity)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.decrypt_value_helper", identity)
|
||||
return mock_prisma
|
||||
|
||||
@staticmethod
|
||||
def _upserted_sso_settings(mock_prisma):
|
||||
import json
|
||||
|
||||
return json.loads(mock_prisma.db.litellm_ssoconfig.upsert.call_args.kwargs["data"]["update"]["sso_settings"])
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"secret_field,env_var",
|
||||
[
|
||||
("google_client_secret", "GOOGLE_CLIENT_SECRET"),
|
||||
("microsoft_client_secret", "MICROSOFT_CLIENT_SECRET"),
|
||||
("generic_client_secret", "GENERIC_CLIENT_SECRET"),
|
||||
],
|
||||
)
|
||||
def test_update_sso_settings_round_tripped_mask_keeps_stored_secret(
|
||||
self, mock_proxy_config, mock_auth, monkeypatch, secret_field, env_var
|
||||
):
|
||||
"""Regression for LIT-8683: saving the settings page (which only ever holds the masked
|
||||
secret from /get/sso_settings) must not overwrite the stored client secret with the mask."""
|
||||
monkeypatch.delenv(env_var, raising=False)
|
||||
stored_secret = "stored-client-secret-0123456789"
|
||||
mock_prisma = self._mock_sso_update_db(monkeypatch, {secret_field: stored_secret})
|
||||
|
||||
shown = client.get("/get/sso_settings").json()["values"]
|
||||
assert shown[secret_field] != stored_secret
|
||||
role_mappings = {"provider": "generic", "group_claim": "groups", "roles": {"proxy_admin": ["admins"]}}
|
||||
response = client.patch("/update/sso_settings", json={**shown, "role_mappings": role_mappings})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
persisted = self._upserted_sso_settings(mock_prisma)
|
||||
assert persisted[secret_field] == stored_secret
|
||||
assert persisted["role_mappings"]["roles"]["proxy_admin"] == ["admins"]
|
||||
assert os.environ[env_var] == stored_secret
|
||||
assert response.json()["settings"][secret_field] == shown[secret_field]
|
||||
|
||||
def test_update_sso_settings_round_tripped_mask_keeps_env_secret(
|
||||
self, mock_proxy_config, mock_auth, monkeypatch
|
||||
):
|
||||
"""A secret that only lives in the process env is shown masked too, and saving the
|
||||
page persists that secret rather than the mask."""
|
||||
monkeypatch.setenv("GENERIC_CLIENT_SECRET", "env-only-secret-0123456789")
|
||||
mock_prisma = self._mock_sso_update_db(monkeypatch, None)
|
||||
|
||||
shown = client.get("/get/sso_settings").json()["values"]
|
||||
response = client.patch("/update/sso_settings", json=shown)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert self._upserted_sso_settings(mock_prisma)["generic_client_secret"] == "env-only-secret-0123456789"
|
||||
assert os.environ["GENERIC_CLIENT_SECRET"] == "env-only-secret-0123456789"
|
||||
|
||||
def test_update_sso_settings_omitted_fields_keep_stored_values(
|
||||
self, mock_proxy_config, mock_auth, monkeypatch
|
||||
):
|
||||
"""The UI access control form sends only ui_access_mode. That must not null out the
|
||||
rest of the stored SSO configuration."""
|
||||
monkeypatch.delenv("GENERIC_CLIENT_SECRET", raising=False)
|
||||
stored = {
|
||||
"generic_client_id": "stored-client-id",
|
||||
"generic_client_secret": "stored-client-secret-0123456789",
|
||||
"generic_token_endpoint": "https://idp.example.com/token",
|
||||
"role_mappings": {"provider": "generic", "group_claim": "groups", "roles": {"proxy_admin": ["admins"]}},
|
||||
}
|
||||
mock_prisma = self._mock_sso_update_db(monkeypatch, stored)
|
||||
|
||||
response = client.patch("/update/sso_settings", json={"ui_access_mode": "none"})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
persisted = self._upserted_sso_settings(mock_prisma)
|
||||
assert persisted["ui_access_mode"] == "none"
|
||||
assert persisted["generic_client_id"] == "stored-client-id"
|
||||
assert persisted["generic_client_secret"] == "stored-client-secret-0123456789"
|
||||
assert persisted["generic_token_endpoint"] == "https://idp.example.com/token"
|
||||
assert persisted["role_mappings"]["roles"]["proxy_admin"] == ["admins"]
|
||||
assert os.environ["GENERIC_CLIENT_SECRET"] == "stored-client-secret-0123456789"
|
||||
|
||||
def test_update_sso_settings_new_secret_replaces_stored_secret(
|
||||
self, mock_proxy_config, mock_auth, monkeypatch
|
||||
):
|
||||
mock_prisma = self._mock_sso_update_db(
|
||||
monkeypatch, {"generic_client_secret": "stored-client-secret-0123456789"}
|
||||
)
|
||||
|
||||
response = client.patch("/update/sso_settings", json={"generic_client_secret": "rotated-secret-9876543210"})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert self._upserted_sso_settings(mock_prisma)["generic_client_secret"] == "rotated-secret-9876543210"
|
||||
assert os.environ["GENERIC_CLIENT_SECRET"] == "rotated-secret-9876543210"
|
||||
assert "rotated-secret-9876543210" not in response.text
|
||||
|
||||
@pytest.mark.parametrize("cleared", [None, ""])
|
||||
def test_update_sso_settings_explicit_clear_drops_stored_secret(
|
||||
self, mock_proxy_config, mock_auth, monkeypatch, cleared
|
||||
):
|
||||
mock_prisma = self._mock_sso_update_db(
|
||||
monkeypatch, {"generic_client_secret": "stored-client-secret-0123456789"}
|
||||
)
|
||||
|
||||
response = client.patch("/update/sso_settings", json={"generic_client_secret": cleared})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert not self._upserted_sso_settings(mock_prisma)["generic_client_secret"]
|
||||
assert "GENERIC_CLIENT_SECRET" not in os.environ
|
||||
|
||||
def test_get_ui_theme_settings(self, mock_proxy_config):
|
||||
"""Test getting UI theme settings without authentication"""
|
||||
response = client.get("/get/ui_theme_settings")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue