[Feature] Add Audit Log Export to External Callbacks#23167
[Feature] Add Audit Log Export to External Callbacks#23167ryan-crabbe merged 5 commits intolitellm_ryan_march_20from
Conversation
Audit logs (CRUD events on keys, teams, users, models) were only stored in
the Prisma DB. This adds a pluggable callback system so audit logs can be
forwarded to external services like S3 for ingestion into security monitoring
tools.
New config key `audit_log_callbacks` under `litellm_settings` reuses the
existing callback infrastructure. Any CustomLogger subclass can opt in by
overriding `async_log_audit_log_event()`. S3Logger (s3_v2) is implemented
as the first handler, storing audit logs under `audit_logs/{date}/` prefix.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Greptile SummaryThis PR adds audit log export to external callbacks (initially S3 via Key changes:
Confidence Score: 3/5
|
| Filename | Overview |
|---|---|
| litellm/proxy/management_helpers/audit_logs.py | Core dispatch logic is well-structured: JSON serialisation happens before dispatch, fire-and-forget tasks have done-callbacks, and errors are non-blocking. No new issues beyond what has been flagged in earlier threads. |
| litellm/integrations/s3_v2.py | New async_log_audit_log_event method uses the callback fire time for the S3 key, not the event's own updated_at timestamp; near-midnight events can be filed under the wrong date. Also contains a redundant in-function from datetime import timezone (flagged in a prior thread). |
| litellm/proxy/proxy_server.py | audit_log_callbacks loading iterates value and calls "." in callback without first verifying each entry is a str, which will raise TypeError on non-string YAML entries. The store_audit_logs warning also doesn't account for the LITELLM_STORE_AUDIT_LOGS env-var path (flagged in prior thread). |
| litellm/types/utils.py | New StandardAuditLogPayload TypedDict is clean, but action is typed as str rather than the existing AUDIT_ACTIONS Literal, weakening the public callback API's type contract. |
| litellm/integrations/custom_logger.py | Adds the async_log_audit_log_event no-op hook to CustomLogger. Clean, minimal change. |
| litellm/init.py | Adds the audit_log_callbacks list alongside existing callback lists. Consistent with the existing pattern. |
| tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py | Good coverage of dispatch, premium gating, and S3 key format. Tests rely on asyncio.sleep(0.1) for task completion which is fragile in slow CI. The fixture also doesn't clear _audit_log_callback_cache (flagged in prior thread). |
Sequence Diagram
sequenceDiagram
participant Caller as API Handler
participant AL as create_audit_log_for_update
participant D as _dispatch_audit_log_to_callbacks
participant R as _resolve_audit_log_callback
participant S3 as S3Logger
participant DB as Prisma DB
Caller->>AL: create_audit_log_for_update(request_data)
AL->>AL: Check store_audit_logs and premium_user gates
AL->>AL: Serialize dict fields to JSON strings
AL->>D: await _dispatch_audit_log_to_callbacks(request_data)
D->>D: _build_audit_log_payload -> StandardAuditLogPayload
loop for each callback in audit_log_callbacks
D->>R: _resolve_audit_log_callback(name) if string
R-->>D: cached CustomLogger instance
D->>D: asyncio.create_task(callback.async_log_audit_log_event(payload))
end
D-->>AL: return
alt prisma_client available
AL->>DB: litellm_auditlog.create(data)
DB-->>AL: success
else prisma_client is None
AL->>AL: log error and return
end
Note over S3: Background task executes
S3->>S3: Build S3 key under audit_logs/date/ prefix
S3->>S3: Append element to log_queue
alt queue size >= batch_size
S3->>S3: flush_queue() and upload to S3
end
Last reviewed commit: "fix black formatting"
Two fixes based on PR feedback: 1. Move callback dispatch before the prisma_client check so audit logs still reach S3/Datadog even if the DB is down. Also changed the prisma_client=None case from raising an exception to logging an error and returning gracefully. 2. Attach a done_callback to asyncio tasks created for audit log callbacks so exceptions are logged through verbose_proxy_logger instead of silently swallowed. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
@greptile |
| original = litellm.audit_log_callbacks | ||
| litellm.audit_log_callbacks = [] | ||
| yield | ||
| litellm.audit_log_callbacks = original | ||
|
|
||
|
|
||
| def _make_audit_log( |
There was a problem hiding this comment.
_audit_log_callback_cache not cleared between tests
The reset_audit_log_callbacks fixture resets litellm.audit_log_callbacks but does not clear _audit_log_callback_cache (the module-level dict in audit_logs.py). If any test calls _resolve_audit_log_callback with a string name (e.g. "s3_v2"), the resolved instance will be cached and leaked into subsequent tests, potentially causing them to use a stale or unexpected logger instance.
Add cache cleanup to the fixture:
| original = litellm.audit_log_callbacks | |
| litellm.audit_log_callbacks = [] | |
| yield | |
| litellm.audit_log_callbacks = original | |
| def _make_audit_log( | |
| @pytest.fixture(autouse=True) | |
| def reset_audit_log_callbacks(): | |
| """Reset audit_log_callbacks before and after each test.""" | |
| from litellm.proxy.management_helpers.audit_logs import _audit_log_callback_cache | |
| original = litellm.audit_log_callbacks | |
| litellm.audit_log_callbacks = [] | |
| _audit_log_callback_cache.clear() | |
| yield | |
| litellm.audit_log_callbacks = original | |
| _audit_log_callback_cache.clear() |
| from datetime import timezone | ||
|
|
There was a problem hiding this comment.
Redundant in-function import of
timezone
datetime is already imported at the module level (from datetime import datetime). The from datetime import timezone inside the function body is redundant — timezone can be accessed directly since it lives in the same datetime module. Move this to the module-level import:
| from datetime import timezone | |
| from datetime import datetime, timezone |
and remove the in-function import entirely.
| else: | ||
| verbose_proxy_logger.warning( | ||
| "'audit_log_callbacks' is configured but 'store_audit_logs' is not enabled. " | ||
| "Audit log callbacks will not fire until 'store_audit_logs: true' is added to litellm_settings." | ||
| ) |
There was a problem hiding this comment.
Warning does not account for
LITELLM_STORE_AUDIT_LOGS env var
The warning check reads litellm_settings.get("store_audit_logs", litellm.store_audit_logs) — but the runtime check in create_audit_log_for_update also consults get_secret_bool("LITELLM_STORE_AUDIT_LOGS"). A user who sets LITELLM_STORE_AUDIT_LOGS=true via environment variable (without writing store_audit_logs: true into the config YAML) will see the "callbacks will not fire" warning, even though callbacks will actually fire at runtime. This can cause unnecessary alarm.
Additionally, the warning doesn't surface the other gate: audit log callbacks are also blocked for non-premium users (premium_user check in create_audit_log_for_update). If a non-premium user configures audit_log_callbacks, they receive a positive "Initialized Audit Log Callbacks" log (assuming store_audit_logs is set), but callbacks will silently never fire.
Consider updating the warning to cover both conditions:
_store_audit_logs = litellm_settings.get(
"store_audit_logs", litellm.store_audit_logs
) or get_secret_bool("LITELLM_STORE_AUDIT_LOGS")
if not _store_audit_logs:
verbose_proxy_logger.warning(
"'audit_log_callbacks' is configured but 'store_audit_logs' is not enabled "
"(set 'store_audit_logs: true' in litellm_settings or LITELLM_STORE_AUDIT_LOGS=true). "
"Additionally, a premium license is required for audit log callbacks to fire."
)| updated_at: str # ISO-8601 | ||
| changed_by: str | ||
| changed_by_api_key: str | ||
| action: str # "created" | "updated" | "deleted" | "blocked" | "rotated" |
There was a problem hiding this comment.
action comment is missing "unblocked"
AUDIT_ACTIONS is defined as Literal["created", "updated", "deleted", "blocked", "unblocked", "rotated"], but the comment here only lists "created" | "updated" | "deleted" | "blocked" | "rotated" — "unblocked" is missing.
| action: str # "created" | "updated" | "deleted" | "blocked" | "rotated" | |
| action: str # "created" | "updated" | "deleted" | "blocked" | "unblocked" | "rotated" |
| for callback in litellm.audit_log_callbacks: | ||
| try: | ||
| resolved = callback | ||
| if isinstance(callback, str): | ||
| resolved = _resolve_audit_log_callback(callback) | ||
| if resolved is None: | ||
| verbose_proxy_logger.warning( | ||
| "Could not resolve audit log callback: %s", callback | ||
| ) | ||
| continue | ||
|
|
||
| if isinstance(resolved, CustomLogger): | ||
| task = asyncio.create_task( | ||
| resolved.async_log_audit_log_event(payload) | ||
| ) | ||
| task.add_done_callback(_audit_log_task_done_callback) | ||
| except Exception as e: | ||
| verbose_proxy_logger.error( | ||
| "Failed dispatching audit log to callback: %s", e | ||
| ) |
There was a problem hiding this comment.
Non-
CustomLogger callback instances are silently dropped
If litellm.audit_log_callbacks contains an entry that is neither a str nor a CustomLogger (e.g., a custom callable or a misconfigured object from get_instance_fn), the loop falls through both isinstance branches without logging a warning. The user's callback is silently ignored with no diagnostics.
Consider adding an else branch to warn when a callback is not usable:
if isinstance(resolved, CustomLogger):
task = asyncio.create_task(
resolved.async_log_audit_log_event(payload)
)
task.add_done_callback(_audit_log_task_done_callback)
else:
verbose_proxy_logger.warning(
"Audit log callback %r is not a CustomLogger instance and will be skipped.",
resolved,
)Made-with: Cursor
| updated_at: str # ISO-8601 | ||
| changed_by: str | ||
| changed_by_api_key: str | ||
| action: str # "created" | "updated" | "deleted" | "blocked" | "rotated" |
There was a problem hiding this comment.
action typed as str instead of AUDIT_ACTIONS Literal
action is typed as str, but the existing AUDIT_ACTIONS Literal ("created" | "updated" | "deleted" | "blocked" | "unblocked" | "rotated") provides a precise contract. Using str means:
- Callback implementors get no IDE/type-checker guidance on valid values
- Invalid action strings can be dispatched to callbacks without any type error
AUDIT_ACTIONS can't be imported directly from litellm.proxy._types here without a potential circular import, but it can be redefined inline or imported lazily. A clean option is to use TYPE_CHECKING:
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from litellm.proxy._types import AUDIT_ACTIONSand then type the field accordingly, or simply copy the Literal definition into types/utils.py:
| action: str # "created" | "updated" | "deleted" | "blocked" | "rotated" | |
| action: Literal["created", "updated", "deleted", "blocked", "unblocked", "rotated"] # AUDIT_ACTIONS |
| for callback in value: | ||
| if "." in callback: | ||
| litellm.audit_log_callbacks.append( | ||
| get_instance_fn(value=callback) | ||
| ) | ||
| else: | ||
| litellm.audit_log_callbacks.append(callback) |
There was a problem hiding this comment.
No type guard before
"." in callback
Every item in value is assumed to be a str before performing the "." in callback membership test. If the YAML config includes a non-string entry (e.g., an integer), Python will raise TypeError: argument of type 'int' is not iterable, which would crash config loading with a cryptic error.
The existing success_callback / failure_callback branches that also call get_instance_fn have the same assumption, but all other callback lists validate that the entry is a string before doing this check. Adding a guard keeps the error explicit:
| for callback in value: | |
| if "." in callback: | |
| litellm.audit_log_callbacks.append( | |
| get_instance_fn(value=callback) | |
| ) | |
| else: | |
| litellm.audit_log_callbacks.append(callback) | |
| for callback in value: | |
| if not isinstance(callback, str): | |
| verbose_proxy_logger.warning( | |
| "audit_log_callbacks entry %r is not a string and will be skipped.", | |
| callback, | |
| ) | |
| continue | |
| if "." in callback: | |
| litellm.audit_log_callbacks.append( | |
| get_instance_fn(value=callback) | |
| ) | |
| else: | |
| litellm.audit_log_callbacks.append(callback) |
| s3_object_key = ( | ||
| f"{s3_path}audit_logs/" | ||
| f"{now.strftime('%Y-%m-%d')}/" | ||
| f"{now.strftime('%H-%M-%S')}_{audit_log_id}.json" | ||
| ) |
There was a problem hiding this comment.
S3 key uses dispatch time, not event time
datetime.now(timezone.utc) records when the callback fires, not when the audit event occurred. If callback dispatch is delayed (e.g., due to queue pressure or task scheduling), the directory path audit_logs/{date}/{HH-MM-SS}_... will reflect the wrong time. Audit events created near midnight could land in the next day's directory.
The actual event timestamp is already available in audit_log["updated_at"] as an ISO-8601 string. Parsing and using that field for the key makes the S3 layout consistent with the event timeline:
from datetime import datetime, timezone
event_time_str = audit_log.get("updated_at", "")
try:
event_time = datetime.fromisoformat(event_time_str)
except (ValueError, TypeError):
event_time = datetime.now(timezone.utc)
s3_object_key = (
f"{s3_path}audit_logs/"
f"{event_time.strftime('%Y-%m-%d')}/"
f"{event_time.strftime('%H-%M-%S')}_{audit_log_id}.json"
)|
|
||
| with patch( | ||
| "litellm.proxy.management_helpers.audit_logs._resolve_audit_log_callback", | ||
| return_value=mock_logger, | ||
| ): |
There was a problem hiding this comment.
asyncio.sleep(0.1) is timing-dependent and fragile
await asyncio.sleep(0.1) is used in multiple tests to wait for asyncio.create_task to complete. On a busy CI runner this 100 ms window may not be sufficient, leading to intermittent failures where the assertion fires before the task has run.
A more robust pattern is to explicitly yield control until all pending tasks complete:
# Replace asyncio.sleep(0.1) with:
await asyncio.gather(*asyncio.all_tasks() - {asyncio.current_task()})
# OR after pytest-asyncio >= 0.21:
# configure asyncio_mode = "auto" and let the event loop drain naturallyThis pattern appears on lines 122, 136, 156, 175, 210, 222, 232, 244, and 257 — update all of them for consistent, deterministic test behaviour.
- Move MCP SDK 1.26.0 upgrade from Bugs → Features in MCP Gateway - Remove duplicate PR BerriAI#23167 (audit log export) from AI Integrations; canonical entry remains in Management Endpoints / UI
Relevant issues
Summary
Problem
Audit logs (CRUD events on keys, teams, users, models) are only stored in the Prisma DB (
LiteLLM_AuditLogtable). There is no way to export them to external services like S3 for ingestion into security monitoring tools (Splunk, Datadog, etc.). Spend logs already support this via the callback system, but audit logs do not.The v0 for this is intended to be used with a database setup. We store the audit logs in the db, and then we move to export those audit logs somewhere else via a callback (tested with S3 first).
Fix
Adds a new
audit_log_callbacksconfig key underlitellm_settingsthat reuses the existing callback infrastructure. When an audit log is created, it is dispatched to all registered callbacks via a newasync_log_audit_log_event()method onCustomLogger. S3Logger (s3_v2) is the first handler implemented, storing audit logs under anaudit_logs/{date}/prefix separate from spend logs.Config example:
Testing
Type
🆕 New Feature
✅ Test