fix(mcp): redact global env var secrets from read-only admin server views

PROXY_ADMIN_VIEW_ONLY callers are treated as admin_view by GET /v1/mcp/server
and GET /v1/mcp/server/{id}, so they received unredacted scope=global env var
values, which can carry upstream API keys used in Authorization headers. Only
a full PROXY_ADMIN needs those values (to pre-fill the edit form); read-only
admins now get the same global-secret redaction already applied to non-admin
and restricted virtual-key views.
This commit is contained in:
mateo-berri 2026-06-03 19:36:43 +00:00
parent 236407a57b
commit 102b73a261
No known key found for this signature in database
2 changed files with 121 additions and 0 deletions

View file

@ -496,6 +496,15 @@ if MCP_AVAILABLE:
if env_var.scope == MCPEnvVarScope.global_:
env_var.value = ""
def _user_is_full_admin(user_api_key_dict: UserAPIKeyAuth) -> bool:
"""True only for ``PROXY_ADMIN``; ``PROXY_ADMIN_VIEW_ONLY`` returns False.
Global env var secrets pre-fill the admin edit form, so a full admin
must see them, but a read-only admin gets the same redacted view as
any other non-managing caller.
"""
return user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
def _is_restricted_virtual_key_request(user_api_key_dict: UserAPIKeyAuth) -> bool:
"""Best-effort detection for route-restricted virtual keys.
@ -1000,6 +1009,10 @@ if MCP_AVAILABLE:
if not _user_has_admin_view(user_api_key_dict):
return _sanitize_mcp_server_list_for_non_admin(redacted_mcp_servers)
if not _user_is_full_admin(user_api_key_dict):
for server in redacted_mcp_servers:
_redact_global_env_var_values(server)
return redacted_mcp_servers
@router.get(
@ -1387,6 +1400,8 @@ if MCP_AVAILABLE:
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)
if not _user_is_full_admin(user_api_key_dict):
_redact_global_env_var_values(redacted)
return redacted
@router.post(

View file

@ -3451,6 +3451,112 @@ def test_sanitize_masks_global_env_var_secrets(sanitizer_name):
assert original_by_name["ADMIN_API_KEY"].value == "super-secret"
def _server_with_env_vars(server_id: str = "srv-env"):
base = generate_mock_mcp_server_db_record(server_id=server_id)
return LiteLLM_MCPServerTable(
**{
**base.model_dump(),
"env_vars": [
{"name": "ADMIN_API_KEY", "value": "super-secret", "scope": "global"},
{"name": "USER_TOKEN", "value": "placeholder-hint", "scope": "user"},
],
}
)
@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."""
server = _server_with_env_vars()
health_result = generate_mock_mcp_server_db_record(server_id=server.server_id)
health_result.status = "healthy"
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),
)
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.
assert by_name["USER_TOKEN"].value == "placeholder-hint"
# The source record must never be mutated.
assert {ev.name: ev.value for ev in server.env_vars}[
"ADMIN_API_KEY"
] == "super-secret"
@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
):
server = _server_with_env_vars()
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,
),
):
result = await mgmt_endpoints.fetch_all_mcp_servers(
user_api_key_dict=generate_mock_user_api_key_auth(user_role=user_role),
)
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"
def _make_env_var_server(
*,
server_id: str = "srv-1",