cayce-ops/tests/test_audit_hardening.py

"""Round-3 hardening tests: append-only audit log + strict legacy-prefix rule
+ anchor cross-verification (issue: operator key WAS the agent key; the
audit trail must now be tamper-evident against the DB owner itself)."""

from __future__ import annotations

import pytest
from sqlalchemy import text

from cayceops.models import AuditEntry, audit, verify_audit_chain


def _seed(sess, n: int = 3) -> list[int]:
    ids = []
    for i in range(n):
        audit(sess, actor=f"daemon:{i}", action=f"act{i}", detail={"i": i})
        sess.flush()
        ids.append(sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first().id)
    return ids


# --- 1. legacy-prefix rule -----------------------------------------------------


def test_null_before_genesis_still_ok(sess):
    """Documented legacy prefix: NULL hashes before the first hashed row."""
    sess.add(AuditEntry(actor="legacy", action="old", detail={}, ts=__import__(
        "datetime").datetime.now(__import__("datetime").UTC)))
    sess.flush()
    _seed(sess, 2)
    sess.flush()
    ok, broken = verify_audit_chain(sess)
    assert ok and broken is None


def test_null_in_middle_fails(sess):
    """THE FIX: erasing a hash mid-chain is tampering and must be reported."""
    ids = _seed(sess, 3)
    middle = ids[1]
    sess.execute(
        text("UPDATE audit_log SET entry_hash = NULL WHERE id = :i"), {"i": middle}
    )
    sess.flush()
    ok, broken = verify_audit_chain(sess)
    assert not ok
    assert broken == middle


def test_null_after_chain_started_reports_the_row(sess):
    ids = _seed(sess, 4)
    sess.execute(
        text("UPDATE audit_log SET entry_hash = NULL WHERE id = :i"), {"i": ids[3]}
    )
    sess.flush()
    ok, broken = verify_audit_chain(sess)
    assert not ok and broken == ids[3]


# --- 2/3. append-only enforcement (migration 0004) ------------------------------


def _apply_0004(sess):
    import os

    path = os.path.join(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
        "alembic",
        "versions",
        "0004_audit_append_only.py",
    )
    spec = {}
    with open(path) as fh:
        exec(compile(fh.read(), path, "exec"), spec)
    # reuse the sqlite trigger DDL directly (dialect-aware inside the module)
    for ddl in spec["SQLITE_TRIGGERS"]:
        sess.execute(text(ddl))
    sess.commit()


def test_update_rejected_after_migration(sess):
    _seed(sess, 1)
    _apply_0004(sess)
    with pytest.raises(Exception, match="append-only"):
        sess.execute(text("UPDATE audit_log SET actor = 'evil'"))
        sess.flush()


def test_delete_rejected_after_migration(sess):
    _seed(sess, 1)
    _apply_0004(sess)
    with pytest.raises(Exception, match="append-only"):
        sess.execute(text("DELETE FROM audit_log"))
        sess.flush()


def test_insert_still_allowed_after_migration(sess):
    _seed(sess, 1)
    _apply_0004(sess)
    audit(sess, actor="daemon:x", action="still-works", detail={})
    sess.flush()
    ok, broken = verify_audit_chain(sess)
    assert ok and broken is None


# --- 4. anchor cross-verification ----------------------------------------------


def test_no_anchors_reported_explicitly(sess):
    from cayceops.components.monitor.monitor import verify_chain_against_anchors

    _seed(sess, 2)
    report = verify_chain_against_anchors(sess)
    assert report["status"] == "no_anchors"
    assert "never been anchored" in report["detail"]


def test_local_only_when_no_fetcher(sess):
    from datetime import UTC, datetime

    from cayceops.components.monitor.monitor import verify_chain_against_anchors

    _seed(sess, 2)
    last = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first()
    sess.add(
        AuditEntry(
            actor="daemon:monitor",
            action="anchor",
            detail={
                "event_id": "deadbeef",
                "last_hash": last.entry_hash,
                "last_entry": last.id,
            },
            ts=datetime.now(UTC),
        )
    )
    sess.flush()
    report = verify_chain_against_anchors(sess)
    assert report["status"] == "local_only"
    assert "LOCAL record" in report["detail"]


def test_anchor_mismatch_detected(sess):
    from datetime import UTC, datetime

    from cayceops.components.monitor.monitor import verify_chain_against_anchors

    _seed(sess, 2)
    last = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first()
    sess.add(
        AuditEntry(
            actor="daemon:monitor",
            action="anchor",
            detail={"event_id": "beef", "last_hash": last.entry_hash, "last_entry": last.id},
            ts=datetime.now(UTC),
        )
    )
    sess.flush()

    def foreign_fetcher(deployment):
        # a note attesting some OTHER chain entirely
        return ["cayce-ops audit anchor deployment=itest last_hash=" + "aa" * 32]

    report = verify_chain_against_anchors(sess, relay_fetcher=foreign_fetcher)
    assert report["status"] == "mismatch"
    assert "does not attest" in report["detail"]


def test_anchor_match_ok(sess):
    from datetime import UTC, datetime

    from cayceops.components.monitor.monitor import verify_chain_against_anchors

    _seed(sess, 2)
    last = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first()
    sess.add(
        AuditEntry(
            actor="daemon:monitor",
            action="anchor",
            detail={"event_id": "beef", "last_hash": last.entry_hash, "last_entry": last.id},
            ts=datetime.now(UTC),
        )
    )
    sess.flush()

    def honest_fetcher(deployment):
        return [f"cayce-ops audit anchor deployment={deployment} last_hash={last.entry_hash}"]

    report = verify_chain_against_anchors(sess, relay_fetcher=honest_fetcher)
    assert report["status"] == "ok"