Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/ui_crud_endpoints/proxy_setting

1#### CRUD ENDPOINTS for UI Settings #####

2import asyncio

3import json

4import os

5from collections import Counter

6from collections.abc import Mapping, MutableMapping, Sequence

7from types import MappingProxyType

8from typing import (

9 Annotated,

10 Final,

11 NamedTuple,

12 Protocol,

13 cast, # noqa: TID251 # prisma types Json columns as fields.Json but de-serializes them to plain python on read

14)

15from urllib.parse import urlparse

17from fastapi import APIRouter, Body, Depends, File, HTTPException, UploadFile

18from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError, create_model

19from pydantic.fields import FieldInfo, PydanticUndefined

20from typing_extensions import NotRequired, ReadOnly, TypedDict

22import litellm

23from litellm._logging import verbose_proxy_logger

24from litellm.litellm_core_utils.sensitive_data_masker import mask_sensitive_keys

25from litellm.proxy._experimental.mcp_server.tool_search import MCP_TOOL_SEARCH_SETTINGS_KEY

26from litellm.proxy._types import *

27from litellm.proxy.auth.user_api_key_auth import user_api_key_auth

28from litellm.proxy.config_resolvers import FieldSource, SettingsStore, source_for

29from litellm.proxy.config_resolvers.settings_store import ConfigOwnedKeyError

30from litellm.proxy.config_resolvers.sso import (

31 SSO_FIELD_ENV_VARS,

32 SSO_SECRET_FIELDS,

33 resolve_sso_config,

34)

35from litellm.proxy.management_endpoints.team_admin_field_permissions import (

36 SUPPORTED_TEAM_ADMIN_PERMISSIONS,

37 TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING,

38)

39from litellm.proxy.spend_tracking.ptu_feature_flag import (

40 PTU_COST_ATTRIBUTION_ENV_VAR,

41 is_ptu_cost_attribution_enabled,

42)

43from litellm.proxy.utils import invalidate_config_param

44from litellm.repositories.config_repository import ConfigRepository

45from litellm.repositories.organization_repository import OrganizationRepository

46from litellm.repositories.prisma_protocols import TableActions

47from litellm.repositories.table_repositories import (

48 SSOConfigRepository,

49 UISettingsRepository,

50)

51from litellm.repositories.team_repository import TeamRepository

52from litellm.secret_managers.main import get_secret

53from litellm.types.mcp import MCPToolSearchSettings

54from litellm.types.proxy.management_endpoints.ui_sso import (

55 DefaultTeamSSOParams,

56 SSOConfig,

57)

59router: Final = APIRouter()

62JsonSchemaItems: Final = TypedDict(

63 "JsonSchemaItems",

64 {"$ref": ReadOnly[str], "enum": ReadOnly[Sequence[JsonValue]]},

65 total=False,

66)

69class JsonSchemaNode(TypedDict, total=False):

70 type: ReadOnly[str]

71 description: ReadOnly[str]

72 enum: ReadOnly[Sequence[JsonValue]]

73 anyOf: ReadOnly[Sequence["JsonSchemaNode"]]

74 items: ReadOnly["JsonSchemaItems"]

75 properties: ReadOnly[Mapping[str, "JsonSchemaNode"]]

78_EMPTY_SCHEMA_DEFS: Final[Mapping[str, "JsonSchemaNode"]] = MappingProxyType({})

81class JsonSchemaPropertyEntry(TypedDict):

82 description: ReadOnly[str]

83 type: ReadOnly[str]

84 items: NotRequired[ReadOnly["JsonSchemaItems"]]

87class _SsoSettingsMappingRow(Protocol):

88 @property

89 def sso_settings(self) -> Mapping[str, object] | None: ... 89 ↛ exitline 89 didn't return from function 'sso_settings' because

92def _sso_settings_mapping_db(repo: SSOConfigRepository) -> TableActions[_SsoSettingsMappingRow]:

93 return cast( # cast-ok: prisma types Json columns as str; the client hands back the deserialized value

94 "TableActions[_SsoSettingsMappingRow]", repo.table

95 )

98class _StoredSsoSettingsRow(Protocol):

99 @property

100 def sso_settings(self) -> object: ... 100 ↛ exitline 100 didn't return from function 'sso_settings' because

101

102

103def _stored_sso_settings_db(repo: SSOConfigRepository) -> TableActions[_StoredSsoSettingsRow]:

104 return repo.table

105

106

107class _UiSettingsRow(Protocol):

108 @property

109 def ui_settings(self) -> str | Mapping[str, JsonValue] | None: ... 109 ↛ exitline 109 didn't return from function 'ui_settings' because

110

111

112def _ui_settings_db(repo: UISettingsRepository) -> TableActions[_UiSettingsRow]:

113 return cast( # cast-ok: prisma types Json columns as str; the client hands back the deserialized value

114 "TableActions[_UiSettingsRow]", repo.table

115 )

116

117

118class _ConfigParamRow(Protocol):

119 @property

120 def param_value(self) -> str | Mapping[str, object] | None: ... 120 ↛ exitline 120 didn't return from function 'param_value' because

121

122

123def _config_param_db(repo: ConfigRepository) -> TableActions[_ConfigParamRow]:

124 return cast( # cast-ok: prisma's LiteLLM_Config actions object, whose Json column parses to a mapping

125 "TableActions[_ConfigParamRow]", repo.table

126 )

127

128

129# Maps each UIThemeConfig field to the env var the UI branding path reads it

130# from. /update/ui_theme_settings writes both the stored ui_theme_config and

131# these env vars, so /get/ui_theme_settings resolves the same env vars to

132# reflect a deployment branded purely through process env.

133_UI_THEME_FIELD_ENV_VARS: Final[dict[str, str]] = {

134 "logo_url": "UI_LOGO_PATH",

135 "logo_url_dark": "UI_LOGO_PATH_DARK",

136 "favicon_url": "LITELLM_FAVICON_URL",

137}

138

139

140def _is_public_http_url(value: str | None) -> bool:

141 """Whether a value is a plain http(s) URL with a host, safe to disclose publicly."""

142 if not isinstance(value, str) or not value.strip(): 142 ↛ 144line 142 didn't jump to line 144 because the condition on line 142 was always true

143 return False

144 parsed: Final = urlparse(value.strip())

145 return parsed.scheme in ("http", "https") and bool(parsed.netloc)

146

147

148def _resolve_ui_theme_field(stored_values: Mapping[str, object], field_name: str) -> str | None:

149 """Resolve one UI theme field to the value the branding path actually uses.

150

151 The stored ui_theme_config wins; a field absent or blank there falls back to

152 the process environment. The branding path reads the env var, and stored

153 settings reach it by being pushed into the environment on save, so a value

154 supplied only as a process env var is live even though no stored entry exists.

155

156 This endpoint is unauthenticated, so the env fallback only surfaces a public

157 http(s) URL: an operator can point UI_LOGO_PATH at a local filesystem path

158 (the branding path serves it server-side), and that path must not be

159 disclosed to anonymous callers. A stored value is already validated as a

160 public URL on write, so it passes through.

161 """

162 stored: Final = stored_values.get(field_name)

163 if isinstance(stored, str) and stored.strip(): 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true

164 return stored

165 env_value: Final = os.environ.get(_UI_THEME_FIELD_ENV_VARS[field_name])

166 return env_value if _is_public_http_url(env_value) else None

167

168

169class IPAddress(BaseModel):

170 ip: str

171

172

173class UIThemeConfig(BaseModel):

174 """Configuration for UI theme customization"""

175

176 # Logo configuration

177 logo_url: str | None = Field(

178 default=None,

179 description="URL or path to custom logo image. Can be a local file path or HTTP/HTTPS URL",

180 )

181

182 logo_url_dark: str | None = Field(

183 default=None,

184 description=(

185 "URL or path to a custom logo image for dark mode. Can be a local file path or HTTP/HTTPS URL. "

186 "Leave unset to reuse logo_url in dark mode"

187 ),

188 )

189

190 # Favicon configuration

191 favicon_url: str | None = Field(

192 default=None,

193 description="URL to custom favicon image. Must be an HTTP/HTTPS URL to a .ico, .png, or .svg file",

194 )

195

196

197class SettingsResponse(BaseModel):

198 """Base response model for settings with values and schema information"""

199

200 values: dict[str, object]

201 """The current configuration values"""

202

203 field_schema: dict[str, object]

204 """Schema information including descriptions and property types for UI display"""

205

206

207class _SettingsWithSchema(BaseModel):

208 values: dict[str, object]

209 field_schema: dict[str, object]

210

211

212class SSOSettingsResponse(SettingsResponse):

213 """Response model for SSO settings"""

214

215 provenance: dict[str, str] = Field(default_factory=dict)

216 """Per-field source of each value: 'db', 'env', 'default', or 'unset'."""

217

218

219class InternalUserSettingsResponse(SettingsResponse):

220 """Response model for internal user settings"""

221

222

223class DefaultTeamSettingsResponse(SettingsResponse):

224 """Response model for default team settings"""

225

226

227class UIThemeSettingsResponse(SettingsResponse):

228 """Response model for UI theme settings"""

229

230

231_TEAM_ADMIN_FIELD_ENUM: Final = tuple(sorted(SUPPORTED_TEAM_ADMIN_PERMISSIONS))

232

233

234class UISettings(BaseModel):

235 """Configuration for UI-specific flags"""

236

237 model_config = ConfigDict(extra="allow")

238

239 disable_model_add_for_internal_users: bool = Field(

240 default=False,

241 description="If true, internal users cannot add models from the UI",

242 )

243

244 disable_team_admin_delete_team_user: bool = Field(

245 default=False,

246 description="Prevents Team Admins from deleting users from the teams they manage. Useful for SCIM provisioning where team membership is defined externally.",

247 )

248

249 enabled_ui_pages_internal_users: list[str] | None = Field(

250 default=None,

251 description="List of page keys that internal users (non-admins) can see in the UI sidebar. If not set, all pages are visible based on role permissions.",

252 )

253

254 require_auth_for_public_ai_hub: bool = Field(

255 default=False,

256 description="If true, requires authentication for accessing the public AI Hub.",

257 )

