54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
|
"""Audit-Helper: strukturiertes Logging aller Aktionen."""
|
||
|
|
import logging
|
||
|
|
import json
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Optional, Dict, Any
|
||
|
|
import aiosqlite
|
||
|
|
|
||
|
|
from config import settings
|
||
|
|
from db import log_audit, get_db
|
||
|
|
|
||
|
|
|
||
|
|
logger = logging.getLogger("audit")
|
||
|
|
|
||
|
|
|
||
|
|
async def record(
|
||
|
|
db: aiosqlite.Connection,
|
||
|
|
agent_id: Optional[str],
|
||
|
|
user_id: str,
|
||
|
|
action: str,
|
||
|
|
target: Optional[str] = None,
|
||
|
|
args: Optional[Dict[str, Any]] = None,
|
||
|
|
result: Optional[Dict[str, Any]] = None,
|
||
|
|
duration_ms: Optional[int] = None,
|
||
|
|
) -> None:
|
||
|
|
"""Schreibt einen Audit-Eintrag in DB + Logger."""
|
||
|
|
await log_audit(
|
||
|
|
db=db,
|
||
|
|
agent_id=agent_id,
|
||
|
|
user_id=user_id,
|
||
|
|
action=action,
|
||
|
|
target=target,
|
||
|
|
args=args,
|
||
|
|
result=result,
|
||
|
|
duration_ms=duration_ms,
|
||
|
|
)
|
||
|
|
|
||
|
|
log_entry = {
|
||
|
|
"ts": datetime.utcnow().isoformat(),
|
||
|
|
"agent_id": agent_id,
|
||
|
|
"user_id": user_id,
|
||
|
|
"action": action,
|
||
|
|
"target": target,
|
||
|
|
"args": args,
|
||
|
|
"result": result,
|
||
|
|
"duration_ms": duration_ms,
|
||
|
|
}
|
||
|
|
logger.info(json.dumps(log_entry, default=str))
|
||
|
|
|
||
|
|
|
||
|
|
async def get_recent(db: aiosqlite.Connection, limit: int = 100, agent_id: Optional[str] = None):
|
||
|
|
"""Holt die letzten Audit-Einträge."""
|
||
|
|
from db import list_audit
|
||
|
|
return await list_audit(db, agent_id=agent_id, limit=limit)
|