cyclone/backend/src/cyclone/audit_log.py

"""Tamper-evident hash-chained audit_log.

SP11.

Each row's hash is SHA-256 of
``(id, event_type, entity_type, entity_id, actor, payload_json,
created_at, prev_hash)`` and ``prev_hash`` is the previous row's hash.
That forms a chain: changing any row's payload invalidates every
subsequent row's hash. :func:`verify_chain` walks the chain and
returns the first mismatch index (or ``None`` for a clean chain).

We use SHA-256 (FIPS-approved, fast on commodity hardware) instead
of a Merkle tree because the chain is linear: every row depends on
exactly one prior row. A Merkle tree would let us prove individual
membership with O(log n) witnesses, but the chain's whole point is
end-to-end integrity, not selective disclosure.

Append-only by convention: the application MUST NOT call
``session.delete(row)`` or modify an existing row. Doing so is
auditable via :func:`verify_chain`. We deliberately do not enforce
this at the DB level (no triggers, no revoked UPDATE permission)
because that breaks the test fixtures that recreate the DB.
"""
from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

from sqlalchemy.orm import Session

from cyclone.db import AuditLog

log = logging.getLogger(__name__)

# 64 hex chars = 256 bits. Constant for easy comparison.
HASH_LEN = 64

# Genesis row's prev_hash — a fixed "all zeros" sentinel so the first
# row in the chain has a deterministic predecessor. This is the same
# convention Bitcoin and other ledgers use.
GENESIS_PREV_HASH = "0" * HASH_LEN


# --------------------------------------------------------------------------- #
# Hashing
# --------------------------------------------------------------------------- #


def _hash_row(
    *,
    row_id: int,
    event_type: str,
    entity_type: str,
    entity_id: str,
    actor: str,
    payload_json: str | None,
    created_at: datetime,
    prev_hash: str,
) -> str:
    """Compute SHA-256 hex of a row's canonical form.

    The fields are concatenated with a separator that cannot appear
    inside any field (``\\x1f`` — the ASCII unit separator). Using a
    delimiter avoids length-ambiguity attacks where two different
    payloads with the same string-joined form would hash to the same
    digest.
    """
    sep = "\x1f"
    # Normalize the timestamp to an ISO 8601 UTC string so the hash is
    # stable across timezone-aware and timezone-naive datetimes (the
    # DB may give us either depending on the SQLite build).
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=timezone.utc)
    created_at_iso = created_at.astimezone(timezone.utc).isoformat()
    payload = payload_json or ""
    canonical = sep.join([
        str(row_id),
        event_type,
        entity_type,
        entity_id,
        actor,
        created_at_iso,
        payload,
        prev_hash,
    ])
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


# --------------------------------------------------------------------------- #
# Append
# --------------------------------------------------------------------------- #


@dataclass
class AuditEvent:
    """An audit event ready to be appended.

    Mirrors the ``AuditLog`` row shape minus the auto-assigned id and
    computed hash. Payload must be JSON-serializable; the audit_log
    module handles the encoding so callers don't need to think about
    canonical form.

    ``user_id`` is the authenticated actor for this event, when known
    (e.g. a parse-999 call made by user 7). It's stored on the row but
    is NOT part of the hash chain — the chain hashes only the fields
    that existed pre-SP-auth so verify_chain stays compatible with
    pre-auth rows.
    """

    event_type: str
    entity_type: str
    entity_id: str
    payload: dict[str, Any] = field(default_factory=dict)
    actor: str = "system"
    created_at: datetime | None = None
    user_id: int | None = None


def append_event(
    session: Session,
    event: AuditEvent,
) -> AuditLog:
    """Append one event to the audit_log chain and return the row.

    The caller is responsible for ``session.commit()`` — this lets
    callers batch multiple appends into one transaction (e.g., a
    parser that appends one event per parsed claim).
    """
    # Read the latest hash within the same session so concurrent
    # appends don't see stale state. SQLite default isolation level
    # gives us serializable reads for this query; for Postgres we'd
    # need SELECT ... FOR UPDATE but that's overkill for v1.
    latest = (
        session.query(AuditLog)
        .order_by(AuditLog.id.desc())
        .first()
    )
    prev_hash = latest.hash if latest is not None else GENESIS_PREV_HASH

    created_at = event.created_at or datetime.now(timezone.utc)
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=timezone.utc)

    # Canonical payload form: sort_keys + compact separators. This
    # makes the hash independent of dict insertion order across
    # Python versions and across API runs.
    payload_json = json.dumps(event.payload, sort_keys=True, separators=(",", ":")) if event.payload else None

    # Insert the row with a placeholder hash, then UPDATE once we
    # know the auto-assigned id. SQLite + SQLAlchemy gives us the id
    # after the INSERT, so we can compute the real hash then.
    row = AuditLog(
        event_type=event.event_type,
        entity_type=event.entity_type,
        entity_id=event.entity_id,
        actor=event.actor,
        payload_json=payload_json,
        created_at=created_at,
        prev_hash=prev_hash,
        hash=GENESIS_PREV_HASH,  # placeholder; updated below
        user_id=event.user_id,
    )
    session.add(row)
    session.flush()  # populate row.id
    row.hash = _hash_row(
        row_id=row.id,
        event_type=row.event_type,
        entity_type=row.entity_type,
        entity_id=row.entity_id,
        actor=row.actor,
        payload_json=row.payload_json,
        created_at=row.created_at,
        prev_hash=row.prev_hash,
    )
    session.flush()
    return row


# --------------------------------------------------------------------------- #
# Verify
# --------------------------------------------------------------------------- #


@dataclass
class VerifyResult:
    """Outcome of :func:`verify_chain`."""

    ok: bool
    checked: int
    first_bad_id: int | None = None
    reason: str | None = None


def verify_chain(session: Session) -> VerifyResult:
    """Walk the audit_log and verify every row's hash. Returns the first mismatch.

    A clean chain returns ``VerifyResult(ok=True, checked=N)``. A
    broken chain returns ``ok=False, first_bad_id=X, reason='...'``
    describing what went wrong (hash mismatch, prev_hash mismatch,
    or non-monotonic id).

    This is intended to be called by the operator (e.g., a nightly
    cron job or the admin UI's "Verify Audit Chain" button). It is
    NOT a fast operation — for a 6-year-old chain with millions of
    rows, expect seconds-to-minutes. Call it rarely.
    """
    rows = session.query(AuditLog).order_by(AuditLog.id.asc()).all()
    if not rows:
        return VerifyResult(ok=True, checked=0)

    expected_prev = GENESIS_PREV_HASH
    last_id = 0
    for i, row in enumerate(rows):
        # Monotonic id check — covers attempted inserts with a
        # custom id, or accidental out-of-order rows.
        if row.id <= last_id:
            return VerifyResult(
                ok=False, checked=i, first_bad_id=row.id,
                reason=f"non-monotonic id (previous={last_id}, this={row.id})",
            )
        last_id = row.id

        # Recompute the hash from the row's content and compare.
        expected_hash = _hash_row(
            row_id=row.id,
            event_type=row.event_type,
            entity_type=row.entity_type,
            entity_id=row.entity_id,
            actor=row.actor,
            payload_json=row.payload_json,
            created_at=row.created_at,
            prev_hash=row.prev_hash,
        )
        if expected_hash != row.hash:
            return VerifyResult(
                ok=False, checked=i, first_bad_id=row.id,
                reason=f"hash mismatch (stored={row.hash[:16]}…, computed={expected_hash[:16]}…)",
            )

        # Check prev_hash linkage.
        if row.prev_hash != expected_prev:
            return VerifyResult(
                ok=False, checked=i, first_bad_id=row.id,
                reason=f"prev_hash mismatch (stored={row.prev_hash[:16]}…, expected={expected_prev[:16]}…)",
            )
        expected_prev = row.hash

    return VerifyResult(ok=True, checked=len(rows))


__all__ = [
    "AuditEvent",
    "GENESIS_PREV_HASH",
    "HASH_LEN",
    "VerifyResult",
    "append_event",
    "verify_chain",
]