258

259 allow_public_health_readiness_details: bool = Field(

260 default=False,

261 description="If true, returns the legacy detailed payload from the unauthenticated /health/readiness endpoint.",

262 )

263

264 forward_client_headers_to_llm_api: bool = Field(

265 default=False,

266 description=(

267 "Forwards client headers (Authorization, anthropic-beta, and x-* "

268 "custom headers) to the upstream LLM. Enable for Claude Code with a "

269 "Max subscription (forwards the OAuth token) or to pass custom/tracing "

270 "headers through to the provider. Independent of the BYOK toggle — "

271 "enable only the one(s) you need."

272 ),

273 )

274

275 forward_llm_provider_auth_headers: bool = Field(

276 default=False,

277 description=(

278 "Forwards provider auth headers (x-api-key, x-goog-api-key, api-key, "

279 "ocp-apim-subscription-key) to the upstream LLM, overriding any "

280 "deployment-configured key for that request. Enable for Claude Code "

281 "BYOK (clients bring their own API key). Independent of the "

282 "client-headers toggle — enable only the one(s) you need."

283 ),

284 )

285

286 disable_agents_for_internal_users: bool = Field(

287 default=False,

288 description="If true, internal users cannot access agent management endpoints or the Agents page in the UI.",

289 )

290

291 allow_agents_for_team_admins: bool = Field(

292 default=False,

293 description="If true, team admins are exempt from the agents disable restriction (only takes effect when disable_agents_for_internal_users is true).",

294 )

295

296 disable_vector_stores_for_internal_users: bool = Field(

297 default=False,

298 description="If true, internal users cannot access vector store management endpoints or the Vector Stores page in the UI.",

299 )

300

301 allow_vector_stores_for_team_admins: bool = Field(

302 default=False,

303 description="If true, team admins are exempt from the vector stores disable restriction (only takes effect when disable_vector_stores_for_internal_users is true).",

304 )

305

306 scope_user_search_to_org: bool = Field(

307 default=False,

308 description="If enabled, the user search endpoint (/user/filter/ui) restricts results by organization. When off, any authenticated user can search all users.",

309 )

310

311 disable_custom_api_keys: bool = Field(

312 default=False,

313 description="If true, users cannot specify custom key values. All keys must be auto-generated.",

314 )

315

316 disable_key_generate_for_org_admin: bool = Field(

317 default=False,

318 description="If true, org admins cannot generate API keys via /key/generate.",

319 )

320

321 enable_chat_ui: bool = Field(

322 default=False,

323 description="If true, shows the Chat page in the UI sidebar, letting users chat with an LLM and connect their own MCP server credentials via OAuth.",

324 )

325

326 team_admin_editable_team_fields: Sequence[str] = Field(

327 default=(),

328 description=(

329 "Team settings fields a team admin may change on the teams they administer. "

330 "Include 'projects' to let team admins create and update projects for those teams. "

331 "Include 'member_key_budgets' to let team admins update budget fields on keys owned by other members of those teams. "

332 "Empty means team admins cannot edit team settings or manage projects at all. "

333 "Proxy admins and org admins are not affected."

334 ),

335 json_schema_extra={ # mutable-ok: pydantic only merges json_schema_extra when it is a plain dict

336 "items": {"type": "string", "enum": [*_TEAM_ADMIN_FIELD_ENUM]}, # mutable-ok: nested in the dict above

337 },

338 )

339

340

341class UISettingsResponse(SettingsResponse):

342 """Response model for UI settings"""

343

344 source: dict[str, FieldSource]

345

346

347# Allowlist of UI settings that can be stored

348ALLOWED_UI_SETTINGS_FIELDS: Final = {

349 "disable_model_add_for_internal_users",

350 "disable_team_admin_delete_team_user",

351 "enabled_ui_pages_internal_users",

352 "require_auth_for_public_ai_hub",

353 "allow_public_health_readiness_details",

354 "forward_client_headers_to_llm_api",

355 "forward_llm_provider_auth_headers",

356 "disable_agents_for_internal_users",

357 "allow_agents_for_team_admins",

358 "disable_vector_stores_for_internal_users",

359 "allow_vector_stores_for_team_admins",

360 "scope_user_search_to_org",

361 "disable_custom_api_keys",

362 "disable_key_generate_for_org_admin",

363 "enable_chat_ui",

364 TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING,

365}

366

367ENABLE_PTU_COST_ATTRIBUTION_UI_SETTING: Final = "enable_ptu_cost_attribution"

368APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING: Final = "apply_user_budget_to_team_keys"

369

370# UI settings derived from the deployment environment. Deliberately kept out of

371# ALLOWED_UI_SETTINGS_FIELDS: they are read-only, never persisted, and PATCH

372# rejects them so an admin cannot flip an env-gated feature at runtime.

373_DERIVED_UI_SETTINGS_FIELDS: Final[frozenset[str]] = frozenset(

374 {ENABLE_PTU_COST_ATTRIBUTION_UI_SETTING, APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING}

375)

376

377

378def _apply_user_budget_to_team_keys_enabled(settings: Mapping[str, object]) -> bool:

379 return settings.get(APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING) is True

380

381

382def _derived_ui_setting_value(key: str) -> object:

383 """The environment-derived value GET reports for ``key``.

384

385 PATCH compares against this rather than rejecting the key outright, so the body GET

386 hands back is still a valid PATCH body. Rejecting on presence broke read-modify-write:

387 a client that edited one setting and sent the rest back unchanged got a 400 and lost

388 the edit it actually wanted.

389 """

390 if key == ENABLE_PTU_COST_ATTRIBUTION_UI_SETTING: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true

391 return is_ptu_cost_attribution_enabled()

392 if key == APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING: 392 ↛ 398line 392 didn't jump to line 398 because the condition on line 392 was always true

393 from litellm.proxy.proxy_server import general_settings

394

395 return _apply_user_budget_to_team_keys_enabled(

396 cast(Mapping[str, object], general_settings) # cast-ok: proxy_server declares general_settings as bare dict

397 )

398 return None

399

400

401# Flags that must be synced from the persisted UISettings into

402# general_settings at runtime (on both read and write).

403_RUNTIME_GENERAL_SETTINGS_FLAGS: Final = [

404 "allow_public_health_readiness_details",

405 "forward_client_headers_to_llm_api",

406 "forward_llm_provider_auth_headers",

407 "disable_agents_for_internal_users",

408 "allow_agents_for_team_admins",

409 "disable_vector_stores_for_internal_users",

410 "allow_vector_stores_for_team_admins",

411 "disable_custom_api_keys",

412 "disable_key_generate_for_org_admin",

413 TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING,

414]

415

416# Extension point: packages outside OSS (e.g. litellm_enterprise) can

417# contribute additional UI settings fields at import time. Each entry

418# maps a field name to a (annotation, FieldInfo) tuple in pydantic

419# create_model's field-definitions format. Registering a field also

420# appends it to ALLOWED_UI_SETTINGS_FIELDS so GET/PATCH pass it through.

421#

422# The annotation is typed ``Any`` because pydantic field annotations

423# include generics like ``Optional[int]`` / ``List[str]`` that are not

424# instances of ``type`` — so tightening this to ``type`` would reject

425# valid inputs.

426_EXTRA_UI_SETTINGS_FIELDS: Final[dict[str, tuple[object, FieldInfo]]] = {}

427

428# Settings OSS knows about as enterprise-gated. If a caller sends one of

429# these keys and no extension package has registered it, the PATCH

430# endpoint returns 403 instead of silently dropping the value, so the

431# client gets a clear signal that the feature requires LiteLLM Enterprise.

432_ENTERPRISE_ONLY_UI_SETTINGS: Final[set[str]] = {"enable_projects_ui"}

433

434# Memoized effective class; invalidated on registration.

435_EFFECTIVE_UI_SETTINGS_CLASS: type[UISettings] | None = None

436

437

438def register_extra_ui_setting(name: str, annotation: object, field: FieldInfo) -> None:

439 """Register an additional UI settings field contributed by an extension package.

440

441 ``field`` must be a ``FieldInfo`` instance — construct it directly

442 (e.g. ``FieldInfo(default=..., description=...)``) rather than via

443 the ``pydantic.Field`` factory, whose stub reports the default's

444 type instead of ``FieldInfo`` and trips mypy at the call site.

445 """

446 global _EFFECTIVE_UI_SETTINGS_CLASS

447 _EXTRA_UI_SETTINGS_FIELDS[name] = (annotation, field)

448 ALLOWED_UI_SETTINGS_FIELDS.add(name)

449 _EFFECTIVE_UI_SETTINGS_CLASS = None

450

451

452def _get_effective_ui_settings_class() -> type[UISettings]:

453 """Return UISettings with any extension-registered fields merged in.

454

455 Memoized — pydantic ``create_model`` runs metaclass + schema work

456 each call, so we cache until a new registration invalidates it.

457 """

458 global _EFFECTIVE_UI_SETTINGS_CLASS

459 if _EFFECTIVE_UI_SETTINGS_CLASS is not None:

460 return _EFFECTIVE_UI_SETTINGS_CLASS

461 if not _EXTRA_UI_SETTINGS_FIELDS: 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true

462 return UISettings

463 _EFFECTIVE_UI_SETTINGS_CLASS = create_model(

464 "EffectiveUISettings",

465 __base__=UISettings,

466 __doc__=UISettings.__doc__,

467 **_EXTRA_UI_SETTINGS_FIELDS,

468 )

469 return _EFFECTIVE_UI_SETTINGS_CLASS

470

471

472class MCPSemanticFilterSettings(BaseModel):

473 """Configuration for MCP Semantic Tool Filter"""

474

475 enabled: bool = Field(

476 default=False,

477 description="Enable semantic filtering of MCP tools based on query relevance",

478 )

479

480 embedding_model: str = Field(

481 default="text-embedding-3-small",

482 description="Embedding model to use for semantic similarity (e.g., 'text-embedding-3-small', 'text-embedding-ada-002')",

483 )

484

