scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

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

Browse source 
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.
This commit is contained in:
yucheng-berriai 2026-06-17 20:31:09 -07:00
parent 53593f697d
commit ad68c6b6a3
4 changed files with 387 additions and 68 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(

33
litellm/proxy/proxy_server.py
View file

@ -15011,6 +15011,27 @@ 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)
)
@router.get(
"/config/field/info",
tags=["config.yaml"],
@ -15063,9 +15084,15 @@ async def get_config_general_settings(
general_settings = dict(db_general_settings.param_value)
if field_name in general_settings:
return ConfigFieldInfo(
field_name=field_name, field_value=general_settings[field_name]
)
# only a full PROXY_ADMIN sees raw secret-bearing fields; others
# get them redacted
field_value = general_settings[field_name]
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
and _is_secret_general_setting_field(field_name)
):
field_value = "REDACTED"
return ConfigFieldInfo(field_name=field_name, field_value=field_value)
else:
raise HTTPException(
status_code=400,

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(

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

@ -8326,3 +8326,89 @@ def test_get_config_list_includes_cancel_on_disconnect(monkeypatch):
assert fields["cancel_on_disconnect"]["field_type"] == "Boolean"
finally:
app.dependency_overrides.clear()
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()