33d7189418
- app/audit.py: eigener Logger "va.audit" (eigener stdout-Handler -> Journal/Log-Tab). log_admin_action() protokolliert Aktion + Auslöser (SSO-Name oder admin-key). - Verdrahtet in: config_set/config_reset (PUT/DELETE /admin/config), llm_backend_switch (+ _rejected), gateway_restart. - Tests: caplog prüft Audit-Zeilen für config_set und abgelehnten Backend-Switch. - Doku §7.5: Audit-Format im Log-Tab. Schließt den Plan (Admin-gesteuerte LLM-/Gateway-Verwaltung, Schritte 1–5) ab. 167 grün. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
44 lines
1.8 KiB
Python
44 lines
1.8 KiB
Python
"""Strukturiertes Audit-Logging für Admin-Aktionen.
|
|
|
|
Schreibt eine Zeile pro schreibender Admin-Aktion (Backend-Wechsel, Config-Änderung,
|
|
Gateway-Neustart …) mit der Identität des Auslösers. Die Zeilen landen über stdout im
|
|
systemd-Journal und sind damit live im Admin-Log-Tab sichtbar.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import sys
|
|
|
|
# Eigener Logger mit eigenem Handler -> unabhängig von der uvicorn-Logging-Config,
|
|
# erscheint zuverlässig auf stdout (= Journal im Dienst-Betrieb).
|
|
logger = logging.getLogger("va.audit")
|
|
if not logger.handlers:
|
|
_h = logging.StreamHandler(sys.stdout)
|
|
_h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
|
|
logger.addHandler(_h)
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
|
|
|
|
def admin_identity(request) -> str:
|
|
"""Ermittelt, wer die Admin-Aktion ausführt (SSO-Username oder 'admin-key')."""
|
|
try:
|
|
from app.auth import authenticate, _bearer_token
|
|
client_host = request.client.host if request.client else ""
|
|
token = _bearer_token(request.headers.get("authorization"))
|
|
user = authenticate(request.headers, client_host, token)
|
|
if user is not None and getattr(user, "external_id", None):
|
|
return user.external_id
|
|
except Exception: # pragma: no cover - Auth-Fehler nie fatal fürs Logging
|
|
pass
|
|
if request.headers.get("x-admin-key"):
|
|
return "admin-key"
|
|
return "unknown"
|
|
|
|
|
|
def log_admin_action(request, action: str, **fields) -> None:
|
|
"""Loggt eine Admin-Aktion strukturiert: action, user + freie Felder."""
|
|
who = admin_identity(request)
|
|
extra = " ".join(f"{k}={v!r}" for k, v in fields.items() if v is not None)
|
|
logger.info("ADMIN action=%s user=%s %s", action, who, extra)
|