vermietungs-tool/app/audit.py

"""The audit log (migration 002): every row a data table gains, loses or changes, with who wrote it — and
undoing chosen entries.

python -m app.audit [--table T] [--row ID] [--actor A] [--tx TX] [--limit N]    newest entries first
python -m app.audit revert (ENTRY... | --tx TX) [--apply]                       a dry run without --apply
"""
import argparse
import json

from app import db

KEYS = {"company_key": ("key",), "dismissed_finding": ("kind", "key")}   # every other table: id


class Conflict(Exception):
    """The row no longer looks as the entry left it: someone changed it since."""


def entries(conn, table=None, row_id=None, actor=None, tx=None, ids=None, limit=50):
    """Log entries, newest first."""
    where, params = [], []
    for sql, value in (("table_name = ?", table), ("row_id = ?", row_id), ("actor = ?", actor), ("tx = ?", tx)):
        if value is not None:
            where.append(sql)
            params.append(value)
    if ids:
        where.append("id = ANY(?)")
        params.append(list(ids))
    sql = "SELECT * FROM audit_log" + (" WHERE " + " AND ".join(where) if where else "")
    return [dict(r) for r in db.rows(conn, sql + " ORDER BY id DESC LIMIT ?", params + [limit])]


def _where(table, row):
    cols = KEYS.get(table, ("id",))
    return " AND ".join(f'"{c}" = ?' for c in cols), [row[c] for c in cols]


def revert(conn, chosen):
    """Undo the chosen entries newest first, in the caller's transaction (commit or roll back after). Each
    step first checks that the row still looks exactly as the entry left it — a later change by anyone else
    raises Conflict, so the caller rolls everything back. Returns one line per step."""
    conn.execute("SET CONSTRAINTS ALL DEFERRED")            # parents and children come back in any order
    steps = []
    for e in sorted(chosen, key=lambda e: e["id"], reverse=True):
        table, op = e["table_name"], e["op"]
        if table not in db.DATA_TABLES or op == "TRUNCATE":
            raise Conflict(f"#{e['id']}: {op} on {table} cannot be undone row by row — restore a snapshot")
        where, key = _where(table, e["new"] or e["old"])
        now = db.row(conn, f"SELECT to_jsonb(x) FROM {table} x WHERE {where}", key)
        if (now[0] if now else None) != e["new"]:
            raise Conflict(f"#{e['id']} {table} {key}: changed since — now {now[0] if now else 'gone'}")
        if op == "INSERT":
            conn.execute(f"DELETE FROM {table} WHERE {where}", key)
        elif op == "DELETE":
            conn.execute(f"INSERT INTO {table} SELECT * FROM jsonb_populate_record(NULL::{table}, "
                         f"(SELECT old FROM audit_log WHERE id = ?))", (e["id"],))
        else:
            cols = ", ".join(f'"{c}"' for c in e["old"] if e["old"][c] != e["new"].get(c))
            conn.execute(f"UPDATE {table} SET ({cols}) = (SELECT {cols} FROM jsonb_populate_record("
                         f"NULL::{table}, (SELECT old FROM audit_log WHERE id = ?))) WHERE {where}",
                         [e["id"]] + key)
        steps.append(f"#{e['id']} {op} {table} {key} undone")
    return steps


def describe(e):
    if e["op"] == "UPDATE":
        what = "; ".join(f"{c}: {e['old'][c]!r} -> {e['new'].get(c)!r}"
                         for c in e["old"] if e["old"][c] != e["new"].get(c))
    else:
        what = json.dumps(e["new"] or e["old"], ensure_ascii=False)
    return (f"#{e['id']} {e['at']:%Y-%m-%d %H:%M:%S} tx {e['tx']} {e['actor']} ({e['db_user']}) "
            f"{e['op']} {e['table_name']} {e['row_id'] or ''}  {what[:300]}")


def main():
    p = argparse.ArgumentParser(prog="python -m app.audit")
    p.add_argument("command", nargs="?", choices=["list", "revert"], default="list")
    p.add_argument("ids", nargs="*", type=int, help="entries to revert")
    p.add_argument("--table")
    p.add_argument("--row", type=int)
    p.add_argument("--actor")
    p.add_argument("--tx", type=int)
    p.add_argument("--limit", type=int, default=50)
    p.add_argument("--apply", action="store_true", help="write the revert (default: dry run)")
    a = p.parse_args()
    if a.command == "list":
        for e in entries(db.connect(), a.table, a.row, a.actor, a.tx, limit=a.limit):
            print(describe(e))
        return
    if not (a.ids or a.tx):
        p.error("revert needs entry ids or --tx")
    if a.apply:
        with db.connect(owner=True) as owner:
            print("Snapshot saved:", db.snapshot(owner, "before revert"))
    label = f"revert {'tx ' + str(a.tx) if a.tx else ' '.join(f'#{i}' for i in a.ids)}"
    conn = db.connect(actor=label)
    chosen = entries(conn, tx=a.tx, ids=a.ids, limit=100000)
    if not chosen:
        raise SystemExit("no such entries")
    try:
        for step in revert(conn, chosen):
            print(step)
    except Conflict as c:
        conn.rollback()
        raise SystemExit(f"NOTHING written — {c}")
    if a.apply:
        conn.commit()
        print(f"reverted {len(chosen)} entries (logged as '{label}')")
    else:
        conn.rollback()
        print("dry run — nothing written; add --apply")


if __name__ == "__main__":
    main()