chore(audit): audit-log /cache/settings + /config_overrides/hashicorp_vault mutations

Follow-up to merged #26859 (team-callback audit log).  The variant
scan from that PR flagged two more high-risk admin endpoints whose
mutations weren't audit-logged:

- ``POST /cache/settings`` (cache_settings_endpoints.py) — writes the
  team / global Redis cache configuration, including credentials.  An
  admin (or compromised admin) flipping the cache backend is a
  data-routing pivot — every subsequent LLM response cache write goes
  to the new destination — so the action needs to be traceable.

- ``POST /config_overrides/hashicorp_vault`` and
  ``DELETE /config_overrides/hashicorp_vault``
  (config_override_endpoints.py) — control the proxy's KMS config.
  Mutating these affects every secret retrieval going forward.

Each endpoint now emits an ``LiteLLM_AuditLogs`` row gated on
``litellm.store_audit_logs`` (Enterprise feature), mirroring the
shape merged in #26859.  Both helpers redact every field value
before serialization (replacing them with ``***REDACTED***`` while
keeping field names) so the audit table cannot itself become a
credential-harvest sink — Redis passwords / vault tokens /
``approle_secret_id`` / ``client_key`` would otherwise be persisted
in plaintext JSON.

Both helpers also attach a ``done_callback`` that surfaces a
``verbose_proxy_logger.warning`` when the fire-and-forget audit-log
task fails, so a transient DB error doesn't silently lose the row.

Adds ``CACHE_CONFIG_TABLE_NAME`` / ``CONFIG_OVERRIDES_TABLE_NAME`` to
``LitellmTableNames`` so the audit rows co-locate with the table they
mutate.

Tests:
- /cache/settings: emits when ``store_audit_logs=True``, no emission
  when off, plaintext credentials don't appear in the serialized row.
- /config_overrides/hashicorp_vault POST: same, plus checks the
  redaction of ``vault_token`` and ``vault_addr``.
- /config_overrides/hashicorp_vault DELETE: emits with ``action="deleted"``
  only when an actual row was removed; idempotent delete of a non-existent
  row produces no audit log.
This commit is contained in:
user 2026-05-01 01:32:41 +00:00
parent 05e6402bdb
commit ddc50e026e
5 changed files with 548 additions and 6 deletions

View file

@ -190,6 +190,8 @@ class LitellmTableNames(str, enum.Enum):
PROXY_MODEL_TABLE_NAME = "LiteLLM_ProxyModelTable"
MANAGED_FILE_TABLE_NAME = "LiteLLM_ManagedFileTable"
TOOL_TABLE_NAME = "LiteLLM_ToolTable"
CACHE_CONFIG_TABLE_NAME = "LiteLLM_CacheConfig"
CONFIG_OVERRIDES_TABLE_NAME = "LiteLLM_ConfigOverrides"
class Litellm_EntityType(enum.Enum):

View file

