server-up/server-up/core/audit.py

"""Audit log backed by SQLite."""
import sqlite3, time, json, os, threading
from pathlib import Path

_db = Path(os.environ.get("SU_AUDIT", "/data/audit.db"))

# Eén gedeelde verbinding achter een lock. Eerder kreeg elke thread een eigen
# thread-local verbinding die nooit gesloten werd; omdat iedere job een nieuwe
# thread start, lekte dat een file descriptor per job.
_conn_obj: sqlite3.Connection | None = None
_lock = threading.Lock()


def init():
    _db.parent.mkdir(parents=True, exist_ok=True)
    with _lock:
        c = _connect()
        with c:
            c.execute("""CREATE TABLE IF NOT EXISTS log(
                id INTEGER PRIMARY KEY, ts REAL,
                src TEXT, action TEXT, status TEXT,
                ref TEXT, detail TEXT, ip TEXT)""")


def _connect() -> sqlite3.Connection:
    """Maak (eenmalig) de gedeelde verbinding. Aanroepen onder _lock."""
    global _conn_obj
    if _conn_obj is None:
        _db.parent.mkdir(parents=True, exist_ok=True)
        _conn_obj = sqlite3.connect(str(_db), timeout=5, check_same_thread=False)
        _conn_obj.row_factory = sqlite3.Row
        # WAL laat lezers en schrijvers naast elkaar werken.
        _conn_obj.execute("PRAGMA journal_mode=WAL")
    return _conn_obj


def close():
    """Sluit de gedeelde verbinding (gebruikt door tests en bij afsluiten)."""
    global _conn_obj
    with _lock:
        if _conn_obj is not None:
            try:
                _conn_obj.close()
            finally:
                _conn_obj = None


def log(src: str, action: str, status="ok", ref="", detail=None, ip=""):
    d = json.dumps(detail) if isinstance(detail, (dict, list)) else str(detail or "")
    try:
        with _lock:
            c = _connect()
            with c:
                c.execute("INSERT INTO log(ts,src,action,status,ref,detail,ip) "
                          "VALUES(?,?,?,?,?,?,?)",
                          (time.time(), src, action, status, ref, d, ip))
    except Exception:
        pass


def _filter(src="", action="", status="", zoek="", sinds=None):
    """Bouw de WHERE-clausule op uit de opgegeven filters."""
    delen, params = [], []
    if src:
        delen.append("src = ?"); params.append(src)
    if action:
        delen.append("action = ?"); params.append(action)
    if status:
        delen.append("status = ?"); params.append(status)
    if sinds:
        delen.append("ts >= ?"); params.append(float(sinds))
    if zoek:
        delen.append("(ref LIKE ? OR detail LIKE ? OR ip LIKE ?)")
        params += [f"%{zoek}%"] * 3
    return (" WHERE " + " AND ".join(delen)) if delen else "", params


def query(limit=100, offset=0, **filters) -> list[dict]:
    """Regels uit het log, nieuwste eerst, optioneel gefilterd.

    Zonder filters groeide het log snel onbruikbaar: sinds er logins,
    rolwijzigingen, backups en updates in landen zijn de laatste tweehonderd
    regels al na een dag beheerwerk niet meer representatief.
    """
    where, params = _filter(**filters)
    try:
        with _lock:
            rows = _connect().execute(
                f"SELECT * FROM log{where} ORDER BY ts DESC LIMIT ? OFFSET ?",
                (*params, limit, offset)).fetchall()
        return [dict(r) for r in rows]
    except Exception:
        return []


def count(**filters) -> int:
    where, params = _filter(**filters)
    try:
        with _lock:
            return _connect().execute(
                f"SELECT count(*) FROM log{where}", params).fetchone()[0]
    except Exception:
        return 0


def facets() -> dict:
    """Welke bronnen, acties en statussen komen voor — voor de filterkeuzes."""
    uit = {"src": [], "action": [], "status": []}
    try:
        with _lock:
            c = _connect()
            for kolom in uit:
                rows = c.execute(
                    f"SELECT DISTINCT {kolom} FROM log "
                    f"WHERE {kolom} != '' ORDER BY {kolom}").fetchall()
                uit[kolom] = [r[0] for r in rows]
    except Exception:
        pass
    return uit


def clear():
    try:
        with _lock:
            c = _connect()
            with c:
                c.execute("DELETE FROM log")
    except Exception:
        pass