feat(proxy): auth_v2 decision audit trail

Every authorization decision now flows through one audit record - the compliance
trail enterprise auth needs and the biggest gap called out in the design review.

- audit.py: a frozen AuthzDecision (decision, subject, domain, obj, action, route,
  reason, auth_method) and record(), which logs the decision and fans out to any
  registered sinks (DB / SIEM). Sinks are isolated: a failing sink is logged but
  never affects the request outcome.
- The authorizer records allow, deny, and loud-open for every governed and
  ungoverned control-plane route; the entry point records the model-call (call)
  decision on the inference path. auth_method is threaded through so the record
  shows how the principal authenticated.
- register_sink / reset_sinks exported for wiring a durable sink.

Fully typed (no new Any), mypy clean on 22 files, 134 tests green, still imports
with casbin/authlib absent.

litellm/proxy/auth/v2/__init__.py

@@ -1,3 +1,4 @@
+from .audit import AuthzDecision, Decision, register_sink, reset_sinks
 from .context import (
     AuthMethod,
     RequestAuthContext,
@@ -20,4 +21,8 @@ __all__ = [
     "attach_end_user",
     "resolve_end_user",
     "enrich_identity",
+    "AuthzDecision",
+    "Decision",
+    "register_sink",
+    "reset_sinks",
 ]

litellm/proxy/auth/v2/audit.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from enum import Enum
+from typing import Callable, List, Optional
+
+logger = logging.getLogger("litellm.proxy.auth.v2.audit")
+
+
+class Decision(str, Enum):
+    ALLOW = "allow"
+    DENY = "deny"
+    # A route auth_v2 does not yet govern, allowed (loudly) during build-out.
+    LOUD_OPEN = "loud_open"
+
+
+@dataclass(frozen=True)
+class AuthzDecision:
+    """One authorization decision, the unit of the compliance audit trail."""
+
+    decision: Decision
+    subject: str
+    domain: str
+    obj: str
+    action: str
+    route: str
+    reason: str
+    auth_method: Optional[str] = None
+
+
+AuditSink = Callable[[AuthzDecision], None]
+
+_sinks: List[AuditSink] = []
+
+
+def register_sink(sink: AuditSink) -> None:
+    """Register a sink (DB / SIEM writer) that receives every authz decision."""
+    _sinks.append(sink)
+
+
+def reset_sinks() -> None:
+    _sinks.clear()
+
+
+def record(decision: AuthzDecision) -> None:
+    """Emit an authorization decision to the audit log and registered sinks.
+
+    Every governed allow/deny and every loud-open passes through here, so the
+    trail is complete and centralized. Sinks are isolated: a failing sink is
+    logged but never affects the request's outcome.
+    """
+    logger.info(
+        "auth_v2 authz %s: subject=%s action=%s obj=%s route=%s reason=%s method=%s",
+        decision.decision.value,
+        decision.subject,
+        decision.action,
+        decision.obj,
+        decision.route,
+        decision.reason,
+        decision.auth_method or "-",
+    )
+    for sink in _sinks:
+        try:
+            sink(decision)
+        except Exception:
+            logger.exception("auth_v2 audit sink raised; decision still enforced")

litellm/proxy/auth/v2/authz/authorizer.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Any, Dict, Optional
 
+from ..audit import AuthzDecision, Decision, record
 from ..principal import Principal
 from ..protocols import SupportsEnforce
 from .route_map import GovernedRoute, match_route
@@ -35,10 +36,12 @@ def authorize(
     request_data: Optional[Dict[str, Any]],
     enforcer: SupportsEnforce,
     method: Optional[str] = None,
+    auth_method: Optional[str] = None,
 ) -> None:
     """Enforce policy for ``route``. No-op (loud) for routes v2 doesn't yet govern.
 
-    Raises :class:`AuthorizationDenied` when a governed route is denied.
+    Raises :class:`AuthorizationDenied` when a governed route is denied. Every
+    outcome (allow, deny, loud-open) is recorded to the audit trail.
     ``enforcer`` is anything exposing ``enforce(subject, domain, obj, action)``.
     ``method`` is required to resolve REST resources (e.g. credentials) whose verb
     is the HTTP method; without it those routes would be treated as loud-open.
@@ -50,10 +53,35 @@ def authorize(
             "This must not reach production with auth_v2 enabled.",
             route,
         )
+        record(
+            AuthzDecision(
+                decision=Decision.LOUD_OPEN,
+                subject=principal.subject,
+                domain=principal.domain,
+                obj=f"route:{route}",
+                action="*",
+                route=route,
+                reason="route not yet governed by auth_v2",
+                auth_method=auth_method,
+            )
+        )
         return
 
     obj = _build_object(rule, request_data)
-    if not enforcer.enforce(principal.subject, principal.domain, obj, rule.action):
+    allowed = enforcer.enforce(principal.subject, principal.domain, obj, rule.action)
+    record(
+        AuthzDecision(
+            decision=Decision.ALLOW if allowed else Decision.DENY,
+            subject=principal.subject,
+            domain=principal.domain,
+            obj=obj,
+            action=rule.action,
+            route=route,
+            reason="control-plane policy",
+            auth_method=auth_method,
+        )
+    )
+    if not allowed:
         raise AuthorizationDenied(
             subject=principal.subject, obj=obj, action=rule.action
         )

litellm/proxy/auth/v2/entry.py

@@ -6,6 +6,7 @@ from fastapi import HTTPException, Request, status
 
 from litellm.integrations.otel.runtime import seed_request_identity
 
+from .audit import AuthzDecision, Decision, record
 from .authn.authenticators import AuthContext, AuthResult, authenticate
 from .authz.authorizer import AuthorizationDenied, authorize
 from .authz.enforcer import CasbinEnforcer
@@ -167,7 +168,14 @@ async def user_api_key_auth_v2(
         enforcer = await _build_enforcer(principal, prisma_client)
 
         try:
-            authorize(principal, route, request_data, enforcer, request.method)
+            authorize(
+                principal,
+                route,
+                request_data,
+                enforcer,
+                request.method,
+                auth_method=result.method.value,
+            )
         except AuthorizationDenied as e:
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
 
@@ -189,9 +197,21 @@ async def user_api_key_auth_v2(
         )
         if requested_model:
             enforcer = await _build_enforcer(principal, prisma_client)
-            if not enforcer.enforce(
-                principal.subject, principal.domain, f"model:{requested_model}", "call"
-            ):
+            obj = f"model:{requested_model}"
+            allowed = enforcer.enforce(principal.subject, principal.domain, obj, "call")
+            record(
+                AuthzDecision(
+                    decision=Decision.ALLOW if allowed else Decision.DENY,
+                    subject=principal.subject,
+                    domain=principal.domain,
+                    obj=obj,
+                    action="call",
+                    route=route,
+                    reason="model call",
+                    auth_method=result.method.value,
+                )
+            )
+            if not allowed:
                 raise HTTPException(
                     status_code=status.HTTP_403_FORBIDDEN,
                     detail=f"auth_v2: not permitted to call model '{requested_model}'",
@@ -205,7 +225,14 @@ async def user_api_key_auth_v2(
     # Loud-open: route v2 doesn't yet govern. No identity required.
     result = await _best_effort_identity(token, ctx)
     principal, identity = _establish_context(request, result, route)
-    authorize(principal, route, None, _DENY_ALL, request.method)
+    authorize(
+        principal,
+        route,
+        None,
+        _DENY_ALL,
+        request.method,
+        auth_method=result.method.value,
+    )
     seed_request_identity(identity)
     identity.request_route = route
     return identity

tests/test_litellm/proxy/auth/v2/test_audit.py

@@ -0,0 +1,104 @@
+from typing import List
+
+import pytest
+
+from litellm.proxy.auth.v2.audit import (
+    AuthzDecision,
+    Decision,
+    record,
+    register_sink,
+    reset_sinks,
+)
+from litellm.proxy.auth.v2.authz.authorizer import AuthorizationDenied, authorize
+from litellm.proxy.auth.v2.principal import Principal
+
+PRINCIPAL = Principal(subject="user:u1", domain="*", groupings=[])
+
+
+@pytest.fixture(autouse=True)
+def _clear_sinks():
+    reset_sinks()
+    yield
+    reset_sinks()
+
+
+def _capture() -> List[AuthzDecision]:
+    captured: List[AuthzDecision] = []
+    register_sink(captured.append)
+    return captured
+
+
+class _Enforcer:
+    def __init__(self, ret: bool):
+        self.ret = ret
+
+    def enforce(self, *_args: object) -> bool:
+        return self.ret
+
+
+class _Exploding:
+    def enforce(self, *_args: object) -> bool:
+        raise AssertionError("enforce must not be called on a loud-open route")
+
+
+def test_record_delivers_to_registered_sink():
+    got = _capture()
+    decision = AuthzDecision(
+        decision=Decision.ALLOW,
+        subject="user:u1",
+        domain="*",
+        obj="model:x",
+        action="read",
+        route="/model/info",
+        reason="t",
+    )
+    record(decision)
+    assert got == [decision]
+
+
+def test_failing_sink_is_isolated():
+    good: List[AuthzDecision] = []
+
+    def boom(_decision: AuthzDecision) -> None:
+        raise RuntimeError("sink down")
+
+    register_sink(boom)
+    register_sink(good.append)
+    # The failing sink must not stop the good one or raise out of record().
+    record(
+        AuthzDecision(
+            decision=Decision.DENY,
+            subject="s",
+            domain="*",
+            obj="o",
+            action="a",
+            route="/r",
+            reason="t",
+        )
+    )
+    assert len(good) == 1
+
+
+def test_authorize_records_allow_with_auth_method():
+    got = _capture()
+    authorize(PRINCIPAL, "/model/new", {}, _Enforcer(True), auth_method="virtual_key")
+    assert len(got) == 1
+    assert got[0].decision is Decision.ALLOW
+    assert got[0].obj == "model:*"
+    assert got[0].action == "write"
+    assert got[0].auth_method == "virtual_key"
+
+
+def test_authorize_records_deny_and_still_raises():
+    got = _capture()
+    with pytest.raises(AuthorizationDenied):
+        authorize(PRINCIPAL, "/model/delete", {"model_id": "m9"}, _Enforcer(False))
+    assert got[0].decision is Decision.DENY
+    assert got[0].obj == "model:m9"
+
+
+def test_authorize_records_loud_open_for_ungoverned_route():
+    got = _capture()
+    authorize(PRINCIPAL, "/chat/completions", {}, _Exploding())
+    assert got[0].decision is Decision.LOUD_OPEN
+    assert got[0].route == "/chat/completions"