485 top_k: int = Field(

486 default=10,

487 description="Number of most relevant tools to return",

488 ge=1,

489 le=100,

490 )

491

492 similarity_threshold: float = Field(

493 default=0.3,

494 description="Minimum similarity score for tool inclusion (0.0 to 1.0, where 1.0 = exact match)",

495 ge=0.0,

496 le=1.0,

497 )

498

499

500class MCPSemanticFilterSettingsResponse(SettingsResponse):

501 """Response model for MCP semantic filter settings"""

502

503

504class MCPToolSearchSettingsResponse(SettingsResponse):

505 """Response model for native MCP tool search settings"""

506

507

508class WebSearchInterceptionSettings(BaseModel):

509 """Configuration for server-side web search interception"""

510

511 enabled: bool = Field(

512 default=False,

513 description="Serve web search tool calls from a configured search tool instead of passing them upstream",

514 )

515

516 enabled_providers: list[str] = Field(

517 default_factory=list,

518 description="LLM providers to intercept for (e.g. 'bedrock', 'vertex_ai'). Empty intercepts Bedrock only.",

519 )

520

521 search_tool_name: str | None = Field(

522 default=None,

523 description="Name of the configured search tool to run searches through. Empty uses the first one available.",

524 )

525

526 max_agentic_loops: int | None = Field(

527 default=None,

528 ge=1,

529 description="How many follow-up model calls one intercepted request may chain. Empty applies the default of 3.",

530 )

531

532

533class WebSearchInterceptionSettingsResponse(SettingsResponse):

534 """Response model for web search interception settings"""

535

536 active_on_this_pod: bool = Field(

537 default=False,

538 description=(

539 "Whether the process answering this request has the interception callback "

540 "registered. Read-only: it reports what is running here, while values.enabled "

541 "is the cluster-wide setting, and the two disagree while a pod is still "

542 "applying a change or failed to apply it."

543 ),

544 )

545

546

547def _with_websearch_enabled_resolved(config: Mapping[str, object]) -> dict[str, object]:

548 """

549 Answer with the stored flag when there is one, and only otherwise with what

550 this process is running.

551

552 A stored flag is the cluster's own answer, so it is the same on every pod and

553 is safe for the page to send back on save. Deriving the answer from this

554 process instead would report off on a pod that has not polled yet, and the

555 next save would persist that as a cluster-wide off. Without a stored flag the

556 only available answer is local: litellm_settings.callbacks activates

557 interception without storing one, and a write through the generic config

558 endpoint can drop the flag from a block that is still live. Reporting the

559 field default there would claim the feature is off while it serves.

560 """

561 from litellm.integrations.websearch_interception.handler import (

562 WebSearchInterceptionLogger,

563 )

564

565 litellm_settings: Final[Mapping[str, object]] = _as_settings_section(config.get("litellm_settings"))

566 stored: Final[Mapping[str, object]] = _as_settings_section(litellm_settings.get("websearch_interception_params"))

567 if "enabled" in stored:

568 return dict(config)

569

570 resolved: Final = {

571 **stored,

572 "enabled": bool(litellm.logging_callback_manager.get_custom_loggers_for_type(WebSearchInterceptionLogger)),

573 }

574 return {

575 **config,

576 "litellm_settings": {**litellm_settings, "websearch_interception_params": resolved},

577 }

578

579

580def _as_settings_section(value: object) -> Mapping[str, object]:

581 return cast("Mapping[str, object]", value) if isinstance(value, Mapping) else MappingProxyType({})

582

583

584@router.get(

585 "/get/allowed_ips",

586 tags=["Budget & Spend Tracking"],

587 dependencies=[Depends(user_api_key_auth)],

588 include_in_schema=False,

589)

590async def get_allowed_ips():

591 from litellm.proxy.proxy_server import general_settings

592

593 _allowed_ip: Final = general_settings.get("allowed_ips")

594 return {"data": _allowed_ip}

595

596

597def _store_allowed_ips(general_settings: MutableMapping[str, object], allowed_ips: Sequence[str]) -> None:

598 try:

599 general_settings["allowed_ips"] = list(allowed_ips) # mutable-ok: compared against the file's own list

600 except ConfigOwnedKeyError as owned:

601 raise HTTPException(

602 status_code=400,

603 detail={ # mutable-ok: HTTPException serializes its detail as json

604 "error": str(owned),

605 "keys": (owned.key,),

606 "section": owned.section,

607 "stored_database_value_ignored": owned.shadows_db_value,

608 },

609 ) from owned

610

611

612@router.post(

613 "/add/allowed_ip",

614 tags=["Budget & Spend Tracking"],

615 dependencies=[Depends(user_api_key_auth)],

616)

617async def add_allowed_ip(

618 ip_address: IPAddress,

619 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

620):

621 from litellm.proxy.proxy_server import (

622 create_config_audit_log,

623 general_settings,

624 prisma_client,

625 proxy_config,

626 store_model_in_db,

627 )

628

629 if prisma_client is None:

630 raise Exception("No DB Connected")

631

632 _allowed_ips: Final[Sequence[str]] = general_settings.get("allowed_ips") or ()

633 if ip_address.ip in _allowed_ips:

634 raise HTTPException(status_code=400, detail="IP address already exists")

635 _store_allowed_ips(general_settings, (*_allowed_ips, ip_address.ip))

636

637 if store_model_in_db is not True:

638 raise HTTPException(

639 status_code=500,

640 detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."},

641 )

642

643 # Load existing config

644 config: Final = await proxy_config.get_config()

645 verbose_proxy_logger.debug("Loaded config: %s", config)

646 if "general_settings" not in config:

647 config["general_settings"] = {}

648

649 if "allowed_ips" not in config["general_settings"]:

650 config["general_settings"]["allowed_ips"] = []

651

652 before_allowed_ips: Final = list(config["general_settings"]["allowed_ips"])

653 if ip_address.ip not in config["general_settings"]["allowed_ips"]:

654 config["general_settings"]["allowed_ips"].append(ip_address.ip)

655

656 await proxy_config.save_config(new_config=config)

657

658 asyncio.create_task(

659 create_config_audit_log(

660 param_name="general_settings",

661 action="updated",

662 before_value={"allowed_ips": before_allowed_ips},

663 after_value={"allowed_ips": config["general_settings"]["allowed_ips"]},

664 user_api_key_dict=user_api_key_dict,

665 )

666 )

667

668 return {

669 "message": f"IP {ip_address.ip} address added successfully",

670 "status": "success",

671 }

672

673

674@router.post(

675 "/delete/allowed_ip",

676 tags=["Budget & Spend Tracking"],

677 dependencies=[Depends(user_api_key_auth)],

678)

679async def delete_allowed_ip(

680 ip_address: IPAddress,

681 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

682):

683 from litellm.proxy.proxy_server import (

684 create_config_audit_log,

685 general_settings,

686 proxy_config,

687 )

688

689 _allowed_ips: Final[Sequence[str]] = general_settings.get("allowed_ips") or ()

690 if ip_address.ip not in _allowed_ips: 690 ↛ 692line 690 didn't jump to line 692 because the condition on line 690 was always true

691 raise HTTPException(status_code=404, detail="IP address not found")

692 _store_allowed_ips(general_settings, tuple(ip for ip in _allowed_ips if ip != ip_address.ip))

693

694 # Load existing config

695 config: Final = await proxy_config.get_config()

696 verbose_proxy_logger.debug("Loaded config: %s", config)

697 if "general_settings" not in config:

698 config["general_settings"] = {}

699

700 if "allowed_ips" not in config["general_settings"]:

701 config["general_settings"]["allowed_ips"] = []

702

703 before_allowed_ips: Final = list(config["general_settings"]["allowed_ips"])

704 if ip_address.ip in config["general_settings"]["allowed_ips"]:

705 config["general_settings"]["allowed_ips"].remove(ip_address.ip)

706

707 await proxy_config.save_config(new_config=config)

708

709 asyncio.create_task(

710 create_config_audit_log(

711 param_name="general_settings",

712 action="deleted",

713 before_value={"allowed_ips": before_allowed_ips},

714 after_value={"allowed_ips": config["general_settings"]["allowed_ips"]},

715 user_api_key_dict=user_api_key_dict,

716 )

717 )

718

719 return {"message": f"IP {ip_address.ip} deleted successfully", "status": "success"}

720

721

722def _resolve_non_null_variant(field_info: JsonSchemaNode) -> JsonSchemaNode:

723 """Pydantic v2 renders Optional fields as ``anyOf: [actual_type, null]``."""

724 if "anyOf" not in field_info:

725 return field_info

726 return next((variant for variant in field_info["anyOf"] if variant.get("type") != "null"), field_info)

727

728

729def _schema_items_entry(resolved: JsonSchemaNode, defs: Mapping[str, JsonSchemaNode]) -> "JsonSchemaItems | None":

730 """Items info (including enum values) for array fields, so the UI can render a multi-select dropdown."""

731 if "items" not in resolved:

732 return None

733 items: Final = resolved["items"]

734 if "$ref" not in items:

735 return items

736 ref_def: Final = defs.get(items["$ref"].split("/")[-1])

737 if ref_def is None or "enum" not in ref_def: 737 ↛ 738line 737 didn't jump to line 738 because the condition on line 737 was never true

738 return None

739 enum_items: Final[JsonSchemaItems] = {"enum": ref_def["enum"]}

740 return enum_items

741

742

743def _schema_property_entry(field_info: JsonSchemaNode, defs: Mapping[str, JsonSchemaNode]) -> JsonSchemaPropertyEntry:

744 resolved: Final = _resolve_non_null_variant(field_info)

745 items_entry: Final = _schema_items_entry(resolved, defs)

746 description: Final = field_info.get("description", "")

747 type_name: Final = resolved.get("type", "string")

748 if items_entry is None:

749 entry: Final[JsonSchemaPropertyEntry] = {"description": description, "type": type_name}

750 return entry

751 entry_with_items: Final[JsonSchemaPropertyEntry] = {

752 "description": description,

753 "type": type_name,

754 "items": items_entry,

755 }

756 return entry_with_items

757

758

759class _RootSchema(NamedTuple):

