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(proxy): redact MCP server URL and headers for non-admin viewers (VERIA-8)

Browse source 
Many MCP integrations (Zapier, etc.) embed an upstream API key
directly in the server URL, e.g.
``https://actions.zapier.com/mcp/<api-key>/sse``. The list and
single-server endpoints were returning the full URL to any
authenticated user — `_redact_mcp_credentials` only stripped the
explicit ``credentials`` field, and `_sanitize_mcp_server_for_virtual_key`
only ran for restricted virtual keys. Non-admin internal users could
read the dashboard, click the unmask toggle, and exfiltrate the raw
token.

Add `_sanitize_mcp_server_for_non_admin` that runs on top of the
existing credential redaction and clears the credential-bearing
fields:

- ``url`` (the primary leak vector)
- ``spec_path`` (OpenAPI spec URLs that may carry tokens)
- ``static_headers`` / ``extra_headers`` (Authorization)
- ``env`` (arbitrary secrets)
- ``authorization_url`` / ``token_url`` / ``registration_url``

Identity fields (``server_id``, ``alias``, ``mcp_info``, etc.) are
preserved so the UI can still list servers a non-admin's team has
access to.

Apply the new sanitizer in `fetch_all_mcp_servers` and the per-server
fetch path right after the existing virtual-key branch. Update the
existing `test_list_mcp_servers_non_admin_user_filtered` assertions
that previously checked URL visibility.

