PacketStation/server/packetstation/audit.py

"""Append-only audit log with a hash chain: row_hash = sha256(prev_hash || canonical json of the row)."""

from __future__ import annotations

import hashlib
import ipaddress
import json
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

AUDIT_LOCK = 7311

FIELDS = (
    "at",
    "user_id",
    "session_id",
    "role_snapshot",
    "action",
    "subject_type",
    "subject_id",
    "packet_id",
    "detail",
    "it_reason",
    "ip",
)


def _norm(name: str, value: Any) -> Any:
    if value is None:
        return None
    if name == "at":
        if not isinstance(value, datetime):
            raise TypeError("at must be a datetime")
        return value.astimezone(UTC).isoformat(timespec="microseconds")
    if name in ("user_id", "session_id"):
        return str(uuid.UUID(str(value)))
    if name == "ip":
        return str(ipaddress.ip_address(str(value).split("/")[0]))
    if name == "packet_id":
        return int(value)
    if name == "detail":
        # Round-trip through JSON so the value hashes the same before insert and after reading jsonb
        return json.loads(json.dumps(value, default=str))
    return str(value)


def canonical(fields: dict[str, Any]) -> bytes:
    """Sorted keys, no spaces, UTF-8; missing fields count as null."""
    body = {k: _norm(k, fields.get(k)) for k in FIELDS}
    return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def row_hash(prev_hash: bytes | None, fields: dict[str, Any]) -> bytes:
    return hashlib.sha256((prev_hash or b"") + canonical(fields)).digest()


@dataclass
class Appended:
    id: int
    row_hash: bytes


def append(
    conn: Any,
    *,
    action: str,
    user_id: Any = None,
    session_id: Any = None,
    role_snapshot: str | None = None,
    subject_type: str | None = None,
    subject_id: Any = None,
    packet_id: int | None = None,
    detail: Any = None,
    it_reason: str | None = None,
    ip: str | None = None,
) -> Appended:
    """Insert one audit row chained to the previous one. Serialised by pg_advisory_xact_lock(7311)."""
    from psycopg.types.json import Jsonb

    with conn.transaction():
        conn.execute("SELECT pg_advisory_xact_lock(%s)", (AUDIT_LOCK,))
        last = conn.execute("SELECT row_hash FROM audit_events ORDER BY id DESC LIMIT 1").fetchone()
        prev = bytes(last[0]) if last and last[0] is not None else None
        at = conn.execute("SELECT now()").fetchone()[0]
        fields = {
            "at": at,
            "user_id": user_id,
            "session_id": session_id,
            "role_snapshot": role_snapshot,
            "action": action,
            "subject_type": subject_type,
            "subject_id": None if subject_id is None else str(subject_id),
            "packet_id": packet_id,
            "detail": detail,
            "it_reason": it_reason,
            "ip": ip,
        }
        h = row_hash(prev, fields)
        row = conn.execute(
            "INSERT INTO audit_events (at, prev_hash, row_hash, user_id, session_id, role_snapshot, action,"
            " subject_type, subject_id, packet_id, detail, it_reason, ip)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id",
            (
                at,
                prev,
                h,
                user_id,
                session_id,
                role_snapshot,
                action,
                subject_type,
                fields["subject_id"],
                packet_id,
                None if detail is None else Jsonb(detail),
                it_reason,
                ip,
            ),
        ).fetchone()
    return Appended(int(row[0]), h)


def verify_chain(conn: Any) -> tuple[bool, int, str]:
    """Walk the chain from the start: (ok, rows checked, detail)."""
    cols = ", ".join(FIELDS)
    cur = conn.execute(f"SELECT id, prev_hash, row_hash, {cols} FROM audit_events ORDER BY id")  # noqa: S608
    prev: bytes | None = None
    n = 0
    for rec in cur:
        rid, p, h = rec[0], rec[1], rec[2]
        p = bytes(p) if p is not None else None
        if p != prev:
            return False, n, f"row {rid}: prev_hash does not match row before it"
        fields = dict(zip(FIELDS, rec[3:], strict=True))
        if row_hash(prev, fields) != bytes(h):
            return False, n, f"row {rid}: row_hash does not match its content"
        prev = bytes(h)
        n += 1
    return True, n, f"{n} rows verified"