760 description: str

761 properties: Mapping[str, JsonSchemaNode]

762 nested_defs: Mapping[str, JsonSchemaNode]

763 defs: Mapping[str, JsonSchemaNode]

764

765

766def _root_schema(settings_class: type[BaseModel]) -> _RootSchema:

767 from pydantic import TypeAdapter

768

769 raw_schema: Final = TypeAdapter(settings_class).json_schema(by_alias=True)

770 return _RootSchema(

771 description=raw_schema.get("description", ""),

772 properties=raw_schema["properties"],

773 nested_defs=raw_schema.get("definitions", _EMPTY_SCHEMA_DEFS),

774 defs=raw_schema["$defs"] if "$defs" in raw_schema else raw_schema.get("definitions", _EMPTY_SCHEMA_DEFS),

775 )

776

777

778def _model_field_default(settings_class: type[BaseModel], field_name: str) -> object:

779 field_info: Final = settings_class.model_fields.get(field_name)

780 if field_info is None or field_info.default is PydanticUndefined: 780 ↛ 781line 780 didn't jump to line 781 because the condition on line 780 was never true

781 return None

782 return cast(object, field_info.default) # cast-ok: Pydantic field defaults are untyped

783

784

785def _ui_setting_source(

786 key: str,

787 value: object,

788 settings: SettingsStore,

789 settings_class: type[BaseModel],

790) -> FieldSource:

791 if key == ENABLE_PTU_COST_ATTRIBUTION_UI_SETTING:

792 configured_value: Final = get_secret(PTU_COST_ATTRIBUTION_ENV_VAR, None)

793 return "config" if configured_value is not None or value is True else "default"

794 if key == APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING:

795 return "config" if value is True else "default"

796 return source_for(settings, key, _model_field_default(settings_class, key))

797

798

799async def _get_settings_with_schema(

800 settings_key: str,

801 settings_class: type[BaseModel],

802 config: dict,

803) -> dict:

804 """

805 Common utility function to get settings with schema information.

806

807 Args:

808 settings_key: The key in litellm_settings to get

809 settings_class: The Pydantic class to use for schema

810 config: The config dictionary

811 """

812 litellm_settings: Final = config.get("litellm_settings", {}) or {}

813 settings_data: Final = litellm_settings.get(settings_key, {}) or {}

814

815 # Create the settings object

816 settings: Final = settings_class(**(settings_data))

817 # Get the schema

818 root_schema: Final = _root_schema(settings_class)

819

820 # Convert to dict for response

821 settings_dict: Final = settings.model_dump()

822

823 # Add descriptions to the response

824 schema_properties_out: Final[Mapping[str, JsonSchemaPropertyEntry]] = {

825 field_name: _schema_property_entry(field_info, root_schema.defs)

826 for field_name, field_info in root_schema.properties.items()

827 }

828

829 # Add nested object descriptions

830 nested_defs_out: Final[Mapping[str, Mapping[str, object]]] = {

831 def_name: {

832 "description": def_schema.get("description", ""),

833 "properties": {

834 prop_name: {"description": prop_info.get("description", "")}

835 for prop_name, prop_info in def_schema.get("properties", {}).items()

836 },

837 }

838 for def_name, def_schema in root_schema.nested_defs.items()

839 }

840

841 return {

842 "values": settings_dict,

843 "field_schema": {

844 "description": root_schema.description,

845 "properties": schema_properties_out,

846 **nested_defs_out,

847 },

848 }

849

850

851@router.get(

852 "/get/internal_user_settings",

853 tags=["SSO Settings"],

854 dependencies=[Depends(user_api_key_auth)],

855 response_model=InternalUserSettingsResponse,

856)

857async def get_internal_user_settings():

858 """

859 Get all SSO settings from the litellm_settings configuration.

860 Returns a structured object with values and descriptions for UI display.

861 """

862 from litellm.proxy.proxy_server import proxy_config

863

864 # Load existing config

865 config: Final = await proxy_config.get_config()

866

867 return await _get_settings_with_schema(

868 settings_key="default_internal_user_params",

869 settings_class=DefaultInternalUserParams,

870 config=config,

871 )

872

873

874@router.get(

875 "/get/default_team_settings",

876 tags=["SSO Settings"],

877 dependencies=[Depends(user_api_key_auth)],

878 response_model=DefaultTeamSettingsResponse,

879)

880async def get_default_team_settings():

881 """

882 Get the default team parameters (litellm_settings.default_team_params).

883 Returns a structured object with values and descriptions for UI display.

884 """

885 from litellm.proxy.proxy_server import proxy_config

886

887 # Load existing config

888 config: Final = await proxy_config.get_config()

889

890 return await _get_settings_with_schema(

891 settings_key="default_team_params",

892 settings_class=DefaultTeamSSOParams,

893 config=config,

894 )

895

896

897def _default_team_ids(teams: list[str] | list[NewUserRequestTeam]) -> tuple[str, ...]:

898 return tuple(team if isinstance(team, str) else team.team_id for team in teams)

899

900

901async def _validate_default_teams_exist(teams: list[str] | list[NewUserRequestTeam]) -> None:

902 """Reject default teams that cannot be assigned.

903

904 New users are added to these teams long after the settings are saved, and that

905 consume path swallows the resulting 404, so an unknown team id would silently

906 drop every future user's team assignment unless it is caught here.

907 """

908 team_ids: Final = _default_team_ids(teams)

909 if not team_ids:

910 return

911

912 duplicate_ids: Final = tuple(team_id for team_id, count in Counter(team_ids).items() if count > 1)

913 if duplicate_ids:

914 raise HTTPException(

915 status_code=400,

916 detail={

917 "error": f"Duplicate default team id(s): {', '.join(duplicate_ids)}. List each default team only once."

918 },

919 )

920

921 from litellm.proxy.proxy_server import prisma_client

922

923 if prisma_client is None: 923 ↛ 924line 923 didn't jump to line 924 because the condition on line 923 was never true

924 raise HTTPException(

925 status_code=500,

926 detail={"error": "Database not connected. Please connect a database."},

927 )

928

929 existing_teams: Final = await TeamRepository(prisma_client).find_many(where={"team_id": {"in": list(team_ids)}})

930 existing_team_ids: Final = {team.team_id for team in existing_teams}

931 missing_ids: Final = tuple(team_id for team_id in team_ids if team_id not in existing_team_ids)

932 if missing_ids: 932 ↛ exitline 932 didn't return from function '_validate_default_teams_exist' because the condition on line 932 was always true

933 raise HTTPException(

934 status_code=400,

935 detail={

936 "error": f"Team(s) not found: {', '.join(missing_ids)}. "

937 "A team must exist before it can be set as a default team for new users."

938 },

939 )

940

941

942async def _validate_default_organization_exists(organization_id: str) -> None:

943 """Reject a default organization that cannot be assigned.

944

945 Teams are created from these settings long after they are saved, and an unknown

946 organization id would fail every future team creation instead of here, where the

947 admin who typed it can still fix it.

948 """

949 from litellm.proxy.proxy_server import prisma_client

950

951 if prisma_client is None: 951 ↛ 952line 951 didn't jump to line 952 because the condition on line 951 was never true

952 raise HTTPException(

953 status_code=500,

954 detail={ # mutable-ok: HTTPException detail must be a plain dict for FastAPI JSON serialization

955 "error": "Database not connected. Please connect a database."

956 },

957 )

958

959 organization_exists: Final = await OrganizationRepository(prisma_client).exists(

960 organization_id, id_field="organization_id"

961 )

962 if not organization_exists: 962 ↛ exitline 962 didn't return from function '_validate_default_organization_exists' because the condition on line 962 was always true

963 raise HTTPException(

964 status_code=400,

965 detail={ # mutable-ok: HTTPException detail must be a plain dict for FastAPI JSON serialization

966 "error": f"Organization not found: {organization_id}. "

967 "An organization must exist before it can be set as the default organization for new teams."

968 },

969 )

970

971

972async def update_default_team_member_budget(teams: list[NewUserRequestTeam], user_api_key_dict: UserAPIKeyAuth):

973 """

974 1. Update the max member budget for the team

975 """

976 from fastapi import Request

977

978 from litellm.proxy.management_endpoints.team_endpoints import update_team

979

980 for team in teams: 980 ↛ 981line 980 didn't jump to line 981 because the loop on line 980 never started

981 team_id = team.team_id

982 max_budget_in_team = team.max_budget_in_team

983 try:

984 await update_team(

985 data=UpdateTeamRequest(

986 team_id=team_id,

987 team_member_budget=max_budget_in_team,

988 ),

989 user_api_key_dict=user_api_key_dict,

990 http_request=Request(scope={"type": "http"}),

991 )

992 except Exception as e:

993 verbose_proxy_logger.info(

994 "Error updating team %s with team member budget %s with error: %s, skipping..",

995 team_id,

996 max_budget_in_team,

997 e,

998 )

999 continue

1000

1001

1002async def _update_litellm_setting(

1003 settings: (

1004 DefaultInternalUserParams

1005 | DefaultTeamSSOParams

1006 | MCPSemanticFilterSettings

1007 | MCPToolSearchSettings

1008 | WebSearchInterceptionSettings

1009 ),

1010 settings_key: str,

1011 success_message: str,

1012 user_api_key_dict: UserAPIKeyAuth,

1013):

1014 """

1015 Common utility function to update `litellm_settings` in both memory and config.

1016

1017 Args:

1018 settings: The settings object to update

1019 settings_key: The key in litellm_settings to update

1020 success_message: Message to return on success

1021 user_api_key_dict: The acting admin, recorded as the audit-log actor.

1022 """

1023 from litellm.proxy.proxy_server import (

1024 create_config_audit_log,

1025 proxy_config,

1026 store_model_in_db,

1027 )

1028

1029 if store_model_in_db is not True: 1029 ↛ 1030line 1029 didn't jump to line 1030 because the condition on line 1029 was never true

1030 raise HTTPException(

1031 status_code=500,

1032 detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."},

1033 )

1034

1035 in_memory_var: Final = settings.model_dump(mode="json", exclude_none=True)

