scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix: tighten role-based visibility of config and MCP fields (#30587)

Browse source 
* fix: redact config and MCP secrets in read-only admin views

GET /config/field/info and the MCP server list/detail endpoints returned
secret-bearing fields to any caller with an admin view, including
read-only admins. They now return those fields in full only to a full
PROXY_ADMIN; every other caller gets the reduced, non-admin view, while
non-sensitive fields remain readable. Regression tests cover the
role-based visibility on both endpoints, including that a full admin
still sees everything needed to populate the edit form.

* fix: redact nested secrets in config field info for non-admins

/config/field/info returned structured general_settings fields verbatim to
any admin-view caller, so a view-only admin reading database_args received
the nested aws_web_identity_token (a DynamoDB role-assumption credential) in
plaintext. Recurse into dict/list field values and redact secret leaves for
non-PROXY_ADMIN callers, leaving non-secret siblings and full-admin reads
unchanged

* fix: redact secret config values in /config/list for non-admins

/config/list shared the same _user_has_admin_view gate as /config/field/info
but returned each field value unredacted, so a view-only admin reading the
list received pass_through_endpoints upstream Authorization headers verbatim.
Route every general_settings value through a shared role-aware redactor
(extracted from /config/field/info) covering the top-level and nested field
paths, so non-PROXY_ADMIN callers get secret-bearing fields redacted while
full-admin reads stay unchanged

* chore(ci): allowlist _redact_secret_values_in_obj in recursive_detector

The config secret redactor recurses over JsonValue, which is acyclic, and
its depth is bounded by the operator-authored general_settings schema. Add
it to the recursive_detector ignore list alongside the other bounded
nested-redaction helpers (mask_dict, _redact_sensitive_litellm_params)

* proxy: cap recursive secret redaction depth at 10

Match the cap on _redact_sensitive_litellm_params (the closest analog
in the proxy, also recursive, key-name driven, returns a sentinel).

The previous justification — bounded by operator-authored schema depth,
JsonValue acyclic — is true today but is a property of the threat model,
not an enforced invariant of the function. If a code path is ever added
that pipes external input into general_settings (config import,
migration tooling, JWT-driven settings, …) the assumption silently
breaks. A local cap makes the invariant local.

The cap branch fails closed: at _REDACT_SECRET_MAX_DEPTH the whole
subtree is replaced with 'REDACTED' rather than returned verbatim. A
future refactor that flips this to fail-open would let a deeply nested
credential leak; the new regression test test_redact_secret_values_in_obj_fails_closed_at_max_depth
guards against that.

Updates the recursive_detector ignore-list rationale to point at the
numeric cap rather than the structural argument.

* test: actually exercise the depth cap in fails-closed test

The previous fixture stored the leaf under the secret-named key
'aws_web_identity_token', which the recursor's key-name short-circuit
redacts regardless of the cap — so the test passed both with and
without the cap in place. Empirically confirmed: under an uncapped
mutant the old fixture still hides the secret (key-name catches it),
the new fixture leaks it (only the cap can stop it). Swap the leaf
key to a non-secret name so the cap is the only redaction path
exercised, making the test fail on mutation as advertised.
This commit is contained in:
yucheng-berri 2026-06-23 15:25:30 -07:00 • committed by GitHub
parent f12c9bec48
commit c2e06890ad
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 628 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -1016,15 +1016,10 @@ if MCP_AVAILABLE:
if is_restricted_virtual_key:
return _sanitize_mcp_server_list_for_virtual_key(redacted_mcp_servers)
# Non-admin authenticated users may see the server inventory but
# not credential-bearing fields like `url` (often contains bearer
# tokens) or headers/env (often contain Authorization).
if not _user_has_admin_view(user_api_key_dict):
return _sanitize_mcp_server_list_for_non_admin(redacted_mcp_servers)
# only a full PROXY_ADMIN sees credential-bearing fields; everyone else
# goes through the non-admin sanitizer
if not _user_is_full_admin(user_api_key_dict):
for server in redacted_mcp_servers:
_redact_global_env_var_values(server)
return _sanitize_mcp_server_list_for_non_admin(redacted_mcp_servers)
return redacted_mcp_servers
@ -1415,10 +1410,10 @@ if MCP_AVAILABLE:
redacted = _redact_mcp_credentials(mcp_server)
if is_restricted_virtual_key:
return _sanitize_mcp_server_for_virtual_key(redacted)
if not _user_has_admin_view(user_api_key_dict):
return _sanitize_mcp_server_for_non_admin(redacted)
# only a full PROXY_ADMIN sees credential-bearing fields; everyone else
# goes through the non-admin sanitizer
if not _user_is_full_admin(user_api_key_dict):
_redact_global_env_var_values(redacted)
return _sanitize_mcp_server_for_non_admin(redacted)
return redacted
@router.post(

90
litellm/proxy/proxy_server.py
View file

@ -38,7 +38,7 @@ from typing import (
import anyio
import websockets
import websockets.exceptions
from pydantic import BaseModel, Json
from pydantic import BaseModel, Json, JsonValue
from litellm._uuid import uuid
from litellm.constants import (
@ -15183,6 +15183,68 @@ async def update_config_general_settings(
return response
# Secret-bearing general_settings fields the segment masker does not match by
# name: database_url and database_extra_connection_params embed DB credentials,
# pass_through_endpoints carry upstream Authorization headers, and
# alert_to_webhook_url is itself a webhook secret
_EXTRA_SECRET_GENERAL_SETTINGS_FIELDS = frozenset(
{
"database_url",
"database_extra_connection_params",
"pass_through_endpoints",
"alert_to_webhook_url",
}
)
def _is_secret_general_setting_field(field_name: str) -> bool:
return (
field_name in _EXTRA_SECRET_GENERAL_SETTINGS_FIELDS
or SENSITIVE_DATA_MASKER.is_sensitive_key(field_name)
)
# Matches the cap on _redact_sensitive_litellm_params (the closest analog in the
# proxy). Past this depth we fail closed by returning "REDACTED" for the whole
# subtree rather than recursing further — better to over-redact a pathological
# config than to silently return a deeply-nested credential verbatim
_REDACT_SECRET_MAX_DEPTH = 10
def _redact_secret_values_in_obj(value: JsonValue, depth: int = 0) -> JsonValue:
"""Recursively redact secret leaves inside a structured field so a nested
credential (e.g. aws_web_identity_token under database_args) is never
returned to a non-admin, while non-secret siblings stay visible. At
_REDACT_SECRET_MAX_DEPTH the whole subtree is replaced with "REDACTED"
so depth-overrun fails closed."""
if depth >= _REDACT_SECRET_MAX_DEPTH:
return "REDACTED"
if isinstance(value, dict):
return {
key: (
"REDACTED"
if _is_secret_general_setting_field(key)
else _redact_secret_values_in_obj(sub, depth + 1)
)
for key, sub in value.items()
}
if isinstance(value, list):
return [_redact_secret_values_in_obj(item, depth + 1) for item in value]
return value
def _redact_general_setting_value(
field_name: str, value: JsonValue, is_full_admin: bool
) -> JsonValue:
if is_full_admin:
return value
if _is_secret_general_setting_field(field_name):
return "REDACTED"
if isinstance(value, (dict, list)):
return _redact_secret_values_in_obj(value)
return value
@router.get(
"/config/field/info",
tags=["config.yaml"],
@ -15235,9 +15297,11 @@ async def get_config_general_settings(
general_settings = dict(db_general_settings.param_value)
if field_name in general_settings:
field_value = general_settings[field_name]
# Redact plugin_key from plugin configs so the shared credential
# is never returned even to admin-viewer callers.
field_value = _redact_general_setting_value(
field_name,
general_settings[field_name],
user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN,
)
if field_name == "plugins" and isinstance(field_value, list):
field_value = [
(
@ -15293,6 +15357,8 @@ async def get_config_list(
},
)
is_full_admin = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
## get general settings from db
db_general_settings = await ConfigRepository(prisma_client).table.find_first(
where={"param_name": "general_settings"}
@ -15341,7 +15407,11 @@ async def get_config_list(
field_name=sub_field,
field_type=sub_field_type.__name__,
field_description="", # Add custom logic if descriptions are available
field_default_value=general_settings.get(sub_field, None),
field_default_value=_redact_general_setting_value(
sub_field,
general_settings.get(sub_field, None),
is_full_admin,
),
stored_in_db=None,
)
for sub_field, sub_field_type in pydantic_class.__annotations__.items()
@ -15371,7 +15441,11 @@ async def get_config_list(
field_name=field_name,
field_type=allowed_args[field_name]["type"],
field_description=field_info.description or "",
field_value=general_settings.get(field_name, None),
field_value=_redact_general_setting_value(
field_name,
general_settings.get(field_name, None),
is_full_admin,
),
stored_in_db=_stored_in_db,
field_default_value=field_info.default,
nested_fields=nested_fields,
@ -15395,7 +15469,9 @@ async def get_config_list(
field_name=field_name,
field_type=allowed_args[field_name]["type"],
field_description=field_info.description or "",
field_value=_field_value,
field_value=_redact_general_setting_value(
field_name, _field_value, is_full_admin
),
stored_in_db=_stored_in_db,
field_default_value=field_info.default,
nested_fields=nested_fields,

1
tests/code_coverage_tests/recursive_detector.py
View file

@ -47,6 +47,7 @@ IGNORE_FUNCTIONS = [
"_read_image_bytes", # max depth set.
"_get_masked_values", # max depth set (default 20) to prevent infinite recursion while masking nested sensitive config dicts.
"_redact_sensitive_litellm_params", # max depth set (default 10).
"_redact_secret_values_in_obj", # max depth set (default 10, _REDACT_SECRET_MAX_DEPTH); fails closed by returning "REDACTED" at the cap.
"_resolve", # OCI: $ref resolver bounded by `resolving_stack` cycle guard.
"resolve_oci_schema_anyof", # OCI: bounded by JSON-schema tree depth (no cycles possible in well-formed input).
"sanitize_oci_schema", # OCI: bounded by JSON-schema tree depth.

319
tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
View file

@ -1197,6 +1197,136 @@ class TestListMCPServers:
# Non-admin viewers get no env var config at all (not even names).
assert result.env_vars is None
@pytest.mark.asyncio
async def test_fetch_single_mcp_server_sanitizes_for_view_only_admin(self):
"""PROXY_ADMIN_VIEW_ONLY must NOT see credential-bearing fields.
It previously passed the _user_has_admin_view gate (which also grants view-only
admins) and only had the explicit `credentials` field cleared, leaking secrets
embedded in url/static_headers/env_vars. Only a FULL PROXY_ADMIN may see those.
This test exercises the real role helpers (no patching of the gate)."""
mock_server = LiteLLM_MCPServerTable.model_construct(
server_id="leaky-server",
server_name="Leaky Server",
alias="Leaky Server",
transport=MCPTransport.http,
url="https://leaky.example.com/mcp?api_key=sk-embedded-in-url",
static_headers={"Authorization": "Bearer sk-secret-header"},
env={"UPSTREAM_TOKEN": "sk-secret-env"},
env_vars=[
{"name": "GLOBAL_KEY", "value": "super-secret", "scope": "global"},
],
credentials={"auth_value": "sk-explicit-credential"},
)
mock_prisma_client = MagicMock()
mock_health_result = generate_mock_mcp_server_db_record(
server_id="leaky-server", alias="Leaky Server"
)
mock_health_result.status = "healthy"
mock_health_result.last_health_check = datetime.now()
mock_health_result.health_check_error = None
mock_manager = MagicMock()
mock_manager.add_server = AsyncMock()
mock_manager.health_check_server = AsyncMock(return_value=mock_health_result)
mock_user_auth = generate_mock_user_api_key_auth(
user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
)
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=mock_prisma_client,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(return_value=mock_server),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager",
mock_manager,
),
):
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
fetch_mcp_server,
)
result = await fetch_mcp_server(
request=_make_mock_request(),
server_id="leaky-server",
user_api_key_dict=mock_user_auth,
)
assert result.server_id == "leaky-server"
assert result.credentials is None
assert result.url is None
assert result.static_headers is None
assert result.env == {}
assert result.env_vars is None
@pytest.mark.asyncio
async def test_fetch_single_mcp_server_full_admin_still_sees_secrets(self):
"""the fix must not over-redact for FULL PROXY_ADMIN,
who needs url/static_headers/env to populate the edit form."""
mock_server = LiteLLM_MCPServerTable.model_construct(
server_id="admin-server",
server_name="Admin Server",
alias="Admin Server",
transport=MCPTransport.http,
url="https://admin.example.com/mcp",
static_headers={"Authorization": "Bearer sk-secret-header"},
credentials={"auth_value": "sk-explicit-credential"},
)
mock_prisma_client = MagicMock()
mock_health_result = generate_mock_mcp_server_db_record(
server_id="admin-server", alias="Admin Server"
)
mock_health_result.status = "healthy"
mock_health_result.last_health_check = datetime.now()
mock_health_result.health_check_error = None
mock_manager = MagicMock()
mock_manager.add_server = AsyncMock()
mock_manager.health_check_server = AsyncMock(return_value=mock_health_result)
mock_user_auth = generate_mock_user_api_key_auth(
user_role=LitellmUserRoles.PROXY_ADMIN
)
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=mock_prisma_client,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(return_value=mock_server),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager",
mock_manager,
),
):
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
fetch_mcp_server,
)
result = await fetch_mcp_server(
request=_make_mock_request(),
server_id="admin-server",
user_api_key_dict=mock_user_auth,
)
# credentials field is always redacted; the rest must survive for full admin.
assert result.credentials is None
assert result.url == "https://admin.example.com/mcp"
assert result.static_headers == {"Authorization": "Bearer sk-secret-header"}
class TestTeamScopedMCPServerAccess:
"""Tests for cross-team information disclosure and restricted key bypass fixes."""
@ -3693,18 +3823,12 @@ def _server_with_env_vars(server_id: str = "srv-env"):
@pytest.mark.asyncio
@pytest.mark.parametrize(
"user_role, expected_global_value",
[
(LitellmUserRoles.PROXY_ADMIN, "super-secret"),
(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, ""),
],
)
async def test_fetch_single_mcp_server_redacts_global_env_for_view_only_admin(
user_role, expected_global_value
):
"""Read-only admins must not receive admin-supplied global env var secrets;
full admins still see them so the edit form can pre-fill."""
async def test_fetch_single_mcp_server_env_vars_full_admin_vs_view_only():
"""full admins see admin-supplied global env var secrets so the edit
form can pre-fill; read-only admins now go through the non-admin sanitizer, which
drops env_vars entirely (the names alone, e.g. ADMIN_API_KEY, leak what secrets the
admin configured). Previously the view-only case merely blanked the global value
while keeping the names, which still leaked configuration metadata."""
server = _server_with_env_vars()
health_result = generate_mock_mcp_server_db_record(server_id=server.server_id)
@ -3712,34 +3836,39 @@ async def test_fetch_single_mcp_server_redacts_global_env_for_view_only_admin(
health_result.last_health_check = datetime.now()
health_result.health_check_error = None
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=MagicMock(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(return_value=server),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.add_server",
AsyncMock(return_value=None),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.health_check_server",
AsyncMock(return_value=health_result),
),
):
result = await mgmt_endpoints.fetch_mcp_server(
request=_make_mock_request(),
server_id=server.server_id,
user_api_key_dict=generate_mock_user_api_key_auth(user_role=user_role),
)
async def _fetch(user_role):
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=MagicMock(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(return_value=server),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.add_server",
AsyncMock(return_value=None),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.health_check_server",
AsyncMock(return_value=health_result),
),
):
return await mgmt_endpoints.fetch_mcp_server(
request=_make_mock_request(),
server_id=server.server_id,
user_api_key_dict=generate_mock_user_api_key_auth(user_role=user_role),
)
by_name = {ev.name: ev for ev in result.env_vars}
assert by_name["ADMIN_API_KEY"].value == expected_global_value
# Per-user placeholders are always preserved.
full_admin = await _fetch(LitellmUserRoles.PROXY_ADMIN)
by_name = {ev.name: ev for ev in full_admin.env_vars}
assert by_name["ADMIN_API_KEY"].value == "super-secret"
assert by_name["USER_TOKEN"].value == "placeholder-hint"
view_only = await _fetch(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
assert view_only.env_vars is None
# The source record must never be mutated.
assert {ev.name: ev.value for ev in server.env_vars}[
"ADMIN_API_KEY"
@ -3747,18 +3876,67 @@ async def test_fetch_single_mcp_server_redacts_global_env_for_view_only_admin(
@pytest.mark.asyncio
@pytest.mark.parametrize(
"user_role, expected_global_value",
[
(LitellmUserRoles.PROXY_ADMIN, "super-secret"),
(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, ""),
],
)
async def test_fetch_all_mcp_servers_redacts_global_env_for_view_only_admin(
user_role, expected_global_value
):
async def test_fetch_all_mcp_servers_env_vars_full_admin_vs_view_only():
"""same posture as the single-server fetch. Full admins
keep the env var values; view-only admins get env_vars dropped via the non-admin
sanitizer rather than only having the global value blanked."""
server = _server_with_env_vars()
async def _fetch_all(user_role):
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_user_mcp_management_mode",
return_value="view_all",
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.get_all_mcp_servers_unfiltered",
AsyncMock(return_value=[server]),
),
patch(
"litellm.proxy.proxy_server.prisma_client",
None,
),
):
return await mgmt_endpoints.fetch_all_mcp_servers(
user_api_key_dict=generate_mock_user_api_key_auth(user_role=user_role),
)
full_admin = await _fetch_all(LitellmUserRoles.PROXY_ADMIN)
by_name = {ev.name: ev for ev in full_admin[0].env_vars}
assert by_name["ADMIN_API_KEY"].value == "super-secret"
assert by_name["USER_TOKEN"].value == "placeholder-hint"
view_only = await _fetch_all(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
assert view_only[0].env_vars is None
assert {ev.name: ev.value for ev in server.env_vars}[
"ADMIN_API_KEY"
] == "super-secret"
def _leaky_list_server() -> "LiteLLM_MCPServerTable":
"""A server whose url/static_headers/env carry embedded secrets, for the
list-endpoint sanitization tests. ``model_construct`` skips validation so
the raw values survive verbatim."""
return LiteLLM_MCPServerTable.model_construct(
server_id="leaky-list-server",
server_name="Leaky List Server",
alias="Leaky List Server",
transport=MCPTransport.http,
url="https://leaky.example.com/mcp?api_key=sk-embedded-in-url",
static_headers={"Authorization": "Bearer sk-secret-header"},
env={"UPSTREAM_TOKEN": "sk-secret-env"},
env_vars=[
{"name": "GLOBAL_KEY", "value": "super-secret", "scope": "global"},
],
credentials={"auth_value": "sk-explicit-credential"},
)
async def _fetch_all_via_view_all(user_role: LitellmUserRoles):
"""Drive GET /v1/mcp/server in view_all mode for the given role using the
real role helpers (the full-admin gate is never patched)."""
server = _leaky_list_server()
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_user_mcp_management_mode",
@ -3776,13 +3954,46 @@ async def test_fetch_all_mcp_servers_redacts_global_env_for_view_only_admin(
result = await mgmt_endpoints.fetch_all_mcp_servers(
user_api_key_dict=generate_mock_user_api_key_auth(user_role=user_role),
)
return server, result
by_name = {ev.name: ev for ev in result[0].env_vars}
assert by_name["ADMIN_API_KEY"].value == expected_global_value
assert by_name["USER_TOKEN"].value == "placeholder-hint"
assert {ev.name: ev.value for ev in server.env_vars}[
"ADMIN_API_KEY"
] == "super-secret"
@pytest.mark.asyncio
async def test_list_mcp_servers_sanitized_for_view_only_admin():
"""PROXY_ADMIN_VIEW_ONLY listing servers must go through the non-admin
sanitizer: url and static_headers cleared, env emptied, env_vars dropped.
A mutation swapping _user_is_full_admin() back to _user_has_admin_view()
(which also grants view-only admins) would return the raw url/headers and
fail this. The real role helpers are exercised; the gate is not patched."""
source, result = await _fetch_all_via_view_all(
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
)
assert len(result) == 1
sanitized = result[0]
assert sanitized.server_id == "leaky-list-server"
assert sanitized.url is None
assert sanitized.static_headers is None
assert sanitized.env == {}
assert sanitized.env_vars is None
assert sanitized.credentials is None
# The source record must never be mutated by sanitization.
assert source.url == "https://leaky.example.com/mcp?api_key=sk-embedded-in-url"
assert source.static_headers == {"Authorization": "Bearer sk-secret-header"}
@pytest.mark.asyncio
async def test_list_mcp_servers_full_admin_still_sees_secrets():
"""The view-only redaction must not over-redact for a FULL PROXY_ADMIN,
who needs url/static_headers to populate the edit form. Only the explicit
credentials field is cleared for full admins on the list endpoint."""
_, result = await _fetch_all_via_view_all(LitellmUserRoles.PROXY_ADMIN)
assert len(result) == 1
raw = result[0]
assert raw.url == "https://leaky.example.com/mcp?api_key=sk-embedded-in-url"
assert raw.static_headers == {"Authorization": "Bearer sk-secret-header"}
assert raw.credentials is None
def _make_env_var_server(

189
tests/test_litellm/proxy/proxy_server/test_routes_config.py
View file

@ -15,8 +15,6 @@ from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
import pytest
from .conftest import VOLATILE_KEYS, normalize
@ -248,6 +246,193 @@ def test_config_field_info_field_not_in_db(client, auth_as, mock_prisma, monkeyp
assert "not in DB" in response.json().get("detail", {}).get("error", "")
def test_config_field_info_redacts_nested_secret_for_view_only_admin(
client, auth_as, mock_prisma, monkeypatch
):
"""A view-only admin reading a structured field must not receive nested
credentials. database_args carries aws_web_identity_token (a DynamoDB
role-assumption credential); it must come back redacted while non-secret
siblings like region_name stay visible."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
table = _install_litellm_config(mock_prisma)
row = MagicMock()
row.param_value = {
"database_args": {
"region_name": "us-east-1",
"user_table_name": "LiteLLM_UserTable",
"aws_web_identity_token": "sk-super-secret-token",
}
}
table.find_first = AsyncMock(return_value=row)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
response = client.get(
"/config/field/info", params={"field_name": "database_args"}
)
assert response.status_code == 200
value = response.json()["field_value"]
assert value["aws_web_identity_token"] == "REDACTED"
assert value["region_name"] == "us-east-1"
assert value["user_table_name"] == "LiteLLM_UserTable"
def test_config_field_info_full_admin_sees_nested_secret(
client, auth_as, mock_prisma, monkeypatch
):
"""The redaction must not over-redact for a full PROXY_ADMIN, who needs
the real nested value to populate the edit form."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
table = _install_litellm_config(mock_prisma)
row = MagicMock()
row.param_value = {
"database_args": {
"region_name": "us-east-1",
"aws_web_identity_token": "sk-super-secret-token",
}
}
table.find_first = AsyncMock(return_value=row)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
response = client.get(
"/config/field/info", params={"field_name": "database_args"}
)
assert response.status_code == 200
value = response.json()["field_value"]
assert value["aws_web_identity_token"] == "sk-super-secret-token"
assert value["region_name"] == "us-east-1"
def test_config_field_info_redacts_top_level_scalar_for_view_only(
client, auth_as, mock_prisma, monkeypatch
):
"""The top-level scalar branch must also redact for a view-only admin.
database_url carries DB credentials and is not caught by the name masker,
so it is in the explicit secret set."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
table = _install_litellm_config(mock_prisma)
row = MagicMock()
row.param_value = {"database_url": "postgresql://admin:p4ss@db:5432/litellm"}
table.find_first = AsyncMock(return_value=row)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
response = client.get(
"/config/field/info", params={"field_name": "database_url"}
)
assert response.status_code == 200
assert response.json()["field_value"] == "REDACTED"
def test_redact_general_setting_value_recurses_list_of_dicts():
"""The list branch of the recursor redacts secret leaves inside each dict
while non-secret keys survive, and a full admin gets the value untouched."""
from litellm.proxy import proxy_server as ps
value = [
{"path": "/foo", "headers": {"Authorization": "Bearer sk-x"}},
{"path": "/bar", "client_secret": "sk-y"},
]
redacted = ps._redact_general_setting_value(
"some_list_field", value, is_full_admin=False
)
assert redacted[0]["headers"]["Authorization"] == "REDACTED"
assert redacted[0]["path"] == "/foo"
assert redacted[1]["client_secret"] == "REDACTED"
assert redacted[1]["path"] == "/bar"
assert (
ps._redact_general_setting_value("some_list_field", value, is_full_admin=True)
== value
)
def test_redact_secret_values_in_obj_fails_closed_at_max_depth():
"""Past _REDACT_SECRET_MAX_DEPTH the whole subtree is replaced with
"REDACTED" rather than returned verbatim, so a secret buried below the cap
can never leak via depth-overrun. A future refactor that flips the cap
branch to fail-open would surface here."""
from litellm.proxy import proxy_server as ps
# leaf and wrap keys are both NON-secret so neither the key-name
# short-circuit nor the explicit-secret set catches the leak. The cap is
# the only thing standing between the secret and the response — flip the
# cap to fail-open and the secret comes back verbatim.
nested: object = {"notes": "sk-leak-bottom"}
for _ in range(ps._REDACT_SECRET_MAX_DEPTH + 2):
nested = {"wrap": nested}
out = ps._redact_general_setting_value(
"some_struct_field", nested, is_full_admin=False
)
# the secret must not survive anywhere in the returned tree
assert "sk-leak-bottom" not in repr(out)
# full admin is unaffected by the cap — the value comes back untouched
admin_out = ps._redact_general_setting_value(
"some_struct_field", nested, is_full_admin=True
)
assert admin_out is nested
def test_config_list_redacts_pass_through_secret_for_view_only(
client, auth_as, mock_prisma, monkeypatch
):
"""/config/list must not leak pass_through_endpoints upstream credentials
to a view-only admin. pass_through_endpoints is a known secret-bearing
field, so a non-admin gets it redacted; a full admin still sees it."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
table = _install_litellm_config(mock_prisma)
row = MagicMock()
row.param_value = {"max_parallel_requests": 3}
table.find_first = AsyncMock(return_value=row)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(
ps,
"general_settings",
{
"pass_through_endpoints": [
{
"path": "/foo",
"target": "https://upstream.example.com",
"headers": {"Authorization": "Bearer sk-UPSTREAM-SECRET"},
}
]
},
)
def _pass_through_value(body):
return next(
entry["field_value"]
for entry in body
if entry["field_name"] == "pass_through_endpoints"
)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
view_resp = client.get(
"/config/list", params={"config_type": "general_settings"}
)
assert view_resp.status_code == 200
assert "sk-UPSTREAM-SECRET" not in view_resp.text
assert _pass_through_value(view_resp.json()) == "REDACTED"
with auth_as(LitellmUserRoles.PROXY_ADMIN):
admin_resp = client.get(
"/config/list", params={"config_type": "general_settings"}
)
assert admin_resp.status_code == 200
admin_value = _pass_through_value(admin_resp.json())
assert admin_value[0]["headers"]["Authorization"] == "Bearer sk-UPSTREAM-SECRET"
# ---------------------------------------------------------------------------
# GET /config/list
# ---------------------------------------------------------------------------

86
tests/test_litellm/proxy/test_proxy_server.py
View file

@ -8364,3 +8364,89 @@ def test_preserve_redacted_plugin_keys_sets_new_and_drops_orphan_placeholder():
[{"name": "p2", "url": "https://p2", "plugin_key": "***"}], existing
)
assert "plugin_key" not in new_plugin[0]
def _config_field_info_client(monkeypatch, user_role):
import types
from unittest.mock import AsyncMock, MagicMock
from fastapi.testclient import TestClient
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.proxy_server import app
db_record = types.SimpleNamespace(
param_value={
"master_key": "sk-super-secret-master",
"database_url": "postgresql://user:p4ssw0rd@db:5432/litellm",
"pass_through_endpoints": [
{
"path": "/upstream",
"target": "https://upstream.example.com",
"headers": {"Authorization": "Bearer sk-upstream-secret"},
}
],
"max_parallel_requests": 100,
}
)
mock_config_table = MagicMock()
mock_config_table.find_first = AsyncMock(return_value=db_record)
mock_prisma = MagicMock()
mock_prisma.db = types.SimpleNamespace(litellm_config=mock_config_table)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="u", user_role=user_role
)
return TestClient(app)
def test_config_field_info_redacts_secrets_for_view_only_admin(monkeypatch):
"""/config/field/info gates on _user_has_admin_view, which also grants
PROXY_ADMIN_VIEW_ONLY. A view-only admin reading master_key/database_url verbatim is
effectively a full admin. Secret-bearing fields must come back REDACTED for anyone who
is not a FULL PROXY_ADMIN, while non-secret fields stay readable."""
from litellm.proxy._types import LitellmUserRoles
client = _config_field_info_client(
monkeypatch, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
)
try:
for secret_field in ("master_key", "database_url", "pass_through_endpoints"):
resp = client.get("/config/field/info", params={"field_name": secret_field})
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["field_value"] == "REDACTED"
assert "secret" not in str(body["field_value"])
assert "p4ssw0rd" not in str(body["field_value"])
resp = client.get(
"/config/field/info", params={"field_name": "max_parallel_requests"}
)
assert resp.status_code == 200, resp.text
assert resp.json()["field_value"] == 100
finally:
app.dependency_overrides.clear()
def test_config_field_info_returns_raw_secrets_for_full_admin(monkeypatch):
"""the redaction must not over-apply. A FULL PROXY_ADMIN still
needs the real master_key value to populate the admin edit form."""
from litellm.proxy._types import LitellmUserRoles
client = _config_field_info_client(monkeypatch, LitellmUserRoles.PROXY_ADMIN)
try:
resp = client.get("/config/field/info", params={"field_name": "master_key"})
assert resp.status_code == 200, resp.text
assert resp.json()["field_value"] == "sk-super-secret-master"
resp = client.get(
"/config/field/info", params={"field_name": "pass_through_endpoints"}
)
assert resp.status_code == 200, resp.text
assert (
resp.json()["field_value"][0]["headers"]["Authorization"]
== "Bearer sk-upstream-secret"
)
finally:
app.dependency_overrides.clear()