chore(proxy): redact secrets on admin-readable config and audit endpoints

Several GET endpoints reachable by the read-only admin role (and, for some,
any authenticated key) returned decrypted operational secrets verbatim:

- /config/field/info returned master_key and database_url; these now 403
  (recovering master_key let a read-only caller mint a full admin key).
- /get/config/callbacks echoed decrypted callback variables; non-admins now
  get credential-bearing vars masked (full admins still see plaintext so the
  edit round-trip is unchanged).
- /cache/settings masked password fields but not passwords embedded in a
  connection url; those are now redacted too.
- /config/pass_through_endpoint returned forwarded auth headers verbatim;
  non-admins now get header values masked.
- /memory-usage-in-mem-cache-items dumped raw caches (including the
  user_api_key cache) to any authenticated caller; now restricted to admin.

Audit log values no longer carry usable credentials: the key create/rotate
writers drop the plaintext virtual key, and the /audit reader masks secrets
the write-time key-only masker leaves untouched (client_secret,
vertex_credentials, token, password), covering existing rows for every role.
This commit is contained in:
user 2026-05-30 07:00:01 +00:00
parent 8aa941b62a
commit d07a1b5a6f
No known key found for this signature in database
10 changed files with 532 additions and 37 deletions

View file

@ -7,6 +7,7 @@ GET - /audit/{id} - Get audit log by id
GET - /audit - Get all audit logs
"""
import json
from typing import Any, Dict, List, Optional
#### AUDIT LOGGING ####
@ -16,11 +17,43 @@ from litellm_enterprise.types.proxy.audit_logging_endpoints import (
PaginatedAuditLogResponse,
)
from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
router = APIRouter()
_AUDIT_VALUE_MASKER = SensitiveDataMasker()
def _redact_audit_log_values(audit_log_dict: Dict[str, Any]) -> Dict[str, Any]:
"""Mask credentials inside the ``before_value`` / ``updated_values`` row
snapshots before they leave the server.
Audit rows snapshot whole table rows, so they can carry provider
credentials (``client_secret``, ``vertex_credentials``, ``token``,
``password``) that the write-time ``LiteLLM_AuditLogs`` masker — which only
targets ``key``-named fields — leaves untouched. These columns are an
informational change-record, never a place to read a secret back from, so
they are masked here for every caller (this also covers rows written before
this masking existed). Input type is preserved (dict in -> dict out) so the
response shape does not change.
"""
for field in ("before_value", "updated_values"):
raw = audit_log_dict.get(field)
if isinstance(raw, dict):
audit_log_dict[field] = _AUDIT_VALUE_MASKER.mask_dict(raw)
elif isinstance(raw, str) and raw:
try:
parsed = json.loads(raw)
except (TypeError, ValueError):
continue
if isinstance(parsed, dict):
audit_log_dict[field] = json.dumps(
_AUDIT_VALUE_MASKER.mask_dict(parsed), default=str
)
return audit_log_dict
def _build_json_field_or_condition(json_key: str, value: str) -> Dict[str, Any]:
"""
@ -153,11 +186,14 @@ async def get_audit_logs(
# Return paginated response
return PaginatedAuditLogResponse(
audit_logs=[
AuditLogResponse(**audit_log.model_dump()) for audit_log in audit_logs
]
if audit_logs
else [],
audit_logs=(
[
AuditLogResponse(**_redact_audit_log_values(audit_log.model_dump()))
for audit_log in audit_logs
]
if audit_logs
else []
),
total=total_count,
page=page,
page_size=page_size,
@ -207,4 +243,4 @@ async def get_audit_log_by_id(
)
# Convert to response model
return AuditLogResponse(**audit_log.model_dump())
return AuditLogResponse(**_redact_audit_log_values(audit_log.model_dump()))

View file

@ -5,7 +5,11 @@ import litellm
from litellm import get_secret
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
from litellm.litellm_core_utils.sensitive_data_masker import (
SensitiveDataMasker,
mask_sensitive_keys,
mask_url_credentials,
)
from litellm.proxy._types import CommonProxyErrors, LiteLLMPromptInjectionParams
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
decrypt_value_helper,
@ -621,18 +625,27 @@ def add_guardrail_response_to_standard_logging_object(
def process_callback(
_callback: str, callback_type: str, environment_variables: dict
_callback: str,
callback_type: str,
environment_variables: dict,
mask_sensitive: bool = False,
) -> dict:
"""Process a single callback and return its data with environment variables"""
"""Process a single callback and return its data with environment variables.
When ``mask_sensitive`` is set, credential-bearing variables are redacted so
the response never hands decrypted callback secrets (provider API keys,
webhook URLs) to a caller who is not a full proxy admin.
"""
env_vars = CustomLogger.get_callback_env_vars(_callback)
env_vars_dict: dict[str, str | None] = {}
for _var in env_vars:
env_variable = environment_variables.get(_var, None)
if env_variable is None:
env_vars_dict[_var] = None
else:
env_vars_dict[_var] = env_variable
env_vars_dict[_var] = environment_variables.get(_var, None)
if mask_sensitive:
sensitive_keys = {k for k in env_vars_dict if _is_sensitive_callback_var(k)}
env_vars_dict = mask_sensitive_keys(env_vars_dict, sensitive_keys)
env_vars_dict = {k: mask_url_credentials(v) for k, v in env_vars_dict.items()}
return {"name": _callback, "variables": env_vars_dict, "type": callback_type}

View file

@ -13,7 +13,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query
from litellm import get_secret_str
from litellm._logging import verbose_proxy_logger
from litellm.constants import PYTHON_GC_THRESHOLD
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
router = APIRouter()
@ -167,7 +167,7 @@ async def memory_usage_in_mem_cache(
@router.get("/memory-usage-in-mem-cache-items", include_in_schema=False)
async def memory_usage_in_mem_cache_items(
_: UserAPIKeyAuth = Depends(user_api_key_auth),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
# returns the size of all in-memory caches on the proxy server
"""
@ -176,6 +176,15 @@ async def memory_usage_in_mem_cache_items(
3. proxy_logging_cache
4. internal_usage_cache
"""
# The response dumps raw cache contents, including the user_api_key_cache
# which holds validated auth objects. Restrict to full PROXY_ADMIN so a
# read-only admin (or any authenticated key) cannot harvest them.
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise HTTPException(
status_code=403,
detail={"error": "Only proxy admin can access this endpoint."},
)
from litellm.proxy.proxy_server import (
llm_router,
proxy_logging_obj,

View file

@ -23,6 +23,20 @@ from litellm.proxy._types import (
LITELLM_PREFIX_STORED_VIRTUAL_KEYS = "litellm/"
def _audit_values_without_plaintext_key(response: GenerateKeyResponse) -> str:
"""Serialize a key response for the audit log with the plaintext key removed.
``GenerateKeyResponse.key`` is the usable virtual key (sk-...); persisting it
verbatim turns the audit log into a credential store that any audit reader
can harvest. The object_id / token_id still record which key the event is
about, so the change-record stays meaningful without the secret.
"""
data = response.model_dump(exclude_none=True)
if data.get("key"):
data["key"] = "***REDACTED***"
return json.dumps(data, default=str)
class KeyManagementEventHooks:
@staticmethod
async def async_key_generated_hook(
@ -56,7 +70,7 @@ class KeyManagementEventHooks:
# Enterprise Feature - Audit Logging. Enable with litellm.store_audit_logs = True
if litellm.store_audit_logs is True:
_updated_values = response.model_dump_json(exclude_none=True)
_updated_values = _audit_values_without_plaintext_key(response)
asyncio.create_task(
create_audit_log_for_update(
request_data=LiteLLM_AuditLogs(
@ -205,7 +219,7 @@ class KeyManagementEventHooks:
table_name=LitellmTableNames.KEY_TABLE_NAME,
object_id=existing_key_row.token,
action="rotated",
updated_values=response.model_dump_json(exclude_none=True),
updated_values=_audit_values_without_plaintext_key(response),
before_value=existing_key_row.model_dump_json(
exclude_none=True
),

View file

@ -19,7 +19,10 @@ from pydantic import BaseModel, Field
import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.litellm_core_utils.sensitive_data_masker import mask_sensitive_keys
from litellm.litellm_core_utils.sensitive_data_masker import (
mask_sensitive_keys,
mask_url_credentials,
)
from litellm.proxy._types import (
AUDIT_ACTIONS,
LiteLLM_AuditLogs,
@ -301,10 +304,16 @@ async def get_cache_settings(
decrypted_settings["redis_type"] = "node"
# Mask credential fields so the GET response never carries
# plaintext Redis / Sentinel passwords off the server.
# plaintext Redis / Sentinel passwords off the server. Also
# redact passwords smuggled inside connection-URL fields
# (e.g. a `url` value like redis://:password@host), which
# exact-name field masking does not catch.
current_values = mask_sensitive_keys(
decrypted_settings, _CACHE_SENSITIVE_FIELDS
)
current_values = {
k: mask_url_credentials(v) for k, v in current_values.items()
}
# Update field values with current values
for field in cache_fields:

View file

@ -38,12 +38,14 @@ from litellm.constants import MAXIMUM_TRACEBACK_LINES_TO_LOG
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.litellm_core_utils.sensitive_data_masker import mask_sensitive_keys
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.passthrough import BasePassthroughUtils
from litellm.proxy._types import (
ConfigFieldInfo,
ConfigFieldUpdate,
LiteLLMRoutes,
LitellmUserRoles,
PassThroughEndpointResponse,
PassThroughGenericEndpoint,
ProxyException,
@ -2632,6 +2634,21 @@ async def _filter_endpoints_by_team_allowed_routes(
return pass_through_endpoints
def _mask_pass_through_endpoint_headers(
endpoint: PassThroughGenericEndpoint,
) -> PassThroughGenericEndpoint:
"""Return a copy of the endpoint with forwarded header values masked.
Header names stay visible so a read-only caller can still see which headers
are configured; the values (which carry upstream credentials) are redacted.
"""
if not endpoint.headers:
return endpoint
masked = endpoint.model_copy(deep=True)
masked.headers = mask_sensitive_keys(masked.headers, set(masked.headers.keys()))
return masked
@router.get(
"/config/pass_through_endpoint",
dependencies=[Depends(user_api_key_auth)],
@ -2685,6 +2702,15 @@ async def get_pass_through_endpoints(
prisma_client=prisma_client,
)
# Forwarded headers carry upstream credentials (Authorization, x-api-key).
# A full proxy admin can edit them and needs the plaintext; everyone else
# (read-only admin, team callers) gets the values masked.
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
pass_through_endpoints = [
_mask_pass_through_endpoint_headers(endpoint)
for endpoint in pass_through_endpoints
]
return PassThroughEndpointResponse(endpoints=pass_through_endpoints)

View file

@ -14251,6 +14251,13 @@ async def update_config_general_settings(
return response
# general_settings fields that hold credentials. These must never be returned
# verbatim through the read endpoint: it is reachable by PROXY_ADMIN_VIEW_ONLY,
# and the master_key in particular would let a read-only caller mint a full
# admin key.
_SECRET_CONFIG_GENERAL_SETTINGS_FIELDS = {"master_key", "database_url"}
@router.get(
"/config/field/info",
tags=["config.yaml"],
@ -14288,6 +14295,16 @@ async def get_config_general_settings(
detail={"error": "Invalid field={} passed in.".format(field_name)},
)
if field_name in _SECRET_CONFIG_GENERAL_SETTINGS_FIELDS:
raise HTTPException(
status_code=403,
detail={
"error": "Field={} holds a credential and cannot be retrieved through this endpoint.".format(
field_name
)
},
)
## get general settings from db
db_general_settings = await prisma_client.db.litellm_config.find_first(
where={"param_name": "general_settings"}
@ -14634,7 +14651,9 @@ async def delete_callback(
include_in_schema=False,
dependencies=[Depends(user_api_key_auth)],
)
async def get_config(): # noqa: PLR0915
async def get_config( # noqa: PLR0915
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
For Admin UI - allows admin to view config via UI
# return the callbacks and the env variables for the callback
@ -14646,6 +14665,13 @@ async def get_config(): # noqa: PLR0915
all_available_callbacks = AllCallbacks()
# Callback variables are decrypted config secrets (provider keys, webhook
# URLs). Only a full proxy admin (who can also edit them) sees plaintext;
# a read-only admin gets them masked.
mask_callback_secrets = (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
)
config_data = await proxy_config.get_config()
_litellm_settings = config_data.get("litellm_settings", {})
_general_settings = config_data.get("general_settings", {})
@ -14685,22 +14711,20 @@ async def get_config(): # noqa: PLR0915
"""
for _callback in _success_callbacks:
_data_to_return.append(
process_callback(_callback, "success", environment_variables)
)
for _callback in _failure_callbacks:
_data_to_return.append(
process_callback(_callback, "failure", environment_variables)
)
for _callback in _success_and_failure_callbacks:
_data_to_return.append(
process_callback(
_callback, "success_and_failure", environment_variables
for _callbacks, _callback_type in (
(_success_callbacks, "success"),
(_failure_callbacks, "failure"),
(_success_and_failure_callbacks, "success_and_failure"),
):
for _callback in _callbacks:
_data_to_return.append(
process_callback(
_callback,
_callback_type,
environment_variables,
mask_sensitive=mask_callback_secrets,
)
)
)
# Check if slack alerting is on
_alerting = _general_settings.get("alerting", [])

View file

@ -82,6 +82,66 @@ def test_process_callback_with_no_required_env_vars(mock_get_env_vars):
assert result["variables"] == {}
@patch(
"litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars",
return_value=["LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"],
)
def test_process_callback_masks_secrets_when_requested(mock_get_env_vars):
"""With mask_sensitive=True the credential-bearing var is redacted while a
non-secret routing var (host) is returned verbatim."""
environment_variables = {
"LANGFUSE_SECRET_KEY": "sk-langfuse-supersecret-value",
"LANGFUSE_HOST": "https://cloud.langfuse.com",
}
result = process_callback(
_callback="langfuse",
callback_type="success",
environment_variables=environment_variables,
mask_sensitive=True,
)
masked_secret = result["variables"]["LANGFUSE_SECRET_KEY"]
assert masked_secret != "sk-langfuse-supersecret-value"
assert "supersecret" not in masked_secret
assert "*" in masked_secret
# Non-credential routing field is left intact.
assert result["variables"]["LANGFUSE_HOST"] == "https://cloud.langfuse.com"
@patch(
"litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars",
return_value=["LANGFUSE_SECRET_KEY"],
)
def test_process_callback_returns_plaintext_without_masking(mock_get_env_vars):
"""Default (mask_sensitive=False) preserves the prior plaintext behaviour so
a full admin can still read/edit the value."""
result = process_callback(
_callback="langfuse",
callback_type="success",
environment_variables={"LANGFUSE_SECRET_KEY": "sk-langfuse-supersecret-value"},
)
assert result["variables"]["LANGFUSE_SECRET_KEY"] == "sk-langfuse-supersecret-value"
@patch(
"litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars",
return_value=["GENERIC_LOGGER_URL"],
)
def test_process_callback_masks_url_embedded_credentials(mock_get_env_vars):
"""A password smuggled into a connection-URL callback var is redacted even
though the field name itself is not credential-like."""
result = process_callback(
_callback="generic",
callback_type="success",
environment_variables={
"GENERIC_LOGGER_URL": "https://user:topsecretpw@logs.example.com/ingest"
},
mask_sensitive=True,
)
assert "topsecretpw" not in result["variables"]["GENERIC_LOGGER_URL"]
def test_normalize_callback_names_none_returns_empty_list():
assert normalize_callback_names(None) == []
assert normalize_callback_names([]) == []

View file

@ -1374,6 +1374,7 @@ async def test_get_pass_through_endpoints_includes_config_and_db():
DB endpoints have is_from_config=False. When same path exists in both, DB overrides.
"""
from litellm.proxy._types import (
LitellmUserRoles,
PassThroughEndpointResponse,
PassThroughGenericEndpoint,
UserAPIKeyAuth,
@ -1436,7 +1437,7 @@ async def test_get_pass_through_endpoints_includes_config_and_db():
mock_get_db.return_value = db_objects
mock_get_config.return_value = config_objects
mock_user = MagicMock(spec=UserAPIKeyAuth)
mock_user = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
result = await get_pass_through_endpoints(
endpoint_id=None,

View file

@ -0,0 +1,303 @@
"""
Regression tests: management/read endpoints must not disclose decrypted secrets
to callers who are not a full proxy admin (read-only admins, org admins, plain
keys), and audit logs must not carry usable credentials.
"""
import json
import os
import sys
import pytest
from fastapi import HTTPException
from unittest.mock import AsyncMock, MagicMock, patch
sys.path.insert(0, os.path.abspath("../.."))
from litellm.proxy._types import GenerateKeyResponse, LitellmUserRoles, UserAPIKeyAuth
def _user(role: LitellmUserRoles) -> UserAPIKeyAuth:
return UserAPIKeyAuth(user_role=role, api_key="hashed-token", user_id="u1")
# --------------------------------------------------------------------------- #
# Audit log writer / reader redaction
# --------------------------------------------------------------------------- #
def test_audit_values_without_plaintext_key_redacts_usable_key():
from litellm.proxy.hooks.key_management_event_hooks import (
_audit_values_without_plaintext_key,
)
response = GenerateKeyResponse(key="sk-this-is-a-usable-virtual-key")
serialized = _audit_values_without_plaintext_key(response)
data = json.loads(serialized)
assert data["key"] == "***REDACTED***"
assert "sk-this-is-a-usable-virtual-key" not in serialized
def test_redact_audit_log_values_masks_provider_secrets_dict_input():
"""Reader masks the secrets the write-time key-only masker misses, and keeps
dict input as a dict (response shape is unchanged)."""
from litellm_enterprise.proxy.audit_logging_endpoints import (
_redact_audit_log_values,
)
row = {
"id": "a1",
"updated_values": {
"key": "sk-plaintext-virtual-key",
"api_key": "sk-provider-plaintext-key",
"client_secret": "client-secret-plaintext",
"vertex_credentials": "vertex-credentials-plaintext",
"token": "tok-plaintext",
"user_id": "user-123",
"model_name": "gpt-4",
},
"before_value": None,
}
out = _redact_audit_log_values(row)
masked = out["updated_values"]
assert isinstance(masked, dict) # dict in -> dict out
blob = json.dumps(masked)
for secret in (
"sk-plaintext-virtual-key",
"sk-provider-plaintext-key",
"client-secret-plaintext",
"vertex-credentials-plaintext",
"tok-plaintext",
):
assert secret not in blob, f"{secret} leaked through /audit"
# Non-secret operational fields stay readable.
assert masked["user_id"] == "user-123"
assert masked["model_name"] == "gpt-4"
def test_redact_audit_log_values_handles_string_and_non_dict_input():
from litellm_enterprise.proxy.audit_logging_endpoints import (
_redact_audit_log_values,
)
# JSON string input -> masked JSON string output (type preserved).
row = {"before_value": json.dumps({"client_secret": "should-be-masked"})}
out = _redact_audit_log_values(row)
assert isinstance(out["before_value"], str)
assert "should-be-masked" not in out["before_value"]
# Non-dict / unparseable / None values must pass through untouched.
assert _redact_audit_log_values({"before_value": None})["before_value"] is None
assert (
_redact_audit_log_values({"updated_values": "not-json"})["updated_values"]
== "not-json"
)
# --------------------------------------------------------------------------- #
# Pass-through endpoint header masking
# --------------------------------------------------------------------------- #
def _make_endpoint(headers):
from litellm.proxy._types import PassThroughGenericEndpoint
return PassThroughGenericEndpoint(
path="/foo", target="https://upstream.example.com", headers=headers
)
def test_mask_pass_through_endpoint_headers_redacts_values_keeps_names():
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
_mask_pass_through_endpoint_headers,
)
ep = _make_endpoint({"Authorization": "Bearer sk-upstream-secret-token"})
masked = _mask_pass_through_endpoint_headers(ep)
assert "Authorization" in masked.headers # name preserved
assert masked.headers["Authorization"] != "Bearer sk-upstream-secret-token"
assert "sk-upstream-secret-token" not in masked.headers["Authorization"]
# Original object is not mutated.
assert ep.headers["Authorization"] == "Bearer sk-upstream-secret-token"
@pytest.mark.asyncio
async def test_pass_through_get_masks_headers_for_non_admin_only():
import litellm.proxy.pass_through_endpoints.pass_through_endpoints as pt
secret = "Bearer sk-upstream-secret-token"
endpoints = [_make_endpoint({"Authorization": secret})]
with (
patch.object(
pt, "_get_pass_through_endpoints_from_db", AsyncMock(return_value=endpoints)
),
patch.object(
pt, "_get_pass_through_endpoints_from_config", MagicMock(return_value=[])
),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
):
viewer_resp = await pt.get_pass_through_endpoints(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
)
admin_resp = await pt.get_pass_through_endpoints(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN)
)
assert secret not in json.dumps(viewer_resp.endpoints[0].headers)
# Full admin can still read the plaintext header (it is editable for them).
assert admin_resp.endpoints[0].headers["Authorization"] == secret
# --------------------------------------------------------------------------- #
# /config/field/info must not return credential general-settings fields
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
@pytest.mark.parametrize("field_name", ["master_key", "database_url"])
async def test_config_field_info_blocks_secret_fields(field_name):
from litellm.proxy.proxy_server import get_config_general_settings
with patch("litellm.proxy.proxy_server.prisma_client", MagicMock()):
with pytest.raises(HTTPException) as exc:
await get_config_general_settings(
field_name=field_name,
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY),
)
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_config_field_info_allows_non_secret_field():
from litellm.proxy.proxy_server import get_config_general_settings
db_row = MagicMock()
db_row.param_value = {"completion_model": "gpt-4"}
mock_prisma = MagicMock()
mock_prisma.db.litellm_config.find_first = AsyncMock(return_value=db_row)
with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma):
result = await get_config_general_settings(
field_name="completion_model",
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY),
)
assert result.field_value == "gpt-4"
# --------------------------------------------------------------------------- #
# /memory-usage-in-mem-cache-items must require a full proxy admin
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
@pytest.mark.parametrize(
"role",
[LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, LitellmUserRoles.INTERNAL_USER],
)
async def test_memory_usage_endpoint_blocks_non_admins(role):
from litellm.proxy.common_utils.debug_utils import memory_usage_in_mem_cache_items
with pytest.raises(HTTPException) as exc:
await memory_usage_in_mem_cache_items(user_api_key_dict=_user(role))
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_memory_usage_endpoint_allows_proxy_admin():
from litellm.proxy.common_utils.debug_utils import memory_usage_in_mem_cache_items
with (
patch("litellm.proxy.proxy_server.llm_router", None),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
):
result = await memory_usage_in_mem_cache_items(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN)
)
assert "user_api_key_cache" in result
# --------------------------------------------------------------------------- #
# /cache/settings must redact a password embedded in a connection URL
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_cache_settings_masks_url_embedded_password():
from litellm.proxy.management_endpoints.cache_settings_endpoints import (
get_cache_settings,
)
stored = {"type": "redis", "url": "redis://:redisPlaintextPw@cache.internal:6379"}
mock_cache_config = MagicMock()
mock_cache_config.cache_settings = json.dumps(stored)
mock_prisma = MagicMock()
mock_prisma.db.litellm_cacheconfig.find_unique = AsyncMock(
return_value=mock_cache_config
)
mock_proxy_config = MagicMock()
mock_proxy_config._decrypt_db_variables = MagicMock(return_value=dict(stored))
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma),
patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
):
result = await get_cache_settings(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
)
assert "redisPlaintextPw" not in json.dumps(result.current_values)
assert result.current_values["url"] == "redis://:****@cache.internal:6379"
# --------------------------------------------------------------------------- #
# /get/config/callbacks masks callback secrets for non-admins only
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_get_config_callbacks_masks_secret_for_view_only_admin():
import litellm.proxy.proxy_server as ps
config = {
"litellm_settings": {"success_callback": ["langfuse"]},
"general_settings": {},
"environment_variables": {
"LANGFUSE_SECRET_KEY": "sk-langfuse-supersecret-value",
"LANGFUSE_HOST": "https://cloud.langfuse.com",
},
}
mock_proxy_config = MagicMock()
mock_proxy_config.get_config = AsyncMock(return_value=config)
with (
patch.object(ps, "proxy_config", mock_proxy_config),
patch.object(ps, "llm_router", None),
patch(
"litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars",
return_value=["LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"],
),
):
viewer = await ps.get_config(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
)
admin = await ps.get_config(
user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN)
)
viewer_secret = viewer["callbacks"][0]["variables"]["LANGFUSE_SECRET_KEY"]
admin_secret = admin["callbacks"][0]["variables"]["LANGFUSE_SECRET_KEY"]
assert "supersecret" not in viewer_secret
# Full admin keeps plaintext (they can edit it; round-trip stays correct).
assert admin_secret == "sk-langfuse-supersecret-value"
# Non-secret routing var is never masked.
assert (
viewer["callbacks"][0]["variables"]["LANGFUSE_HOST"]
== "https://cloud.langfuse.com"
)