1036

1037 # Load existing config first, then set in-memory value after,

1038 # because get_config() may overwrite litellm.<key> with stale DB values

1039 # via LITELLM_SETTINGS_SAFE_DB_OVERRIDES.

1040 config: Final = await proxy_config.get_config()

1041 before_value: Final = config.get("litellm_settings", {}).get(settings_key)

1042

1043 # Update the in-memory settings (after get_config to avoid stale override)

1044 setattr(litellm, settings_key, in_memory_var)

1045

1046 # Update config with new settings

1047 if "litellm_settings" not in config: 1047 ↛ 1048line 1047 didn't jump to line 1048 because the condition on line 1047 was never true

1048 config["litellm_settings"] = {}

1049

1050 config["litellm_settings"][settings_key] = in_memory_var

1051

1052 # Save the updated config

1053 await proxy_config.save_config(new_config=config)

1054

1055 # Fire-and-forget so an audit-log failure (transient DB blip, etc.)

1056 # never surfaces as a 500 after save_config has already committed,

1057 # matching the create_object_audit_log pattern used elsewhere

1058 # (e.g. model_management_endpoints).

1059 asyncio.create_task(

1060 create_config_audit_log(

1061 param_name=settings_key,

1062 action="updated",

1063 before_value=before_value,

1064 after_value=in_memory_var,

1065 user_api_key_dict=user_api_key_dict,

1066 )

1067 )

1068

1069 return {

1070 "message": success_message,

1071 "status": "success",

1072 "settings": in_memory_var,

1073 }

1074

1075

1076@router.patch(

1077 "/update/internal_user_settings",

1078 tags=["SSO Settings"],

1079 dependencies=[Depends(user_api_key_auth)],

1080)

1081async def update_internal_user_settings(

1082 settings: DefaultInternalUserParams,

1083 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1084):

1085 """

1086 Update the default internal user parameters for SSO users.

1087 These settings will be applied to new users who sign in via SSO.

1088 """

1089 if settings.teams is not None:

1090 await _validate_default_teams_exist(settings.teams)

1091

1092 if settings.teams is not None and all(isinstance(team, NewUserRequestTeam) for team in settings.teams):

1093 await update_default_team_member_budget(

1094 settings.teams,

1095 user_api_key_dict=user_api_key_dict,

1096 )

1097

1098 return await _update_litellm_setting(

1099 settings=settings,

1100 settings_key="default_internal_user_params",

1101 success_message="Internal user settings updated successfully",

1102 user_api_key_dict=user_api_key_dict,

1103 )

1104

1105

1106@router.patch(

1107 "/update/default_team_settings",

1108 tags=["SSO Settings"],

1109 dependencies=[Depends(user_api_key_auth)],

1110)

1111async def update_default_team_settings(

1112 settings: DefaultTeamSSOParams,

1113 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1114):

1115 """

1116 Update the default team parameters (litellm_settings.default_team_params).

1117 Applied to every new team for fields not explicitly provided in the create request;

1118 `models` only applies to teams automatically created via SSO Groups.

1119 """

1120 if settings.organization_id is not None:

1121 await _validate_default_organization_exists(settings.organization_id)

1122

1123 return await _update_litellm_setting(

1124 settings=settings,

1125 settings_key="default_team_params",

1126 success_message="Default team settings updated successfully",

1127 user_api_key_dict=user_api_key_dict,

1128 )

1129

1130

1131@router.get(

1132 "/get/sso_settings",

1133 tags=["SSO Settings"],

1134 dependencies=[Depends(user_api_key_auth)],

1135 response_model=SSOSettingsResponse,

1136)

1137async def get_sso_settings():

1138 """

1139 Get all SSO configuration settings from the dedicated SSO table.

1140 Returns a structured object with values and descriptions for UI display.

1141 """

1142

1143 from litellm.proxy.proxy_server import prisma_client

1144

1145 if prisma_client is None: 1145 ↛ 1146line 1145 didn't jump to line 1146 because the condition on line 1145 was never true

1146 raise HTTPException(

1147 status_code=500,

1148 detail={"error": "Database not connected. Please connect a database."},

1149 )

1150

1151 # Resolve the effective SSO config: the stored row wins, else the process

1152 # environment, else each field's default. Unlike the legacy read path this

1153 # does not write os.environ; a GET has no business mutating the environment.

1154 sso_db_record: Final = await _sso_settings_mapping_db(SSOConfigRepository(prisma_client)).find_unique(

1155 where={"id": "sso_config"}

1156 )

1157 sso_db_settings: Final = dict(sso_db_record.sso_settings) if sso_db_record and sso_db_record.sso_settings else None

1158 resolved: Final = resolve_sso_config(sso_db_settings, os.environ)

1159

1160 # Get the schema for UI display

1161 root_schema: Final = _root_schema(SSOConfig)

1162

1163 # Convert to dict for response, masking OAuth client secrets so plaintext

1164 # is never sent to the UI.

1165 sso_dict: Final = mask_sensitive_keys(resolved.config.model_dump(), set(SSO_SECRET_FIELDS))

1166

1167 # Add descriptions to the response

1168 schema_properties_out: Final[Mapping[str, Mapping[str, str]]] = {

1169 field_name: {

1170 "description": field_info.get("description", ""),

1171 "type": field_info.get("type", "string"),

1172 }

1173 for field_name, field_info in root_schema.properties.items()

1174 }

1175

1176 return {

1177 "values": sso_dict,

1178 "provenance": resolved.provenance,

1179 "field_schema": {

1180 "description": root_schema.description,

1181 "properties": schema_properties_out,

1182 },

1183 }

1184

1185

1186@router.patch(

1187 "/update/sso_settings",

1188 tags=["SSO Settings"],

1189 dependencies=[Depends(user_api_key_auth)],

1190)

1191async def update_sso_settings(

1192 sso_config: SSOConfig,

1193 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1194):

1195 """

1196 Update SSO configuration by saving to the dedicated SSO table.

1197 """

1198 import json

1199 import os

1200

1201 from litellm.proxy.proxy_server import (

1202 create_config_audit_log,

1203 prisma_client,

1204 proxy_config,

1205 store_model_in_db,

1206 )

1207

1208 if prisma_client is None: 1208 ↛ 1209line 1208 didn't jump to line 1209 because the condition on line 1208 was never true

1209 raise HTTPException(

1210 status_code=500,

1211 detail={"error": "Database not connected. Please connect a database."},

1212 )

1213

1214 if store_model_in_db is not True: 1214 ↛ 1215line 1214 didn't jump to line 1215 because the condition on line 1214 was never true

1215 raise HTTPException(

1216 status_code=500,

1217 detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."},

1218 )

1219

1220 # Read the existing SSO row first so the audit log captures a real

1221 # before/after diff. Stored values are encrypted; decrypt them so the

1222 # before-snapshot has the same shape as after_value, and rely on

1223 # create_config_audit_log's secret-name redaction to mask the

1224 # *_client_secret fields before the audit row is written.

1225 existing_sso_record: Final = await _stored_sso_settings_db(SSOConfigRepository(prisma_client)).find_unique(

1226 where={"id": "sso_config"}

1227 )

1228 before_sso_data: dict[str, JsonValue] | None = None

1229 if existing_sso_record and existing_sso_record.sso_settings:

1230 stored = existing_sso_record.sso_settings

1231 if isinstance(stored, str): 1231 ↛ 1232line 1231 didn't jump to line 1232 because the condition on line 1231 was never true

1232 stored = json.loads(stored)

1233 if isinstance(stored, dict): 1233 ↛ 1237line 1233 didn't jump to line 1237 because the condition on line 1233 was always true

1234 before_sso_data = proxy_config._decrypt_db_variables(stored)

1235

1236 # Load existing config

1237 config: Final = await proxy_config.get_config()

1238

1239 # Update config with new environment variables

1240 if "environment_variables" not in config: 1240 ↛ 1244line 1240 didn't jump to line 1244 because the condition on line 1240 was always true

1241 config["environment_variables"] = {}

1242

1243 # Update general_settings for user_email (admin email)

1244 if "general_settings" not in config: 1244 ↛ 1245line 1244 didn't jump to line 1245 because the condition on line 1244 was never true

1245 config["general_settings"] = {}

1246

1247 # Update environment variables in config and in memory

1248 sso_data: Final = sso_config.model_dump()

1249 for field_name, value in sso_data.items():

1250 if field_name in SSO_FIELD_ENV_VARS:

1251 env_var_name = SSO_FIELD_ENV_VARS[field_name]

1252 if value:

1253 os.environ[env_var_name] = value

1254 else:

1255 # Clear environment variable if value is null/empty

1256 os.environ.pop(env_var_name, None)

1257

1258 encrypted_sso_data: Final = proxy_config._encrypt_env_variables(environment_variables=sso_data)

1259

1260 # Save to dedicated SSO table

1261 await _stored_sso_settings_db(SSOConfigRepository(prisma_client)).upsert(

1262 where={"id": "sso_config"},

1263 data={

1264 "create": {

1265 "id": "sso_config",

1266 "sso_settings": json.dumps(encrypted_sso_data),

1267 },

1268 "update": {

1269 "sso_settings": json.dumps(encrypted_sso_data),

1270 },

1271 },

1272 )

1273

1274 asyncio.create_task(

1275 create_config_audit_log(

1276 param_name="sso_config",

1277 action="updated",

1278 before_value=before_sso_data,

1279 after_value=sso_data,

1280 user_api_key_dict=user_api_key_dict,

1281 table_name=LitellmTableNames.SSO_CONFIG_TABLE_NAME,

1282 )

1283 )

1284

1285 # Remove SSO-related env vars from config.environment_variables

1286 try:

1287 env_var_entry: Final = await _config_param_db(ConfigRepository(prisma_client)).find_unique(

1288 where={"param_name": "environment_variables"}

1289 )

1290

1291 # If no environment_variables entry exists, nothing to clean up

1292 if env_var_entry is not None:

1293 if env_var_entry.param_value is not None: 1293 ↛ 1299line 1293 didn't jump to line 1299 because the condition on line 1293 was always true

