feat(admin): Audit-Logging schreibender Admin-Aktionen — Plan-Schritt 5
- 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>
This commit is contained in:
parent
b2576fd465
commit
33d7189418
4 changed files with 119 additions and 7 deletions
44
app/audit.py
Normal file
44
app/audit.py
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
"""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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue