58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
"""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"))
|
|
_local = threading.local()
|
|
|
|
|
|
def init():
|
|
_db.parent.mkdir(parents=True, exist_ok=True)
|
|
with _conn() as 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 _conn():
|
|
if not hasattr(_local, "c") or _local.c is None:
|
|
_local.c = sqlite3.connect(str(_db), timeout=5)
|
|
_local.c.row_factory = sqlite3.Row
|
|
return _local.c
|
|
|
|
|
|
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 _conn() as 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 query(limit=100, offset=0) -> list[dict]:
|
|
try:
|
|
with _conn() as c:
|
|
rows = c.execute("SELECT * FROM log ORDER BY ts DESC LIMIT ? OFFSET ?",
|
|
(limit, offset)).fetchall()
|
|
return [dict(r) for r in rows]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def count() -> int:
|
|
try:
|
|
with _conn() as c:
|
|
return c.execute("SELECT count(*) FROM log").fetchone()[0]
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def clear():
|
|
try:
|
|
with _conn() as c:
|
|
c.execute("DELETE FROM log")
|
|
except Exception:
|
|
pass
|