cyclone/backend/tests/test_audit_log.py

"""Tests for the tamper-evident hash-chained audit_log.

SP11 T1.
"""
from __future__ import annotations

from datetime import datetime, timezone

import json
import pytest

from cyclone import db
from cyclone.audit_log import (
    GENESIS_PREV_HASH,
    HASH_LEN,
    AuditEvent,
    append_event,
    verify_chain,
)
from cyclone.db import AuditLog, init_db


@pytest.fixture(autouse=True)
def _fresh_db():
    init_db()
    yield


def _row_count() -> int:
    with db.SessionLocal()() as s:
        return s.query(AuditLog).count()


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


class TestAppendEvent:
    def test_first_row_has_genesis_prev_hash(self):
        """The first row in a fresh chain uses the all-zeros prev_hash."""
        with db.SessionLocal()() as s:
            row = append_event(s, AuditEvent(
                event_type="claim.parsed",
                entity_type="claim",
                entity_id="c1",
                payload={"patient_control_number": "PCN-1"},
            ))
            s.commit()
        assert row.id == 1
        assert row.prev_hash == GENESIS_PREV_HASH
        assert len(row.hash) == HASH_LEN

    def test_second_row_chains_to_first(self):
        """Row N's prev_hash == row N-1's hash."""
        with db.SessionLocal()() as s:
            r1 = append_event(s, AuditEvent(
                event_type="claim.parsed", entity_type="claim", entity_id="c1",
            ))
            s.commit()
            r2 = append_event(s, AuditEvent(
                event_type="claim.rejected", entity_type="claim", entity_id="c1",
            ))
            s.commit()
        assert r2.prev_hash == r1.hash
        assert r2.id == r1.id + 1

    def test_payload_canonicalization(self):
        """Two payloads with the same content but different key order hash the same.

        We rely on sort_keys=True to make the canonical form independent
        of dict insertion order.
        """
        with db.SessionLocal()() as s:
            r1 = append_event(s, AuditEvent(
                event_type="x", entity_type="y", entity_id="z",
                payload={"a": 1, "b": 2},
            ))
            s.commit()
            # New session to drop any in-memory ordering cache.
            with db.SessionLocal()() as s2:
                r2 = append_event(s2, AuditEvent(
                    event_type="x", entity_type="y", entity_id="z",
                    payload={"b": 2, "a": 1},
                ))
                s2.commit()
        # The two rows have different IDs and different prev_hash
        # inputs, but their hash *recipe* (modulo id + prev_hash) is
        # the same. Just confirm the rows are different.
        assert r1.id != r2.id
        # The payload_json fields ARE byte-identical (canonical form).
        assert r1.payload_json == r2.payload_json


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


class TestVerifyChain:
    def test_empty_chain_is_ok(self):
        """No rows = nothing to verify = ok."""
        with db.SessionLocal()() as s:
            result = verify_chain(s)
        assert result.ok is True
        assert result.checked == 0

    def test_single_row_chain_is_ok(self):
        with db.SessionLocal()() as s:
            append_event(s, AuditEvent("x", "y", "z"))
            s.commit()
            result = verify_chain(s)
        assert result.ok is True
        assert result.checked == 1

    def test_long_chain_is_ok(self):
        with db.SessionLocal()() as s:
            for i in range(50):
                append_event(s, AuditEvent("x", "y", f"id-{i}"))
            s.commit()
            result = verify_chain(s)
        assert result.ok is True
        assert result.checked == 50

    def test_tampered_payload_detected(self):
        """Modifying a row's payload_json breaks the chain at that row."""
        with db.SessionLocal()() as s:
            for i in range(5):
                append_event(s, AuditEvent("x", "y", f"id-{i}"))
            s.commit()
        # Tamper with row #3's payload (the first 3 rows are still
        # valid; row 3 will fail).
        with db.SessionLocal()() as s:
            row = s.query(AuditLog).filter(AuditLog.id == 3).first()
            row.payload_json = json.dumps({"evil": "tampered"})
            s.commit()
        with db.SessionLocal()() as s:
            result = verify_chain(s)
        assert result.ok is False
        assert result.first_bad_id == 3
        assert "hash" in (result.reason or "").lower()

    def test_tampered_prev_hash_detected(self):
        """Modifying a row's prev_hash invalidates that row's own hash.

        Because the row's hash is computed from prev_hash (and the
        other fields), changing prev_hash changes the row's own hash,
        so the verifier detects the breakage at the tampered row
        itself (not the row after). The next-row check (which compares
        prev_hash to the previous row's hash) would catch a different
        class of tampering: a row whose prev_hash was set to a value
        matching the previous row's hash but whose own hash was also
        regenerated — but that requires recomputing the hash to match,
        which is what verify_chain would not detect by itself (the
        content-vs-hash check still catches content tampering).
        """
        with db.SessionLocal()() as s:
            for i in range(5):
                append_event(s, AuditEvent("x", "y", f"id-{i}"))
            s.commit()
        with db.SessionLocal()() as s:
            row = s.query(AuditLog).filter(AuditLog.id == 2).first()
            row.prev_hash = "f" * 64  # fake
            s.commit()
        with db.SessionLocal()() as s:
            result = verify_chain(s)
        assert result.ok is False
        # The tampered row itself fails its hash check first.
        assert result.first_bad_id == 2
        assert "hash" in (result.reason or "").lower()

    def test_deleted_row_detected(self):
        """Deleting a middle row breaks the chain at the row after it."""
        with db.SessionLocal()() as s:
            for i in range(5):
                append_event(s, AuditEvent("x", "y", f"id-{i}"))
            s.commit()
        with db.SessionLocal()() as s:
            row = s.query(AuditLog).filter(AuditLog.id == 3).first()
            s.delete(row)
            s.commit()
        with db.SessionLocal()() as s:
            result = verify_chain(s)
        assert result.ok is False
        # Row 3 is gone; row 4's prev_hash now points at row 2's hash,
        # which doesn't match row 4's stored prev_hash.
        assert result.first_bad_id == 4