1294 if isinstance(env_var_entry.param_value, str): 1294 ↛ 1295line 1294 didn't jump to line 1295 because the condition on line 1294 was never true

1295 environment_variables: Mapping[str, object] = json.loads(env_var_entry.param_value)

1296 else:

1297 environment_variables = dict(env_var_entry.param_value)

1298 else:

1299 environment_variables = {}

1300

1301 env_vars_to_remove: Final = set(SSO_FIELD_ENV_VARS.values())

1302 filtered_env_vars: Final = {

1303 key: value for key, value in environment_variables.items() if key not in env_vars_to_remove

1304 }

1305

1306 await _config_param_db(ConfigRepository(prisma_client)).update(

1307 where={"param_name": "environment_variables"},

1308 data={

1309 "param_value": json.dumps(filtered_env_vars, default=str),

1310 },

1311 )

1312 await invalidate_config_param("environment_variables")

1313 except Exception as e:

1314 raise HTTPException(

1315 status_code=500,

1316 detail={"error": f"Error updating environment_variables: {e}"},

1317 )

1318

1319 return {

1320 "message": "SSO settings updated successfully",

1321 "status": "success",

1322 "settings": sso_data,

1323 }

1324

1325

1326@router.get(

1327 "/get/ui_theme_settings",

1328 tags=["UI Theme Settings"],

1329 response_model=UIThemeSettingsResponse,

1330)

1331async def get_ui_theme_settings():

1332 """

1333 Get UI theme configuration from the litellm_settings.

1334 Returns current logo settings for UI customization.

1335

1336 Note: This endpoint is public (no authentication required) so all users can see custom branding.

1337 Only the /update/ui_theme_settings endpoint requires authentication for admins to change settings.

1338 """

1339 from litellm.proxy.proxy_server import proxy_config

1340

1341 # Load existing config

1342 config: Final = await proxy_config.get_config()

1343

1344 result: Final = await _get_settings_with_schema(

1345 settings_key="ui_theme_config",

1346 settings_class=UIThemeConfig,

1347 config=config,

1348 )

1349

1350 stored_values: Final = result.get("values", {})

1351 result["values"] = {

1352 **stored_values,

1353 **{field: _resolve_ui_theme_field(stored_values, field) for field in _UI_THEME_FIELD_ENV_VARS},

1354 }

1355 return result

1356

1357

1358def _validate_public_image_url(value: str | None, field_name: str) -> None:

1359 """

1360 Reject anything that isn't a plain http(s) URL with a host. This value is

1361 later served via the unauthenticated /get_image endpoint, so local paths

1362 like "/etc/passwd" or "file://..." must not be accepted.

1363 """

1364 if value is None:

1365 return

1366 if not isinstance(value, str) or not value.strip():

1367 return

1368 parsed: Final = urlparse(value.strip())

1369 if parsed.scheme not in ("http", "https") or not parsed.netloc: 1369 ↛ exitline 1369 didn't return from function '_validate_public_image_url' because the condition on line 1369 was always true

1370 raise HTTPException(

1371 status_code=400,

1372 detail={

1373 "error": (

1374 f"Invalid {field_name}: must be an http(s) URL with a host. "

1375 "Local filesystem paths and non-http schemes are not allowed."

1376 )

1377 },

1378 )

1379

1380

1381@router.patch(

1382 "/update/ui_theme_settings",

1383 tags=["UI Theme Settings"],

1384 dependencies=[Depends(user_api_key_auth)],

1385)

1386async def update_ui_theme_settings(

1387 theme_config: UIThemeConfig,

1388 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1389):

1390 """

1391 Update UI theme configuration.

1392 Updates logo settings for the admin UI.

1393 """

1394 import os

1395

1396 from litellm.proxy.proxy_server import (

1397 create_config_audit_log,

1398 proxy_config,

1399 store_model_in_db,

1400 )

1401

1402 _validate_public_image_url(theme_config.logo_url, "logo_url")

1403 _validate_public_image_url(theme_config.logo_url_dark, "logo_url_dark")

1404 _validate_public_image_url(theme_config.favicon_url, "favicon_url")

1405

1406 if store_model_in_db is not True: 1406 ↛ 1407line 1406 didn't jump to line 1407 because the condition on line 1406 was never true

1407 raise HTTPException(

1408 status_code=500,

1409 detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."},

1410 )

1411

1412 # Load existing config

1413 config: Final = await proxy_config.get_config()

1414 before_theme: Final = config.get("litellm_settings", {}).get("ui_theme_config")

1415

1416 # Convert theme config to dict

1417 theme_data: Final = theme_config.model_dump(exclude_none=True)

1418

1419 # Store UI theme config in litellm_settings (where it's retrieved from)

1420 if "litellm_settings" not in config: 1420 ↛ 1421line 1420 didn't jump to line 1421 because the condition on line 1420 was never true

1421 config["litellm_settings"] = {}

1422 config["litellm_settings"]["ui_theme_config"] = theme_data

1423

1424 # The vars below are the only environment variables this endpoint owns, and

1425 # they must stay in step with _UI_THEME_FIELD_ENV_VARS. A non-empty value

1426 # sets the var; an empty or missing one clears it back to the default. Apply

1427 # to the live process immediately, then persist only those keys so an

1428 # unrelated env var (a YAML/OS value merged in by get_config) is never

1429 # snapshotted into the DB.

1430 def _clean(url: str | None) -> str | None:

1431 return url if url is not None and url.strip() else None

1432

1433 env_updates: Final[dict[str, str | None]] = {

1434 "UI_LOGO_PATH": _clean(theme_config.logo_url),

1435 "UI_LOGO_PATH_DARK": _clean(theme_config.logo_url_dark),

1436 "LITELLM_FAVICON_URL": _clean(theme_config.favicon_url),

1437 }

1438 for env_key, env_value in env_updates.items():

1439 if env_value is not None: 1439 ↛ 1440line 1439 didn't jump to line 1440 because the condition on line 1439 was never true

1440 os.environ[env_key] = env_value

1441 else:

1442 os.environ.pop(env_key, None)

1443

1444 # Persist the theme config (litellm_settings). save_config defaults to

1445 # include_env_vars=False, so it does not snapshot environment_variables.

1446 await proxy_config.save_config(new_config=config)

1447 # Persist only the two owned env vars, merged against the existing DB row.

1448 await proxy_config.save_environment_variables(env_updates)

1449

1450 asyncio.create_task(

1451 create_config_audit_log(

1452 param_name="ui_theme_config",

1453 action="updated",

1454 before_value=before_theme,

1455 after_value=theme_data,

1456 user_api_key_dict=user_api_key_dict,

1457 )

1458 )

1459

1460 return {

1461 "message": "UI theme settings updated successfully.",

1462 "status": "success",

1463 "theme_config": theme_data,

1464 }

1465

1466

1467@router.get(

1468 "/get/mcp_semantic_filter_settings",

1469 tags=["Settings"],

1470 dependencies=[Depends(user_api_key_auth)],

1471 response_model=MCPSemanticFilterSettingsResponse,

1472)

1473async def get_mcp_semantic_filter_settings(

1474 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1475):

1476 """

1477 Get MCP semantic filter configuration.

1478 Returns current settings for semantic tool filtering.

1479 """

1480 from litellm.proxy.proxy_server import prisma_client, proxy_config

1481

1482 if prisma_client is None: 1482 ↛ 1483line 1482 didn't jump to line 1483 because the condition on line 1482 was never true

1483 raise HTTPException(

1484 status_code=500,

1485 detail={"error": "Database not connected. Please connect a database."},

1486 )

1487

1488 config: Final = await proxy_config.get_config()

1489

1490 return await _get_settings_with_schema(

1491 settings_key="mcp_semantic_tool_filter",

1492 settings_class=MCPSemanticFilterSettings,

1493 config=config,

1494 )

1495

1496

1497@router.patch(

1498 "/update/mcp_semantic_filter_settings",

1499 tags=["Settings"],

1500 dependencies=[Depends(user_api_key_auth)],

1501)

1502async def update_mcp_semantic_filter_settings(

1503 settings: MCPSemanticFilterSettings,

1504 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1505):

1506 """

1507 Update MCP semantic filter settings in database.

1508 Settings will be picked up by all pods within approximately 10 seconds via background polling.

1509 """

1510 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1510 ↛ 1511line 1510 didn't jump to line 1511 because the condition on line 1510 was never true

1511 raise HTTPException(

1512 status_code=403,

1513 detail="Only proxy admins can update MCP semantic filter settings.",

1514 )

1515

1516 result: Final = await _update_litellm_setting(

1517 settings=settings,

1518 settings_key="mcp_semantic_tool_filter",

1519 success_message="MCP Semantic Filter settings updated successfully. Changes will be applied across all pods within 10 seconds.",

1520 user_api_key_dict=user_api_key_dict,

1521 )

1522 try:

1523 from litellm.proxy.proxy_server import prisma_client, proxy_config

1524

1525 if prisma_client is not None: 1525 ↛ 1530line 1525 didn't jump to line 1530 because the condition on line 1525 was always true

1526 await proxy_config._init_semantic_filter_settings_in_db(prisma_client=prisma_client)

1527 except Exception as e:

1528 verbose_proxy_logger.warning("Failed to reinitialize MCP semantic filter settings immediately: %s", e)

1529

1530 return result

1531

1532

1533@router.get(

1534 "/get/websearch_interception_settings",

1535 tags=["Settings"],

1536 dependencies=[Depends(user_api_key_auth)],

1537 response_model=WebSearchInterceptionSettingsResponse,

1538)

1539async def get_websearch_interception_settings(

1540 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],

1541):

1542 """

1543 Get web search interception configuration.

1544

1545 Returns the current settings plus their schema, for the Admin UI to render.

1546 """

1547 from litellm.proxy.proxy_server import prisma_client, proxy_config

1548

1549 if prisma_client is None: 1549 ↛ 1550line 1549 didn't jump to line 1550 because the condition on line 1549 was never true

1550 raise HTTPException(

1551 status_code=500,

1552 detail={"error": "Database not connected. Please connect a database."},

1553 )

1554

1555 config: Final = await proxy_config.get_config()

1556

1557 from litellm.integrations.websearch_interception.handler import (

1558 WebSearchInterceptionLogger,

1559 )

1560

1561 settings: Final = await _get_settings_with_schema(

1562 settings_key="websearch_interception_params",

1563 settings_class=WebSearchInterceptionSettings,

1564 config=_with_websearch_enabled_resolved(config),

1565 )

1566 return WebSearchInterceptionSettingsResponse(

1567 values=settings["values"],

1568 field_schema=settings["field_schema"],

1569 active_on_this_pod=bool(

1570 litellm.logging_callback_manager.get_custom_loggers_for_type(WebSearchInterceptionLogger)

1571 ),

1572 )

1573

1574

1575@router.patch(

1576 "/update/websearch_interception_settings",

1577 tags=["Settings"],

1578 dependencies=[Depends(user_api_key_auth)],

1579)

1580async def update_websearch_interception_settings(

1581 settings: WebSearchInterceptionSettings,

1582 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],

1583):

1584 """

1585 Update web search interception settings in database.

1586

1587 Settings will be picked up by all pods within approximately 10 seconds via background polling.

1588 """

1589 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1589 ↛ 1590line 1589 didn't jump to line 1590 because the condition on line 1589 was never true

1590 raise HTTPException(

1591 status_code=403,

1592 detail="Only proxy admins can update web search interception settings.",

1593 )

1594

1595 result: Final = await _update_litellm_setting(

1596 settings=settings,

1597 settings_key="websearch_interception_params",

1598 success_message=(

1599 "Web search interception settings updated successfully. "

1600 "Changes will be applied across all pods within 10 seconds."

1601 ),

1602 user_api_key_dict=user_api_key_dict,

1603 )

1604 try:

1605 from litellm.proxy.proxy_server import prisma_client, proxy_config

1606

1607 if prisma_client is not None: 1607 ↛ 1612line 1607 didn't jump to line 1612 because the condition on line 1607 was always true

1608 await proxy_config.init_websearch_interception_settings_in_db(prisma_client=prisma_client)

1609 except Exception as e:

1610 verbose_proxy_logger.warning("Failed to reinitialize web search interception settings immediately: %s", e)

1611

1612 return result

1613

1614

1615@router.get(

1616 "/get/mcp_tool_search_settings",

1617 tags=["Settings"], # mutable-ok: FastAPI's route decorator only accepts a list

1618 dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI's route decorator only accepts a list

1619 response_model=MCPToolSearchSettingsResponse,

1620)

1621async def get_mcp_tool_search_settings(

1622 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1623) -> Mapping[str, object]:

1624 """

1625 Get the `litellm_settings.mcp_tool_search` configuration used by the native `mcp_tool_search` virtual tool.

1626 """

1627 from litellm.proxy.proxy_server import prisma_client, proxy_config

1628

1629 if prisma_client is None: 1629 ↛ 1630line 1629 didn't jump to line 1630 because the condition on line 1629 was never true

1630 raise HTTPException(status_code=500, detail="Database not connected. Please connect a database.")

1631

1632 config: Final = await proxy_config.get_config()

1633

1634 return await _get_settings_with_schema(

1635 settings_key=MCP_TOOL_SEARCH_SETTINGS_KEY,

1636 settings_class=MCPToolSearchSettings,

1637 config=config,

1638 )

1639

1640

1641@router.patch(

1642 "/update/mcp_tool_search_settings",

1643 tags=["Settings"], # mutable-ok: FastAPI's route decorator only accepts a list

1644 dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI's route decorator only accepts a list

1645)

1646async def update_mcp_tool_search_settings(

1647 settings: MCPToolSearchSettings,

1648 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1649) -> Mapping[str, object]:

1650 """

1651 Update `litellm_settings.mcp_tool_search` in the database.

1652 Settings will be picked up by all pods within approximately 10 seconds via background polling.

1653 """

1654 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1654 ↛ 1655line 1654 didn't jump to line 1655 because the condition on line 1654 was never true

1655 raise HTTPException(

1656 status_code=403,

1657 detail="Only proxy admins can update MCP tool search settings.",

1658 )

1659

1660 return await _update_litellm_setting(

1661 settings=settings,

1662 settings_key=MCP_TOOL_SEARCH_SETTINGS_KEY,

1663 success_message="MCP tool search settings updated successfully. Changes will be applied across all pods within 10 seconds.",

1664 user_api_key_dict=user_api_key_dict,

1665 )

1666

1667

1668UI_SETTINGS_CACHE_KEY: Final = "ui_settings:settings_dict"

1669UI_SETTINGS_CACHE_TTL: Final = 600 # 10 minutes

1670

1671

1672async def get_ui_settings_cached() -> dict[str, JsonValue]:

1673 """

1674 Return the persisted UI settings dict, using DualCache for reads.

1675

1676 Cache hit → return cached dict immediately.

1677 Cache miss → read from DB, populate cache, return dict.

1678 """

1679 from litellm.proxy.proxy_server import prisma_client, user_api_key_cache

1680

1681 # 1. Try cache

1682 cached: Final = await user_api_key_cache.async_get_cache(key=UI_SETTINGS_CACHE_KEY)

1683 if cached is not None and isinstance(cached, dict):

1684 return cached

1685

1686 # 2. Fallback to DB

1687 if prisma_client is None:

1688 return {}

1689

1690 db_record: Final = await _ui_settings_db(UISettingsRepository(prisma_client)).find_unique(

1691 where={"id": "ui_settings"}

1692 )

1693 ui_settings: dict[str, JsonValue] = {}

1694 if db_record and db_record.ui_settings:

1695 raw: Final = db_record.ui_settings

1696 ui_settings = json.loads(raw) if isinstance(raw, str) else dict(raw)

1697

1698 # Sanitize

1699 ui_settings = {k: v for k, v in ui_settings.items() if k in ALLOWED_UI_SETTINGS_FIELDS}

1700

1701 # 3. Populate cache with TTL

1702 await user_api_key_cache.async_set_cache(key=UI_SETTINGS_CACHE_KEY, value=ui_settings, ttl=UI_SETTINGS_CACHE_TTL)

1703

1704 return ui_settings

1705

1706

1707_UI_SETTINGS_OBJECT: Final = TypeAdapter(dict[str, JsonValue])

1708

1709

