diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index e86982307e7..fb284e707a7 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -1016,15 +1016,10 @@ async def fetch_all_mcp_servers( 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 @@ async def fetch_mcp_server( 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( diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 9238cf91301..8f7b2f0b679 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -38,7 +38,7 @@ 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 ( @@ -15181,6 +15181,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"], @@ -15233,9 +15295,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 = [ ( @@ -15291,6 +15355,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"} @@ -15339,7 +15405,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() @@ -15369,7 +15439,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, @@ -15393,7 +15467,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, diff --git a/tests/code_coverage_tests/recursive_detector.py b/tests/code_coverage_tests/recursive_detector.py index 254d700ee5a..1d11d676207 100644 --- a/tests/code_coverage_tests/recursive_detector.py +++ b/tests/code_coverage_tests/recursive_detector.py @@ -47,6 +47,7 @@ "_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. diff --git a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py index 0b5b5fb6ceb..ce6c7e9b6fa 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py @@ -1197,6 +1197,136 @@ async def test_fetch_single_mcp_server_drops_env_vars_for_non_admin(self): # 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( diff --git a/tests/test_litellm/proxy/proxy_server/test_routes_config.py b/tests/test_litellm/proxy/proxy_server/test_routes_config.py index e89ada5bdef..df14dc5b5dc 100644 --- a/tests/test_litellm/proxy/proxy_server/test_routes_config.py +++ b/tests/test_litellm/proxy/proxy_server/test_routes_config.py @@ -15,8 +15,6 @@ 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 # --------------------------------------------------------------------------- diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py index c85c5ccb39f..a203fcc7ec0 100644 --- a/tests/test_litellm/proxy/test_proxy_server.py +++ b/tests/test_litellm/proxy/test_proxy_server.py @@ -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()