188 lines
7.1 KiB
Python
188 lines
7.1 KiB
Python
import logging
|
|
from typing import Any
|
|
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy.exc import NoInspectionAvailable
|
|
from sqlalchemy.orm import Session
|
|
from starlette.requests import Request
|
|
|
|
from app.models.audit import AuditLog
|
|
from app.models.user import User
|
|
from app.repositories.audit_repository import AuditRepository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SENSITIVE_KEYS = {"password", "password_hash", "smtp_password", "token", "access_token", "secret", "secret_key"}
|
|
|
|
|
|
def to_audit_data(value: Any, seen: set[int] | None = None) -> Any:
|
|
seen = seen or set()
|
|
|
|
if value is None or isinstance(value, str | int | float | bool):
|
|
return value
|
|
|
|
value_id = id(value)
|
|
if value_id in seen:
|
|
return None
|
|
|
|
if isinstance(value, list | tuple | set):
|
|
seen.add(value_id)
|
|
return [to_audit_data(item, seen) for item in value]
|
|
|
|
if isinstance(value, dict):
|
|
seen.add(value_id)
|
|
return {key: to_audit_data(item, seen) for key, item in value.items()}
|
|
|
|
try:
|
|
state = inspect(value)
|
|
except NoInspectionAvailable:
|
|
return jsonable_encoder(value)
|
|
|
|
seen.add(value_id)
|
|
data: dict[str, Any] = {}
|
|
|
|
for column in state.mapper.column_attrs:
|
|
data[column.key] = getattr(value, column.key)
|
|
|
|
for relationship in state.mapper.relationships:
|
|
if relationship.key in state.unloaded:
|
|
continue
|
|
related_value = getattr(value, relationship.key)
|
|
data[relationship.key] = to_audit_data(related_value, seen)
|
|
|
|
return data
|
|
|
|
|
|
def sanitize(value: Any) -> Any:
|
|
encoded = jsonable_encoder(to_audit_data(value))
|
|
if isinstance(encoded, dict):
|
|
return {
|
|
key: "***" if key in SENSITIVE_KEYS else sanitize(item)
|
|
for key, item in encoded.items()
|
|
}
|
|
if isinstance(encoded, list):
|
|
return [sanitize(item) for item in encoded]
|
|
return encoded
|
|
|
|
|
|
def request_ip(request: Request | None) -> str:
|
|
if request is None:
|
|
return ""
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.client.host if request.client else ""
|
|
|
|
|
|
def write_audit_log(
|
|
db: Session,
|
|
*,
|
|
action: str,
|
|
entity_type: str,
|
|
entity_id: int | None = None,
|
|
entity_label: str = "",
|
|
actor: User | None = None,
|
|
actor_user_id: int | None = None,
|
|
actor_username: str = "",
|
|
request: Request | None = None,
|
|
before_data: Any = None,
|
|
after_data: Any = None,
|
|
metadata: Any = None,
|
|
) -> None:
|
|
try:
|
|
AuditRepository.create(
|
|
db,
|
|
actor_user_id=actor.id if actor is not None else actor_user_id,
|
|
actor_username=actor.username if actor is not None else actor_username,
|
|
action=action,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
entity_label=entity_label,
|
|
ip_address=request_ip(request),
|
|
user_agent=request.headers.get("user-agent", "") if request is not None else "",
|
|
before_data=sanitize(before_data),
|
|
after_data=sanitize(after_data),
|
|
metadata=sanitize(metadata),
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
logger.exception("audit.write_failed", extra={"action": action, "entity_type": entity_type})
|
|
|
|
|
|
def action_title(action: str) -> str:
|
|
labels = {
|
|
"auth.login.success": "Login erfolgreich",
|
|
"auth.login.failed": "Login fehlgeschlagen",
|
|
"auth.logout": "Logout",
|
|
"users.create": "Benutzer erstellt",
|
|
"users.update": "Benutzer bearbeitet",
|
|
"users.delete": "Benutzer gelöscht",
|
|
"users.password_update": "Passwort geändert",
|
|
"roles.create": "Rolle erstellt",
|
|
"roles.update": "Rolle bearbeitet",
|
|
"roles.delete": "Rolle gelöscht",
|
|
"roles.permissions_update": "Rollenberechtigungen geändert",
|
|
"customers.create": "Kunde erstellt",
|
|
"customers.update": "Kunde bearbeitet",
|
|
"customers.delete": "Kunde gelöscht",
|
|
"customers.import": "Kunden importiert",
|
|
"customer_contacts.create": "Ansprechpartner erstellt",
|
|
"customer_contacts.update": "Ansprechpartner bearbeitet",
|
|
"customer_contacts.delete": "Ansprechpartner gelöscht",
|
|
"knowledge.manufacturers.create": "Hersteller erstellt",
|
|
"knowledge.manufacturers.update": "Hersteller bearbeitet",
|
|
"knowledge.manufacturers.delete": "Hersteller gelöscht",
|
|
"knowledge.devices.create": "Gerät erstellt",
|
|
"knowledge.devices.update": "Gerät bearbeitet",
|
|
"knowledge.devices.delete": "Gerät gelöscht",
|
|
"knowledge.documents.create": "Dokument erstellt",
|
|
"knowledge.documents.upload": "Dokument hochgeladen",
|
|
"knowledge.documents.update": "Dokument bearbeitet",
|
|
"knowledge.documents.delete": "Dokument gelöscht",
|
|
"knowledge.notes.create": "Notiz erstellt",
|
|
"knowledge.notes.update": "Notiz bearbeitet",
|
|
"knowledge.notes.delete": "Notiz gelöscht",
|
|
"users.initial_admin_bootstrap": "Initialer Administrator erstellt",
|
|
"knowledge.manufacturers.create": "Hersteller erstellt",
|
|
"knowledge.manufacturers.update": "Hersteller bearbeitet",
|
|
"knowledge.manufacturers.delete": "Hersteller gelöscht",
|
|
"knowledge.devices.create": "Gerät erstellt",
|
|
"knowledge.devices.update": "Gerät bearbeitet",
|
|
"knowledge.devices.delete": "Gerät gelöscht",
|
|
"knowledge.documents.create": "Dokument erstellt",
|
|
"knowledge.documents.upload": "Dokument hochgeladen",
|
|
"knowledge.documents.update": "Dokument bearbeitet",
|
|
"knowledge.documents.delete": "Dokument gelöscht",
|
|
"knowledge.notes.create": "Wissensnotiz erstellt",
|
|
"knowledge.notes.update": "Wissensnotiz bearbeitet",
|
|
"knowledge.notes.delete": "Wissensnotiz gelöscht",
|
|
"repairs.create": "Reparatur erstellt",
|
|
"repairs.update": "Reparatur bearbeitet",
|
|
"repairs.status_update": "Reparaturstatus geändert",
|
|
"repairs.cancel": "Reparatur storniert",
|
|
"repairs.intake_processed": "Reparatur-Intake verarbeitet",
|
|
"repairs.intake_failed": "Reparatur-Intake fehlgeschlagen",
|
|
"repairs.public_link.create": "Reparatur-Statuslink erstellt",
|
|
"repairs.public_link.revoke": "Reparatur-Statuslink deaktiviert",
|
|
"repairs.public_link.auto_create": "Reparatur-Statuslink automatisch erstellt",
|
|
"repairs.public_link.regenerate": "Reparatur-Statuslink erneut erstellt",
|
|
"repairs.status_mail.sent": "Statusmail versendet",
|
|
"repairs.status_mail.failed": "Statusmail fehlgeschlagen",
|
|
}
|
|
return labels.get(action, action)
|
|
|
|
|
|
def audit_to_activity(audit_log: AuditLog) -> dict[str, Any]:
|
|
title = action_title(audit_log.action)
|
|
actor = audit_log.actor_username or "System"
|
|
label = audit_log.entity_label or audit_log.entity_type
|
|
return {
|
|
"id": audit_log.id,
|
|
"title": title,
|
|
"description": f"{actor} · {label}",
|
|
"actor": actor,
|
|
"entity_type": audit_log.entity_type,
|
|
"entity_id": audit_log.entity_id,
|
|
"created_at": audit_log.created_at,
|
|
}
|