1710def apply_runtime_general_settings_flags(ui_settings: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:

1711 """Copy the UI settings that gate runtime behavior into ``general_settings``. Returns what was applied."""

1712 from litellm.proxy.config_resolvers import SettingsStore

1713 from litellm.proxy.proxy_server import general_settings

1714

1715 flags: Final = {k: ui_settings[k] for k in _RUNTIME_GENERAL_SETTINGS_FLAGS if k in ui_settings}

1716 if isinstance(general_settings, SettingsStore): 1716 ↛ 1718line 1716 didn't jump to line 1718 because the condition on line 1716 was always true

1717 general_settings.apply_db_row("ui_settings", flags)

1718 elif flags:

1719 general_settings.update(flags)

1720 return MappingProxyType(flags)

1721

1722

1723async def sync_ui_settings_to_general_settings(prisma_client: object) -> Mapping[str, JsonValue]:

1724 """Re-read the persisted UI settings and apply the runtime flags to ``general_settings``.

1725

1726 Runs on startup and on every periodic config reload: the PATCH handler only updates the pod

1727 that served it, so every other pod needs its own read to pick up a change without a restart.

1728 Never raises. A read that fails leaves this pod on the flags it already had.

1729 """

1730 try:

1731 db_record: Final = await _ui_settings_db(UISettingsRepository(prisma_client)).find_unique(

1732 where={"id": "ui_settings"}

1733 )

1734 stored: Final = (db_record.ui_settings if db_record else None) or "{}"

1735 parsed: Final = (

1736 _UI_SETTINGS_OBJECT.validate_json(stored)

1737 if isinstance(stored, str)

1738 else _UI_SETTINGS_OBJECT.validate_python(stored)

1739 )

1740 except Exception as e:

1741 verbose_proxy_logger.warning("Could not refresh UI settings from the database: %s", e)

1742 return MappingProxyType({})

1743 return apply_runtime_general_settings_flags(parsed)

1744

1745

1746@router.get(

1747 "/get/ui_settings",

1748 tags=["UI Settings"],

1749 response_model=UISettingsResponse,

1750)

1751async def get_ui_settings():

1752 """

1753 Get UI-specific configuration flags.

1754 All authenticated users can fetch these settings for client-side behavior.

1755 """

1756 from litellm.proxy.proxy_server import prisma_client, proxy_config

1757

1758 if prisma_client is None: 1758 ↛ 1759line 1758 didn't jump to line 1759 because the condition on line 1758 was never true

1759 raise HTTPException(

1760 status_code=500,

1761 detail={"error": "Database not connected. Please connect a database."},

1762 )

1763

1764 db_record: Final = await _ui_settings_db(UISettingsRepository(prisma_client)).find_unique(

1765 where={"id": "ui_settings"}

1766 )

1767

1768 stored: Final = (db_record.ui_settings if db_record else None) or "{}"

1769 parsed: Final = json.loads(stored) if isinstance(stored, str) else stored

1770

1771 # Sanitize any unexpected keys from persisted config before returning

1772 ui_settings: Final = {k: v for k, v in parsed.items() if k in ALLOWED_UI_SETTINGS_FIELDS}

1773

1774 apply_runtime_general_settings_flags(ui_settings)

1775

1776 # Refresh DualCache so other code paths (e.g. /user/filter/ui) see fresh values

1777 from litellm.proxy.proxy_server import user_api_key_cache

1778

1779 await user_api_key_cache.async_set_cache(key=UI_SETTINGS_CACHE_KEY, value=ui_settings, ttl=UI_SETTINGS_CACHE_TTL)

1780

1781 effective_ui_settings: Final[Mapping[str, object]] = MappingProxyType(

1782 {

1783 **ui_settings,

1784 **{key: proxy_config.settings[key] for key in ALLOWED_UI_SETTINGS_FIELDS if key in proxy_config.settings},

1785 }

1786 )

1787 config: Final[Mapping[str, object]] = MappingProxyType(

1788 {"litellm_settings": MappingProxyType({"ui_settings": effective_ui_settings})}

1789 )

1790 settings_class: Final = _get_effective_ui_settings_class()

1791 resolved_settings: Final = _SettingsWithSchema.model_validate(

1792 await _get_settings_with_schema(

1793 settings_key="ui_settings",

1794 settings_class=settings_class,

1795 config=config,

1796 )

1797 )

1798 values: Final[Mapping[str, object]] = MappingProxyType(

1799 {

1800 **resolved_settings.values,

1801 ENABLE_PTU_COST_ATTRIBUTION_UI_SETTING: is_ptu_cost_attribution_enabled(),

1802 APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING: _derived_ui_setting_value(

1803 APPLY_USER_BUDGET_TO_TEAM_KEYS_UI_SETTING

1804 ),

1805 }

1806 )

1807 source: Final[Mapping[str, FieldSource]] = MappingProxyType(

1808 {

1809 key: (

1810 _ui_setting_source(key, values[key], proxy_config.settings, settings_class)

1811 if key in proxy_config.settings or key not in ui_settings

1812 else "db"

1813 )

1814 for key in values

1815 }

1816 )

1817 return UISettingsResponse(

1818 values=values,

1819 field_schema=resolved_settings.field_schema,

1820 source=source,

1821 )

1822

1823

1824@router.patch(

1825 "/update/ui_settings",

1826 tags=["UI Settings"],

1827 dependencies=[Depends(user_api_key_auth)],

1828)

1829async def update_ui_settings(

1830 settings_body: dict[str, object] = Body(...),

1831 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1832):

1833 """

1834 Update UI-specific configuration flags.

1835 Only proxy admins are allowed to modify these settings.

1836 """

1837 from litellm.proxy.proxy_server import (

1838 create_config_audit_log,

1839 prisma_client,

1840 store_model_in_db,

1841 )

1842

1843 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1843 ↛ 1844line 1843 didn't jump to line 1844 because the condition on line 1843 was never true

1844 raise HTTPException(status_code=403, detail="Only proxy admins can update UI settings.")

1845

1846 if prisma_client is None: 1846 ↛ 1847line 1846 didn't jump to line 1847 because the condition on line 1846 was never true

1847 raise HTTPException(

1848 status_code=500,

1849 detail={"error": "Database not connected. Please connect a database."},

1850 )

1851

1852 if store_model_in_db is not True: 1852 ↛ 1853line 1852 didn't jump to line 1853 because the condition on line 1852 was never true

1853 raise HTTPException(

1854 status_code=500,

1855 detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."},

1856 )

1857

1858 conflicting_keys: Final = sorted(

1859 key

1860 for key, value in settings_body.items()

1861 if key in _DERIVED_UI_SETTINGS_FIELDS and value != _derived_ui_setting_value(key)

1862 )

1863 if conflicting_keys: 1863 ↛ 1864line 1863 didn't jump to line 1864 because the condition on line 1863 was never true

1864 raise HTTPException(

1865 status_code=400,

1866 detail=(

1867 f"Setting(s) {conflicting_keys} are derived from the deployment environment "

1868 "and cannot be changed from the UI."

1869 ),

1870 )

1871

1872 # Validate against the same effective class GET advertises, so

1873 # enterprise-registered fields are typed consistently on both sides.

1874 effective_cls: Final = _get_effective_ui_settings_class()

1875 try:

1876 settings: Final = effective_cls.model_validate(settings_body)

1877 except ValidationError as e:

1878 raise HTTPException(status_code=422, detail=e.errors())

1879

1880 unsupported_team_fields: Final = sorted(

1881 frozenset(settings.team_admin_editable_team_fields) - SUPPORTED_TEAM_ADMIN_PERMISSIONS

1882 )

1883 if unsupported_team_fields: 1883 ↛ 1884line 1883 didn't jump to line 1884 because the condition on line 1883 was never true

1884 raise HTTPException(

1885 status_code=400,

1886 detail={ # mutable-ok: HTTPException detail must be a plain dict for FastAPI JSON serialization

1887 "error": (

1888 f"{TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING} does not support {unsupported_team_fields}. "

1889 f"Supported fields: {sorted(SUPPORTED_TEAM_ADMIN_PERMISSIONS)}."

1890 )

1891 },

1892 )

1893

1894 # Only include fields the caller actually sent (not Pydantic defaults).

1895 settings_dict: Final[Mapping[str, JsonValue]] = settings.model_dump(exclude_unset=True)

1896

1897 # Reject enterprise-only settings up front so the caller gets a clear

1898 # signal instead of a silent drop.

1899 blocked_enterprise_keys = sorted((settings_dict.keys() & _ENTERPRISE_ONLY_UI_SETTINGS) - ALLOWED_UI_SETTINGS_FIELDS)

1900 if blocked_enterprise_keys: 1900 ↛ 1901line 1900 didn't jump to line 1901 because the condition on line 1900 was never true

1901 raise HTTPException(

1902 status_code=403,

1903 detail={

1904 "error": (

1905 f"Setting(s) {blocked_enterprise_keys} are a LiteLLM "

1906 "Enterprise feature and are not available on this build."

1907 )

1908 },

1909 )

1910

1911 # Enforce allowlist and drop anything unexpected

1912 incoming: Final = {k: v for k, v in settings_dict.items() if k in ALLOWED_UI_SETTINGS_FIELDS}

1913

1914 # Merge with existing persisted settings so a partial PATCH doesn't

1915 # overwrite fields the caller didn't send.

1916 existing: dict[str, JsonValue] = {}

1917 db_existing: Final = await _ui_settings_db(UISettingsRepository(prisma_client)).find_unique(

1918 where={"id": "ui_settings"}

1919 )

1920 if db_existing and db_existing.ui_settings: 1920 ↛ 1921line 1920 didn't jump to line 1921 because the condition on line 1920 was never true

1921 raw: Final = db_existing.ui_settings

1922 existing = json.loads(raw) if isinstance(raw, str) else dict(raw)

1923

1924 ui_settings: Final = {**existing, **incoming}

1925

1926 await _ui_settings_db(UISettingsRepository(prisma_client)).upsert(

1927 where={"id": "ui_settings"},

1928 data={

1929 "create": {

1930 "id": "ui_settings",

1931 "ui_settings": json.dumps(ui_settings),

1932 },

1933 "update": {

1934 "ui_settings": json.dumps(ui_settings),

1935 },

1936 },

1937 )

1938

1939 apply_runtime_general_settings_flags(ui_settings)

1940

1941 # Invalidate + set DualCache so subsequent reads see the new values immediately

1942 from litellm.proxy.proxy_server import user_api_key_cache

1943

1944 sanitized: Final = {k: v for k, v in ui_settings.items() if k in ALLOWED_UI_SETTINGS_FIELDS}

1945 await user_api_key_cache.async_set_cache(key=UI_SETTINGS_CACHE_KEY, value=sanitized, ttl=UI_SETTINGS_CACHE_TTL)

1946

1947 asyncio.create_task(

1948 create_config_audit_log(

1949 param_name="ui_settings",

1950 action="updated",

1951 before_value=existing,

1952 after_value=ui_settings,

1953 user_api_key_dict=user_api_key_dict,

1954 table_name=LitellmTableNames.UI_SETTINGS_TABLE_NAME,

1955 )

1956 )

1957

1958 return {

1959 "message": "UI settings updated successfully",

1960 "status": "success",

1961 "settings": ui_settings,

1962 }

1963

1964

1965@router.post(

1966 "/upload/logo",

1967 tags=["UI Theme Settings"],

1968 dependencies=[Depends(user_api_key_auth)],

1969)

1970async def upload_logo(

1971 file: UploadFile = File(...),

1972 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),

1973):

1974 """

1975 Upload a custom logo for the admin UI.

1976 Accepts image files (PNG, JPG, JPEG, SVG) and stores them for use in the UI.

1977 """

1978 import os

1979 from pathlib import Path

1980

1981 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1981 ↛ 1982line 1981 didn't jump to line 1982 because the condition on line 1981 was never true

1982 raise HTTPException(

1983 status_code=403,

1984 detail="Only proxy admins can upload a UI logo.",

1985 )

1986

1987 # Validate file type

1988 allowed_extensions: Final = {".png", ".jpg", ".jpeg", ".svg"}

1989 file_extension: Final = Path(file.filename or "").suffix.lower()

1990

1991 if file_extension not in allowed_extensions: 1991 ↛ 1999line 1991 didn't jump to line 1999 because the condition on line 1991 was always true

1992 raise HTTPException(

1993 status_code=400,

1994 detail=f"Invalid file type. Allowed types: {', '.join(allowed_extensions)}",

1995 )

1996

1997 # Read bounded to one byte past the limit, so an oversized upload is never

1998 # fully buffered in memory before being rejected.

1999 max_logo_size_bytes: Final = 5 * 1024 * 1024

2000 file_content: Final = await file.read(max_logo_size_bytes + 1)

2001 if len(file_content) > max_logo_size_bytes:

2002 raise HTTPException(status_code=400, detail="File size too large. Maximum size is 5MB.")

2003

2004 # Create uploads directory if it doesn't exist

2005 current_dir: Final = os.path.dirname(os.path.abspath(__file__))

2006 upload_dir: Final = os.path.join(current_dir, "..", "uploads")

2007 os.makedirs(upload_dir, exist_ok=True)

2008

2009 # Generate unique filename

2010 from litellm._uuid import uuid

2011

2012 unique_filename: Final = f"logo_{uuid.uuid4().hex}{file_extension}"

2013 file_path: Final = os.path.join(upload_dir, unique_filename)

2014

2015 # Save the file

2016 with open(file_path, "wb") as buffer:

2017 buffer.write(file_content)

2018

2019 return {

2020 "message": "Logo uploaded successfully",

2021 "status": "success",

2022 "file_path": file_path,

2023 "filename": unique_filename,

2024 "file_size": len(file_content),

2025 }

Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/ui_crud_endpoints/proxy_setting_endpoints.py: 74%

618 statements