Frontend defense-in-depth: hide the URL unmask toggle on
`mcp_server_view.tsx` unless the viewer is a proxy admin.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
user 2026-05-01 23:59:38 +00:00
parent 934ecdca78
commit 4ea32d13c9
No known key found for this signature in database
3 changed files with 212 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -477,6 +477,40 @@ if MCP_AVAILABLE:
allowed_routes = getattr(user_api_key_dict, "allowed_routes", None)
return isinstance(allowed_routes, list) and len(allowed_routes) > 0
def _sanitize_mcp_server_for_non_admin(
mcp_server: LiteLLM_MCPServerTable,
) -> LiteLLM_MCPServerTable:
"""Strip credential-bearing fields for non-admin viewers.
Non-admin users may legitimately need to discover MCP servers
their team has access to (so they can pick one in the UI), but
they must never see fields that can carry bearer tokens or
upstream API keys. ``_redact_mcp_credentials`` already clears
the explicit ``credentials`` field; this layers on top to catch
the URL+headers+env vectors that the virtual-key sanitizer also
strips. Reset values match each field's declared default on
``LiteLLM_MCPServerTable`` (``None`` for Optional fields,
``[]``/``{}`` for required list/dict fields).
"""
sanitized = _redact_mcp_credentials(mcp_server)
# URL is the highest-impact vector: many MCP integrations embed
# the upstream API key directly in the path. spec_path can carry
# similar tokens in the OpenAPI spec URL.
sanitized.url = None
sanitized.spec_path = None
sanitized.static_headers = None
sanitized.extra_headers = []
sanitized.env = {}
sanitized.authorization_url = None
sanitized.token_url = None
sanitized.registration_url = None
return sanitized
def _sanitize_mcp_server_list_for_non_admin(
mcp_servers: Iterable[LiteLLM_MCPServerTable],
) -> List[LiteLLM_MCPServerTable]:
return [_sanitize_mcp_server_for_non_admin(s) for s in mcp_servers]
def _sanitize_mcp_server_for_virtual_key(
mcp_server: LiteLLM_MCPServerTable,
) -> LiteLLM_MCPServerTable:
@ -926,6 +960,12 @@ 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)
return redacted_mcp_servers
@router.get(
@ -1293,6 +1333,8 @@ 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)
return redacted
@router.post(

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

@ -626,14 +626,16 @@ class TestListMCPServers:
assert "config_server_allowed" in server_ids
assert "config_server_not_allowed" not in server_ids
# Check server details
# Check server details — non-admin viewers must not see the
# raw `url` (it can carry bearer tokens for many MCP
# integrations). Identity fields stay so the UI can list
# the server.
for server in result:
assert server.url is None
if server.server_id == "db_server_allowed":
assert server.alias == "Allowed Gmail MCP"
assert server.url == "https://gmail-mcp.example.com/mcp"
elif server.server_id == "config_server_allowed":
assert server.alias == "Allowed Zapier MCP"
assert server.url == "https://actions.zapier.com/mcp/sse"
@pytest.mark.asyncio
async def test_admin_user_with_object_permission_respects_mcp_servers(self):
@ -2897,3 +2899,162 @@ async def test_list_mcp_user_credentials_batch_server_fetch():
assert result[0].alias == "My Server"
# expires_at should always be the raw timestamp (not set to None when expired)
assert result[0].expires_at == "2099-01-01T00:00:00+00:00"
# ---------------------------------------------------------------------------
# VERIA-8: non-admin viewers must not see the raw MCP server URL
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_mcp_servers_non_admin_url_redacted():
"""A standard authenticated user (no admin role, not a restricted
virtual key) used to receive the raw `url` field, which can contain
bearer tokens like `https://actions.zapier.com/mcp/<api-key>/sse`.
They must now get the credential-bearing fields stripped."""
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
fetch_all_mcp_servers,
)
user = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="alice",
api_key="sk-alice",
# NOT allowed_routes — a normal authenticated user.
)
server = generate_mock_mcp_server_db_record(
server_id="zapier-1",
alias="Zapier",
url="https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse",
)
server.static_headers = {"Authorization": "Bearer SUPER-SECRET-TOKEN"}
server.env = {"API_KEY": "another-secret"}
server.extra_headers = ["Authorization"]
server.authorization_url = "https://oauth.example.com/authorize?token=foo"
server.token_url = "https://oauth.example.com/token"
server.registration_url = "https://oauth.example.com/register"
mock_manager = MagicMock()
mock_manager.get_all_mcp_servers_unfiltered = AsyncMock(return_value=[server])
mock_manager.get_all_allowed_mcp_servers = AsyncMock(return_value=[server])
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",
mock_manager,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=MagicMock(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts",
AsyncMock(return_value=[user]),
),
):
result = await fetch_all_mcp_servers(user_api_key_dict=user)
assert len(result) == 1
s = result[0]
# Identity fields stay so the UI can list the server.
assert s.server_id == "zapier-1"
assert s.alias == "Zapier"
# Credential-bearing fields must all be cleared.
assert s.url is None
assert s.static_headers is None
assert s.env == {}
assert s.extra_headers == []
assert s.authorization_url is None
assert s.token_url is None
assert s.registration_url is None
@pytest.mark.asyncio
async def test_list_mcp_servers_admin_keeps_url():
"""Proxy admins must continue to see the raw URL — the redaction
only applies to non-admin viewers."""
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
fetch_all_mcp_servers,
)
admin = UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN,
user_id="root",
api_key="sk-admin",
)
server = generate_mock_mcp_server_db_record(
server_id="zapier-1",
alias="Zapier",
url="https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse",
)
server.static_headers = {"Authorization": "Bearer SUPER-SECRET-TOKEN"}
mock_manager = MagicMock()
mock_manager.get_all_mcp_servers_unfiltered = AsyncMock(return_value=[server])
mock_manager.get_all_allowed_mcp_servers = AsyncMock(return_value=[server])
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",
mock_manager,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=MagicMock(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts",
AsyncMock(return_value=[admin]),
),
):
result = await fetch_all_mcp_servers(user_api_key_dict=admin)
assert len(result) == 1
assert result[0].url == "https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse"
assert result[0].static_headers == {"Authorization": "Bearer SUPER-SECRET-TOKEN"}
def test_sanitize_mcp_server_for_non_admin_clears_credential_fields():
"""Direct unit test on the helper for fast feedback."""
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
_sanitize_mcp_server_for_non_admin,
)
server = generate_mock_mcp_server_db_record(
url="https://example.com/mcp/SUPER-SECRET/sse",
)
server.credentials = {"auth_value": "secret"}
server.static_headers = {"Authorization": "Bearer x"}
server.spec_path = "https://example.com/specs/openapi-with-token.yaml"
server.env = {"API_KEY": "y"}
server.extra_headers = ["Authorization"]
server.authorization_url = "https://idp/authorize"
server.token_url = "https://idp/token"
server.registration_url = "https://idp/register"
sanitized = _sanitize_mcp_server_for_non_admin(server)
assert sanitized.credentials is None
assert sanitized.url is None
assert sanitized.spec_path is None
assert sanitized.static_headers is None
assert sanitized.env == {}
assert sanitized.extra_headers == []
assert sanitized.authorization_url is None
assert sanitized.token_url is None
assert sanitized.registration_url is None
# Identity / metadata fields are preserved so the UI can list the
# server without exposing secrets.
assert sanitized.server_id == server.server_id
assert sanitized.alias == server.alias

7
ui/litellm-dashboard/src/components/mcp_tools/mcp_server_view.tsx
View file

@ -147,7 +147,12 @@ export const MCPServerView: React.FC<MCPServerViewProps> = ({
<Text className="break-all overflow-wrap-anywhere font-mono text-sm">
{renderUrlWithToggle(mcpServer.url, showFullUrl)}
</Text>
{hasToken && (
{/* Only proxy admins may reveal the raw URL — non-admins
receive a sanitized server object from the backend
with `url=null`, but hide the toggle anyway as
defense-in-depth in case the URL ever leaks back
into the response. */}
{hasToken && isProxyAdmin && (
<button onClick={() => setShowFullUrl(!showFullUrl)} className="p-1 hover:bg-gray-100 rounded flex-shrink-0">
<Icon icon={showFullUrl ? EyeOffIcon : EyeIcon} size="sm" className="text-gray-500" />
</button>