Merge pull request #26953 from stuxf/chore/audit-log-cache-vault-settings

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

litellm/proxy/_types.py

@@ -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):

litellm/proxy/management_endpoints/cache_settings_endpoints.py

@@ -8,14 +8,23 @@ 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 (
+    AUDIT_ACTIONS,
+    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 +35,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: AUDIT_ACTIONS,
+    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 +370,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 +406,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: AUDIT_ACTIONS = "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 +458,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",

litellm/proxy/management_endpoints/config_override_endpoints.py

@@ -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
 
@@ -18,8 +21,11 @@ from litellm._logging import verbose_proxy_logger
 from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.proxy._types import (
+    AUDIT_ACTIONS,
     CommonProxyErrors,
     KeyManagementSystem,
+    LiteLLM_AuditLogs,
+    LitellmTableNames,
     LitellmUserRoles,
     UserAPIKeyAuth,
 )
@@ -32,6 +38,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: AUDIT_ACTIONS,
+    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 +226,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.
@@ -171,6 +257,8 @@ async def update_hashicorp_vault_config(
     existing_record = await prisma_client.db.litellm_configoverrides.find_unique(
         where={"config_type": "hashicorp_vault"}
     )
+    existing_decrypted: Optional[Dict[str, Any]] = None
+    env_values: Dict[str, Any] = {}
     if existing_record is not None and existing_record.config_value is not None:
         existing_data = _parse_config_value(existing_record.config_value)
         existing_decrypted = proxy_config._decrypt_db_variables(existing_data)
@@ -178,7 +266,8 @@ async def update_hashicorp_vault_config(
             if field not in config_data and existing_decrypted.get(field):
                 config_data[field] = existing_decrypted[field]
     else:
-        # No DB record yet — merge from current env vars
+        # No DB record (or DB record with null config_value) — merge from
+        # current env vars instead.
         env_values = _get_current_env_values(HASHICORP_ENV_VAR_MAPPING)
         for field in HASHICORP_ENV_VAR_MAPPING:
             if field not in config_data and env_values.get(field):
@@ -248,6 +337,22 @@ 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.
+    # Action keys off row existence (a row with NULL ``config_value`` is
+    # still an update). ``before_config`` falls back to env vars when the
+    # row was absent or its ``config_value`` was NULL.
+    before_config = existing_decrypted if existing_decrypted is not None else env_values
+    action: AUDIT_ACTIONS = "updated" if existing_record is not None else "created"
+    await _emit_hashicorp_vault_audit_log(
+        action=action,
+        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",
@@ -321,6 +426,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
@@ -337,11 +446,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"
@@ -349,6 +474,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",

tests/test_litellm/proxy/management_endpoints/test_cache_settings_endpoints.py

@@ -2,6 +2,8 @@
 Unit tests for cache settings management endpoints
 """
 
+import asyncio
+import json
 import os
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -12,12 +14,15 @@ sys.path.insert(
     0, os.path.abspath("../../../..")
 )  # Adds the parent directory to the system path
 
-from litellm.proxy._types import LitellmUserRoles
+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,140 @@ 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 == []

tests/test_litellm/proxy/management_endpoints/test_config_override_endpoints.py

@@ -1,6 +1,7 @@
+import asyncio
 import json
 import os
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
@@ -272,3 +273,189 @@ 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)
+
+        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
+                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_post_action_is_updated_when_row_exists_with_null_config_value(
+        self, client, monkeypatch
+    ):
+        """A row can exist in ``litellm_configoverrides`` with a NULL
+        ``config_value`` (e.g. an earlier failed write left a stub).
+        Re-POSTing must label the audit row as ``updated`` — the row
+        already exists — not ``created``."""
+        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()
+
+        # Row exists but ``config_value`` is NULL.
+        null_record = MagicMock()
+        null_record.config_value = None
+        mock_db.find_unique = AsyncMock(return_value=null_record)
+
+        audit_calls = []
+
+        async def capture(request_data):
+            audit_calls.append(request_data)
+
+        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
+                for _ in range(3):
+                    await asyncio.sleep(0)
+
+            assert len(audit_calls) == 1
+            assert audit_calls[0].action == "updated"
+        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)
+
+        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
+                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)
+
+        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()