@ -8,14 +8,22 @@ POST /cache/settings/test - Test cache connection with provided credentials
POST /cache/settings - Save cache settings to database
"""
import asyncio
import json
from typing import Any, Dict, List, Optional
from datetime import datetime, timezone
from typing import Any, Dict, List, Mapping, Optional
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel, Field
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import UserAPIKeyAuth
from litellm._uuid import uuid
from litellm.proxy._types import (
LiteLLM_AuditLogs,
LitellmTableNames,
UserAPIKeyAuth,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.types.management_endpoints import (
CACHE_SETTINGS_FIELDS,
@ -26,6 +34,85 @@ from litellm.types.management_endpoints import (
router = APIRouter()
_REDACTED_VALUE = "***REDACTED***"
def _redact_settings(settings: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
"""Replace every value in a settings map with a fixed marker.
Cache config carries Redis credentials (passwords, connection strings).
The audit-log row preserves the field names so a reader can see *which*
fields changed, but values are stripped so the audit table can't itself
become a credential-harvest sink.
"""
if not settings:
return {}
return {k: _REDACTED_VALUE for k in settings.keys()}
def _log_audit_task_exception(task: "asyncio.Task[None]") -> None:
"""Surface a fire-and-forget audit-log task failure as a warning.
``asyncio.create_task`` swallows exceptions silently if the audit
write fails we'd otherwise lose the row without any signal.
"""
if task.cancelled():
return
exc = task.exception()
if exc is not None:
verbose_proxy_logger.warning(
"Failed to write cache-settings audit log: %s", exc
)
async def _emit_cache_settings_audit_log(
*,
action: str,
before_settings: Optional[Mapping[str, Any]],
after_settings: Optional[Mapping[str, Any]],
user_api_key_dict: UserAPIKeyAuth,
litellm_changed_by: Optional[str],
) -> None:
"""Emit an audit-log row for a /cache/settings mutation.
Mirrors the ``store_audit_logs``-gated pattern used in
``team_callback_endpoints.py``: fire-and-forget, no-op when audit
logging is disabled, with a done-callback that surfaces any task
exception. Captured under ``LiteLLM_CacheConfig`` so the row
co-locates with the table it mutates.
"""
if litellm.store_audit_logs is not True:
return
from litellm.proxy.management_helpers.audit_logs import (
create_audit_log_for_update,
)
from litellm.proxy.proxy_server import litellm_proxy_admin_name
task = asyncio.create_task(
create_audit_log_for_update(
request_data=LiteLLM_AuditLogs(
id=str(uuid.uuid4()),
updated_at=datetime.now(timezone.utc),
changed_by=litellm_changed_by
or user_api_key_dict.user_id
or litellm_proxy_admin_name,
changed_by_api_key=user_api_key_dict.api_key,
table_name=LitellmTableNames.CACHE_CONFIG_TABLE_NAME,
object_id="cache_config",
action=action,
updated_values=json.dumps(
{"settings": _redact_settings(after_settings)}, default=str
),
before_value=json.dumps(
{"settings": _redact_settings(before_settings)}, default=str
),
)
)
)
task.add_done_callback(_log_audit_task_exception)
class CacheSettingsManager:
"""
Manages cache settings initialization and updates.
@ -282,6 +369,10 @@ async def test_cache_connection(
async def update_cache_settings(
request: CacheSettingsUpdateRequest,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
litellm_changed_by: Optional[str] = Header(
None,
description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
),
):
"""
Save cache settings to database and initialize cache.
@ -314,6 +405,19 @@ async def update_cache_settings(
try:
cache_settings = request.cache_settings.copy()
# Snapshot the prior settings (key set only — values get redacted in
# the audit row) so the audit-log entry shows which fields changed.
existing_row = await prisma_client.db.litellm_cacheconfig.find_unique(
where={"id": "cache_config"}
)
before_settings: Optional[Dict[str, Any]] = None
if existing_row is not None and existing_row.cache_settings:
try:
before_settings = json.loads(existing_row.cache_settings)
except (TypeError, ValueError):
before_settings = None
action = "updated" if existing_row is not None else "created"
# Encrypt sensitive fields (keep redis_type for storage)
encrypted_settings = proxy_config._encrypt_env_variables(
environment_variables=cache_settings
@ -353,6 +457,18 @@ async def update_cache_settings(
# Switch on LLM response caching
proxy_config.switch_on_llm_response_caching()
# Cache settings carry Redis credentials and connection strings that
# control where LLM responses are cached. An admin (or compromised
# admin) flipping the cache backend silently is a data-routing
# pivot; emit an audit-log row so the action is traceable.
await _emit_cache_settings_audit_log(
action=action,
before_settings=before_settings,
after_settings=cache_settings,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=litellm_changed_by,
)
return {
"message": "Cache settings updated successfully",
"status": "success",

View file

@ -1,10 +1,13 @@
import asyncio
import json
import os
from typing import Any, Dict, Set
from datetime import datetime, timezone
from typing import Any, Dict, Mapping, Optional, Set
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import TypeAdapter
from litellm._uuid import uuid
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
@ -20,6 +23,8 @@ from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.proxy._types import (
CommonProxyErrors,
KeyManagementSystem,
LiteLLM_AuditLogs,
LitellmTableNames,
LitellmUserRoles,
UserAPIKeyAuth,
)
@ -32,6 +37,82 @@ from litellm.types.proxy.management_endpoints.config_overrides import (
router = APIRouter()
_AUDIT_REDACTED = "***REDACTED***"
def _redact_config(config: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
"""Strip values from a config snapshot before audit-log emission.
Hashicorp Vault config carries ``vault_token``, ``approle_secret_id``,
``client_key`` etc. Persisting them verbatim into ``LiteLLM_AuditLogs``
would let anyone with read access to the audit table harvest the
proxy's KMS credentials. Keep keys, redact values.
"""
if not config:
return {}
return {k: _AUDIT_REDACTED for k in config.keys()}
def _log_audit_task_exception(task: "asyncio.Task[None]") -> None:
if task.cancelled():
return
exc = task.exception()
if exc is not None:
verbose_proxy_logger.warning(
"Failed to write hashicorp-vault config audit log: %s", exc
)
async def _emit_hashicorp_vault_audit_log(
*,
action: str,
before_config: Optional[Mapping[str, Any]],
after_config: Optional[Mapping[str, Any]],
user_api_key_dict: UserAPIKeyAuth,
litellm_changed_by: Optional[str],
) -> None:
"""Emit an audit-log row for a /config_overrides/hashicorp_vault mutation.
Mirrors the ``store_audit_logs``-gated pattern from
``team_callback_endpoints.py``. Captured under
``LiteLLM_ConfigOverrides`` so the row co-locates with the table it
mutates.
"""
import litellm
if litellm.store_audit_logs is not True:
return
from litellm.proxy.management_helpers.audit_logs import (
create_audit_log_for_update,
)
from litellm.proxy.proxy_server import litellm_proxy_admin_name
task = asyncio.create_task(
create_audit_log_for_update(
request_data=LiteLLM_AuditLogs(
id=str(uuid.uuid4()),
updated_at=datetime.now(timezone.utc),
changed_by=litellm_changed_by
or user_api_key_dict.user_id
or litellm_proxy_admin_name,
changed_by_api_key=user_api_key_dict.api_key,
table_name=LitellmTableNames.CONFIG_OVERRIDES_TABLE_NAME,
object_id="hashicorp_vault",
action=action,
updated_values=json.dumps(
{"config": _redact_config(after_config)}, default=str
),
before_value=json.dumps(
{"config": _redact_config(before_config)}, default=str
),
)
)
)
task.add_done_callback(_log_audit_task_exception)
# --- Hashicorp Vault constants ---
HASHICORP_ENV_VAR_MAPPING: Dict[str, str] = {
@ -144,6 +225,10 @@ def _clear_hashicorp_vault_state(proxy_config: Any) -> None:
async def update_hashicorp_vault_config(
config: HashicorpVaultConfig,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
litellm_changed_by: Optional[str] = Header(
None,
description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
),
):
"""
Update Hashicorp Vault secret manager configuration.
@ -248,6 +333,18 @@ async def update_hashicorp_vault_config(
# Update change-detection cache so the background reload doesn't redundantly re-init
proxy_config._last_hashicorp_vault_config = safe_json_loads(config_value)
# Mutating the proxy's KMS config affects every secret retrieval going
# forward — emit an audit-log row so the action is traceable even
# though the secret_manager_client itself was just swapped under us.
before_config = existing_decrypted if existing_record is not None else env_values
await _emit_hashicorp_vault_audit_log(
action="updated" if existing_record is not None else "created",
before_config=before_config,
after_config=config_data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=litellm_changed_by,
)
return {
"message": "Hashicorp Vault configuration updated successfully",
"status": "success",
@ -319,6 +416,10 @@ async def get_hashicorp_vault_config(
)
async def delete_hashicorp_vault_config(
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
litellm_changed_by: Optional[str] = Header(
None,
description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
),
):
"""Delete Hashicorp Vault configuration. Idempotent."""
from litellm.proxy.proxy_server import prisma_client, proxy_config
@ -335,11 +436,27 @@ async def delete_hashicorp_vault_config(
detail=CommonProxyErrors.db_not_connected_error.value,
)
# Capture the prior config before delete so the audit-log row can
# show *what* was removed (keys only — values get redacted).
existing_record = await prisma_client.db.litellm_configoverrides.find_unique(
where={"config_type": "hashicorp_vault"}
)
before_config: Optional[Dict[str, Any]] = None
if existing_record is not None and existing_record.config_value is not None:
try:
before_config = proxy_config._decrypt_db_variables(
_parse_config_value(existing_record.config_value)
)
except Exception:
before_config = None
# Delete DB record if it exists — ignore if not found
deleted = False
try:
await prisma_client.db.litellm_configoverrides.delete(
where={"config_type": "hashicorp_vault"}
)
deleted = True
except RecordNotFoundError:
verbose_proxy_logger.debug(
"No existing Hashicorp Vault config record to delete"
@ -347,6 +464,17 @@ async def delete_hashicorp_vault_config(
_clear_hashicorp_vault_state(proxy_config)
# Only emit audit log if a row was actually removed; an idempotent
# delete on a non-existent row produces no security-relevant change.
if deleted:
await _emit_hashicorp_vault_audit_log(
action="deleted",
before_config=before_config,
after_config=None,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=litellm_changed_by,
)
return {
"message": "Hashicorp Vault configuration deleted successfully",
"status": "success",

View file

@ -12,12 +12,17 @@ sys.path.insert(
0, os.path.abspath("../../../..")
) # Adds the parent directory to the system path
from litellm.proxy._types import LitellmUserRoles
import json
import litellm
from litellm.proxy._types import LitellmTableNames, LitellmUserRoles
from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth
from litellm.proxy.management_endpoints.cache_settings_endpoints import (
CacheSettingsManager,
CacheSettingsUpdateRequest,
CacheTestRequest,
test_cache_connection,
update_cache_settings,
)
@ -253,3 +258,144 @@ class TestCacheSettingsManager:
# Verify cache was NOT initialized (params unchanged)
mock_proxy_config._init_cache.assert_not_called()
mock_proxy_config.switch_on_llm_response_caching.assert_not_called()
# ── Audit-log emission for /cache/settings ────────────────────────────────────
def _admin_auth() -> UserAPIKeyAuth:
return UserAPIKeyAuth(
api_key="hashed",
user_id="admin-user",
user_role=LitellmUserRoles.PROXY_ADMIN,
)
@pytest.mark.asyncio
async def test_update_cache_settings_emits_audit_log_when_enabled(monkeypatch):
"""Cache config carries Redis credentials; mutation must emit an
audit-log row when ``store_audit_logs`` is on, with values redacted."""
monkeypatch.setattr(litellm, "store_audit_logs", True)
mock_prisma = MagicMock()
mock_prisma.db.litellm_cacheconfig.find_unique = AsyncMock(return_value=None)
mock_prisma.db.litellm_cacheconfig.upsert = AsyncMock()
proxy_config = MagicMock()
proxy_config._encrypt_env_variables = MagicMock(
side_effect=lambda environment_variables: dict(environment_variables)
)
proxy_config._decrypt_db_variables = MagicMock(
side_effect=lambda variables_dict: dict(variables_dict)
)
proxy_config._init_cache = MagicMock()
proxy_config.switch_on_llm_response_caching = MagicMock()
audit_calls = []
async def capture(request_data):
audit_calls.append(request_data)
with (
patch(
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch(
"litellm.proxy.proxy_server.proxy_config",
proxy_config,
),
patch(
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
patch(
"litellm.proxy.management_helpers.audit_logs.create_audit_log_for_update",
new=capture,
),
):
await update_cache_settings(
request=CacheSettingsUpdateRequest(
cache_settings={
"type": "redis",
"host": "redis.example.com",
"password": "super-secret-redis-pw",
}
),
user_api_key_dict=_admin_auth(),
litellm_changed_by=None,
)
# asyncio.create_task fires the coroutine eagerly; await one tick to let
# the audit-log emit run before the test exits.
for _ in range(3):
await asyncio.sleep(0)
assert len(audit_calls) == 1
log = audit_calls[0]
assert log.table_name == LitellmTableNames.CACHE_CONFIG_TABLE_NAME
assert log.object_id == "cache_config"
assert log.action == "created" # no existing row → create
after = json.loads(log.updated_values)
# Field names are preserved so an auditor can see what changed.
assert set(after["settings"].keys()) == {"type", "host", "password"}
# Plaintext values must NOT appear in the serialized row.
assert "super-secret-redis-pw" not in log.updated_values
assert "redis.example.com" not in log.updated_values
@pytest.mark.asyncio
async def test_update_cache_settings_no_audit_when_disabled(monkeypatch):
monkeypatch.setattr(litellm, "store_audit_logs", False)
mock_prisma = MagicMock()
mock_prisma.db.litellm_cacheconfig.find_unique = AsyncMock(return_value=None)
mock_prisma.db.litellm_cacheconfig.upsert = AsyncMock()
proxy_config = MagicMock()
proxy_config._encrypt_env_variables = MagicMock(
side_effect=lambda environment_variables: dict(environment_variables)
)
proxy_config._decrypt_db_variables = MagicMock(
side_effect=lambda variables_dict: dict(variables_dict)
)
proxy_config._init_cache = MagicMock()
proxy_config.switch_on_llm_response_caching = MagicMock()
audit_calls = []
async def capture(request_data):
audit_calls.append(request_data)
with (
patch(
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch(
"litellm.proxy.proxy_server.proxy_config",
proxy_config,
),
patch(
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch(
"litellm.proxy.management_helpers.audit_logs.create_audit_log_for_update",
new=capture,
),
):
await update_cache_settings(
request=CacheSettingsUpdateRequest(
cache_settings={"type": "redis", "host": "redis.example.com"}
),
user_api_key_dict=_admin_auth(),
litellm_changed_by=None,
)
assert audit_calls == []
# Need an asyncio import for the eager-task drain pattern above.
import asyncio # noqa: E402

View file

@ -272,3 +272,153 @@ async def test_hashicorp_vault_validation_errors_and_access_control(
litellm.secret_manager_client = old_client
litellm._key_management_system = old_kms
_cleanup()
# ── Audit-log emission for /config_overrides/hashicorp_vault ─────────────────
class TestHashicorpVaultAuditLog:
"""The KMS config endpoint controls every secret retrieval on the proxy.
A mutation (create/update/delete) must emit an audit-log row when
``litellm.store_audit_logs`` is True, with credential values redacted
so the audit table can't itself be a credential-harvest sink."""
@pytest.mark.asyncio
async def test_post_emits_audit_log_with_redacted_values(self, client, monkeypatch):
monkeypatch.setattr(litellm, "store_audit_logs", True)
mock_prisma, mock_db = _make_mock_db()
mock_cfg = _make_mock_proxy_config()
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(ps, "proxy_config", mock_cfg)
_set_admin()
audit_calls = []
async def capture(request_data):
audit_calls.append(request_data)
from unittest.mock import patch
try:
with patch(
"litellm.proxy.management_helpers.audit_logs.create_audit_log_for_update",
new=capture,
):
r = client.post(
VAULT_URL,
json={
"vault_addr": "https://vault.example.com",
"vault_token": "my-very-secret-token",
},
)
assert r.status_code == 200
import asyncio
for _ in range(3):
await asyncio.sleep(0)
assert len(audit_calls) == 1
log = audit_calls[0]
assert log.action == "created"
assert log.object_id == "hashicorp_vault"
# Plaintext credentials must NOT appear anywhere in the row.
assert "my-very-secret-token" not in log.updated_values
assert "vault.example.com" not in log.updated_values
# Field names are kept so the auditor can see what changed.
after = json.loads(log.updated_values)
assert "vault_token" in after["config"]
assert "vault_addr" in after["config"]
finally:
_cleanup()
@pytest.mark.asyncio
async def test_delete_emits_audit_log_only_when_row_existed(
self, client, monkeypatch
):
monkeypatch.setattr(litellm, "store_audit_logs", True)
mock_prisma, mock_db = _make_mock_db()
mock_cfg = _make_mock_proxy_config()
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(ps, "proxy_config", mock_cfg)
_set_admin()
audit_calls = []
async def capture(request_data):
audit_calls.append(request_data)
from unittest.mock import patch
try:
with patch(
"litellm.proxy.management_helpers.audit_logs.create_audit_log_for_update",
new=capture,
):
# Idempotent delete on an empty table → no row, no audit log.
mock_db.find_unique = AsyncMock(return_value=None)
mock_db.delete = AsyncMock(side_effect=RecordNotFoundError(MagicMock()))
r = client.delete(VAULT_URL)
assert r.status_code == 200
import asyncio
for _ in range(3):
await asyncio.sleep(0)
assert audit_calls == []
# Delete a real row → audit log fires with action="deleted".
mock_db.find_unique = AsyncMock(
return_value=_db_record(
{
"vault_addr": "enc_https://v.example.com",
"vault_token": "enc_t",
}
)
)
mock_db.delete = AsyncMock(return_value=None)
r = client.delete(VAULT_URL)
assert r.status_code == 200
for _ in range(3):
await asyncio.sleep(0)
assert len(audit_calls) == 1
log = audit_calls[0]
assert log.action == "deleted"
# The before-snapshot must redact the token before logging.
assert "enc_t" not in log.before_value
assert "v.example.com" not in log.before_value
finally:
_cleanup()
@pytest.mark.asyncio
async def test_no_audit_when_store_audit_logs_is_off(self, client, monkeypatch):
monkeypatch.setattr(litellm, "store_audit_logs", False)
mock_prisma, mock_db = _make_mock_db()
mock_cfg = _make_mock_proxy_config()
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(ps, "proxy_config", mock_cfg)
_set_admin()
audit_calls = []
async def capture(request_data):
audit_calls.append(request_data)
from unittest.mock import patch
try:
with patch(
"litellm.proxy.management_helpers.audit_logs.create_audit_log_for_update",
new=capture,
):
r = client.post(
VAULT_URL,
json={
"vault_addr": "https://vault.example.com",
"vault_token": "tok",
},
)
assert r.status_code == 200
assert audit_calls == []
finally:
_cleanup()