my_voice_assistant_v3/app/audit.py

44 lines
1.8 KiB
Python
Raw Permalink Normal View History

"""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)