From 7c64636ce664b63f5db139e349b576b8f907bf42 Mon Sep 17 00:00:00 2001 From: BillionClaw <267901332+BillionClaw@users.noreply.github.com> Date: Tue, 17 Mar 2026 14:23:04 +0800 Subject: [PATCH] fix(proxy): mask sensitive fields in key/info endpoint metadata Prevents Langfuse API keys (and other sensitive credentials) from leaking in the key/info endpoint response by masking them before returning. Uses the existing SensitiveDataMasker to redact sensitive fields like: - langfuse_public_key - langfuse_secret_key - api_key, password, token, secret, etc. Fixes #23776 --- .../key_management_endpoints.py | 30 ++++ .../test_key_info_sensitive_data_masking.py | 158 ++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 tests/test_litellm/proxy/management_endpoints/test_key_info_sensitive_data_masking.py diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 1c0c212b60b..ad7c6e7506b 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -35,6 +35,7 @@ from litellm.constants import ( ) from litellm.litellm_core_utils.duration_parser import duration_in_seconds from litellm.litellm_core_utils.safe_json_dumps import safe_dumps +from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker from litellm.proxy._experimental.mcp_server.db import ( rotate_mcp_server_credentials_master_key, ) @@ -2525,6 +2526,8 @@ async def info_key_fn_v2( except Exception: # if using pydantic v1 k = k.dict() + # Mask sensitive fields in metadata to prevent credential leaks + k = _mask_sensitive_fields_in_key_info(k) filtered_key_info.append(k) return {"key": data.keys, "info": filtered_key_info} @@ -2611,6 +2614,9 @@ async def info_key_fn( # Attach object_permission if object_permission_id is set key_info = await attach_object_permission_to_dict(key_info, prisma_client) + # Mask sensitive fields in metadata to prevent credential leaks + key_info = _mask_sensitive_fields_in_key_info(key_info) + return {"key": key, "info": key_info} except Exception as e: raise handle_exception_on_proxy(e) @@ -5077,6 +5083,30 @@ async def _can_user_query_key_info( return False +def _mask_sensitive_fields_in_key_info(key_info: Dict[str, Any]) -> Dict[str, Any]: + """ + Masks sensitive fields (API keys, secrets, tokens) in key metadata before returning + to the client. This prevents credential leaks in the key/info endpoint response. + + Specifically handles: + - metadata.logging[*].callback_vars (contains langfuse_public_key, langfuse_secret_key, etc.) + - metadata.callback_settings.callback_vars + - Any other nested dicts with sensitive field names + """ + if not key_info or not isinstance(key_info, dict): + return key_info + + # Create a copy to avoid modifying the original + masked_key_info = copy.deepcopy(key_info) + metadata = masked_key_info.get("metadata") + + if metadata and isinstance(metadata, dict): + masker = SensitiveDataMasker() + masked_key_info["metadata"] = masker.mask_dict(metadata) + + return masked_key_info + + async def test_key_logging( user_api_key_dict: UserAPIKeyAuth, request: Request, diff --git a/tests/test_litellm/proxy/management_endpoints/test_key_info_sensitive_data_masking.py b/tests/test_litellm/proxy/management_endpoints/test_key_info_sensitive_data_masking.py new file mode 100644 index 00000000000..cd2b5d1cb49 --- /dev/null +++ b/tests/test_litellm/proxy/management_endpoints/test_key_info_sensitive_data_masking.py @@ -0,0 +1,158 @@ +from litellm.proxy.management_endpoints.key_management_endpoints import ( + _mask_sensitive_fields_in_key_info, +) + + +class TestMaskSensitiveFieldsInKeyInfo: + """Test the _mask_sensitive_fields_in_key_info function to ensure sensitive data is masked.""" + + def test_mask_langfuse_credentials_in_metadata_logging(self): + """Test that langfuse_public_key and langfuse_secret_key are masked in metadata.logging.""" + key_info = { + "token": "hashed_token", + "user_id": "user123", + "metadata": { + "logging": [ + { + "callback_name": "langfuse", + "callback_type": "success_and_failure", + "callback_vars": { + "langfuse_public_key": "pk-lf-1234567890abcdef", + "langfuse_secret_key": "sk-lf-0987654321fedcba", + "langfuse_host": "https://langfuse.example.com", + }, + } + ] + }, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + # Verify the sensitive keys are masked (format: first 4 chars + **** + last 4 chars) + callback_vars = result["metadata"]["logging"][0]["callback_vars"] + assert callback_vars["langfuse_public_key"] == "pk-l**************cdef" + assert callback_vars["langfuse_secret_key"] == "sk-l**************dcba" + # Non-sensitive fields should not be masked + assert callback_vars["langfuse_host"] == "https://langfuse.example.com" + + def test_mask_callback_settings_callback_vars(self): + """Test that callback_vars in callback_settings are masked.""" + key_info = { + "token": "hashed_token", + "metadata": { + "callback_settings": { + "success_callback": ["langfuse"], + "callback_vars": { + "langfuse_public_key": "pk-lf-test-key-1234", + "langfuse_secret_key": "sk-lf-test-secret-5678", + }, + } + }, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + callback_vars = result["metadata"]["callback_settings"]["callback_vars"] + assert "****" in callback_vars["langfuse_public_key"] + assert "****" in callback_vars["langfuse_secret_key"] + + def test_no_metadata(self): + """Test that key_info without metadata is returned unchanged.""" + key_info = { + "token": "hashed_token", + "user_id": "user123", + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + assert result["token"] == "hashed_token" + assert result["user_id"] == "user123" + assert "metadata" not in result + + def test_empty_metadata(self): + """Test that key_info with empty metadata is handled correctly.""" + key_info = { + "token": "hashed_token", + "metadata": {}, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + assert result["metadata"] == {} + + def test_metadata_without_sensitive_data(self): + """Test that metadata without sensitive fields is not modified.""" + key_info = { + "token": "hashed_token", + "metadata": { + "team": "engineering", + "app": "my-app", + "tags": ["production", "api"], + }, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + assert result["metadata"]["team"] == "engineering" + assert result["metadata"]["app"] == "my-app" + assert result["metadata"]["tags"] == ["production", "api"] + + def test_mask_other_sensitive_fields(self): + """Test that other sensitive field patterns are also masked.""" + key_info = { + "token": "hashed_token", + "metadata": { + "api_key": "sk-1234567890abcdef", + "password": "super_secret_password", + "token": "bearer_token_123", + "secret": "my_secret_value", + "normal_field": "this_should_remain", + }, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + metadata = result["metadata"] + assert "****" in metadata["api_key"] + assert "****" in metadata["password"] + assert "****" in metadata["token"] + assert "****" in metadata["secret"] + assert metadata["normal_field"] == "this_should_remain" + + def test_original_key_info_not_modified(self): + """Test that the original key_info dict is not modified (deep copy works).""" + key_info = { + "token": "hashed_token", + "metadata": { + "logging": [ + { + "callback_vars": { + "langfuse_secret_key": "sk-lf-original-secret", + } + } + ] + }, + } + + result = _mask_sensitive_fields_in_key_info(key_info) + + # Verify original is unchanged + assert ( + key_info["metadata"]["logging"][0]["callback_vars"]["langfuse_secret_key"] + == "sk-lf-original-secret" + ) + # Verify result is masked + assert ( + result["metadata"]["logging"][0]["callback_vars"]["langfuse_secret_key"] + != "sk-lf-original-secret" + ) + + def test_none_key_info(self): + """Test that None key_info is handled gracefully.""" + result = _mask_sensitive_fields_in_key_info(None) + assert result is None + + def test_non_dict_key_info(self): + """Test that non-dict key_info is handled gracefully.""" + result = _mask_sensitive_fields_in_key_info("not a dict") + assert result == "not a dict"