audit-core/tests/test_integrity.py

import json

from audit_core.ingestion import IngestionApplication
from audit_core.integrity import GENESIS, chain_link, load_attestation, write_attestation
from audit_core.interface import AuditEvent
from audit_core.sqlite_backend import SQLiteAuditBackend

from test_ingestion import invoke


def _event(event_id: str, **kw) -> AuditEvent:
    fields = dict(
        event_id=event_id,
        source="user-engine",
        action="membership.added",
        resource="membership-1",
        outcome="recorded",
        tenant="tenant:friendly:binky",
        scope="tenant",
        details={"correlation_id": "corr-1"},
        observed_at="2026-08-09T00:00:00+00:00",
    )
    fields.update(kw)
    return AuditEvent(**fields)


def _digest(event: AuditEvent) -> str:
    import hashlib

    return hashlib.sha256(
        json.dumps(event.as_record(), sort_keys=True).encode()
    ).hexdigest()


def test_first_accept_sets_genesis(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    event = _event("e1")
    backend.accept(event, _digest(event))
    row = backend.db.execute(
        "SELECT chain_prev, chain_hash FROM events WHERE event_id = 'e1'"
    ).fetchone()
    assert row[0] == GENESIS
    assert row[1] == chain_link(GENESIS, _digest(event), "e1")


def test_second_accept_links(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    first = _event("e1")
    second = _event("e2")
    backend.accept(first, _digest(first))
    backend.accept(second, _digest(second))
    head = backend.db.execute(
        "SELECT chain_hash FROM events WHERE event_id = 'e1'"
    ).fetchone()[0]
    prev = backend.db.execute(
        "SELECT chain_prev FROM events WHERE event_id = 'e2'"
    ).fetchone()[0]
    assert prev == head


def test_duplicate_does_not_fork(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    event = _event("e1")
    digest = _digest(event)
    assert backend.accept(event, digest).duplicate is False
    assert backend.accept(event, digest).duplicate is True
    count = backend.db.execute("SELECT count(*) FROM events").fetchone()[0]
    assert count == 1
    assert backend.verify_chain().events == 1


def test_verify_clean_on_fresh_store(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    empty = backend.verify_chain()
    assert empty.intact is True
    assert empty.events == 0
    assert empty.head == GENESIS
    first = _event("e1")
    second = _event("e2")
    backend.accept(first, _digest(first))
    backend.accept(second, _digest(second))
    report = backend.verify_chain()
    assert report.intact is True
    assert report.events == 2
    assert report.first_break is None


def test_rewritten_payload_fails_verify(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    first = _event("e1")
    second = _event("e2")
    backend.accept(first, _digest(first))
    backend.accept(second, _digest(second))
    backend.db.execute("UPDATE events SET payload_hash = 'deadbeef' WHERE event_id = 'e2'")
    report = backend.verify_chain()
    assert report.intact is False
    assert report.first_break == "e2"


def test_attestation_mismatch(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    event = _event("e1")
    backend.accept(event, _digest(event))
    path = tmp_path / "head.json"
    write_attestation(path, backend.verify_chain())
    cited = load_attestation(path)
    assert backend.verify_chain(cited).attestation_match is True
    cited["chain_hash"] = "f" * 64
    broken = backend.verify_chain(cited)
    assert broken.intact is False
    assert broken.first_break == "attestation_mismatch"
    assert broken.attestation_match is False


def test_attestation_still_matches_after_growth(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    first = _event("e1")
    backend.accept(first, _digest(first))
    cited = backend.attest_chain()
    second = _event("e2")
    backend.accept(second, _digest(second))
    report = backend.verify_chain(cited)
    assert report.intact is True
    assert report.attestation_match is True
    assert report.events == 2


def test_http_integrity_requires_read(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "c.db"))
    app = IngestionApplication(backend, "opaque")
    status, _ = invoke(app, None, path="/v1/integrity", method="GET", body=b"")
    assert status.startswith("200")
    event = _event("e1")
    backend.accept(event, _digest(event))
    status, body = invoke(app, None, path="/v1/integrity", method="GET", body=b"")
    assert status.startswith("200")
    assert body["intact"] is True
    assert body["events"] == 1
    assert "record" not in body


# --- AUDIT-WP-0009-T01: tamper_evidence is derived, not declared -----------


def _report(**kw):
    from audit_core.integrity import ChainReport

    fields = dict(
        intact=True,
        events=2,
        head="a" * 64,
        head_event_id="e2",
        head_accepted_at="2026-09-01T00:00:00+00:00",
        first_break=None,
        attestation_match=True,
    )
    fields.update(kw)
    return ChainReport(**fields)


def _at(hours_ago: float):
    from datetime import datetime, timedelta, timezone

    stamp = datetime(2026, 9, 6, tzinfo=timezone.utc) - timedelta(hours=hours_ago)
    return stamp.isoformat()


def _now():
    from datetime import datetime, timezone

    return datetime(2026, 9, 6, tzinfo=timezone.utc)


def test_fresh_matching_attestation_earns_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(
        _report(), {"chain_hash": "a" * 64, "observed_at": _at(6)}, now=_now()
    )
    assert state.claimed is True
    assert state.reason == "attested"


def test_absent_attestation_degrades_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(_report(attestation_match=None), None, now=_now())
    assert state.claimed is False
    assert state.reason == "no_attestation"


def test_stale_attestation_degrades_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    # The one attestation on record when this task was written was 21 days old.
    state = evaluate_tamper_evidence(
        _report(), {"chain_hash": "a" * 64, "observed_at": _at(21 * 24)}, now=_now()
    )
    assert state.claimed is False
    assert state.reason == "attestation_stale"
    assert state.age_seconds > state.max_age_seconds


def test_attestation_just_inside_the_window_still_counts():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(
        _report(), {"chain_hash": "a" * 64, "observed_at": _at(167)}, now=_now()
    )
    assert state.claimed is True


def test_mismatched_attestation_degrades_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(
        _report(attestation_match=False),
        {"chain_hash": "b" * 64, "observed_at": _at(1)},
        now=_now(),
    )
    assert state.claimed is False
    assert state.reason == "attestation_mismatch"


def test_undated_attestation_degrades_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(_report(), {"chain_hash": "a" * 64}, now=_now())
    assert state.claimed is False
    assert state.reason == "attestation_undated"


def test_broken_chain_degrades_the_claim_even_with_fresh_attestation():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(
        _report(intact=False, first_break="e2"),
        {"chain_hash": "a" * 64, "observed_at": _at(1)},
        now=_now(),
    )
    assert state.claimed is False
    assert state.reason == "chain_break"


def test_unreadable_chain_degrades_the_claim():
    from audit_core.integrity import evaluate_tamper_evidence

    state = evaluate_tamper_evidence(
        None, {"chain_hash": "a" * 64, "observed_at": _at(1)}, now=_now()
    )
    assert state.claimed is False
    assert state.reason == "chain_unreadable"


def _bare_postgres_backend(tmp_path, attestation_path=None):
    """A PostgresAuditBackend with no pool — retention_policy needs no I/O."""
    import pytest

    pytest.importorskip("psycopg", reason="needs psycopg to import the backend")
    from audit_core.integrity import DEFAULT_ATTESTATION_MAX_AGE_HOURS
    from audit_core.postgres_backend import PostgresAuditBackend

    backend = object.__new__(PostgresAuditBackend)
    backend.retention_days = None
    backend.recoverable_days = 30
    backend.recoverable_source = "test"
    backend.recoverable_basis = "measured"
    backend.attestation_path = str(attestation_path) if attestation_path else None
    backend.attestation_max_age_hours = DEFAULT_ATTESTATION_MAX_AGE_HOURS
    backend._attestation_cache_seconds = 0.0
    backend._tamper_state = None
    backend._tamper_state_at = 0.0
    return backend


def test_postgres_policy_drops_the_claim_without_an_attestation(tmp_path):
    backend = _bare_postgres_backend(tmp_path)
    backend.verify_chain = lambda attestation=None: _report(attestation_match=None)
    policy = backend.retention_policy
    assert policy.tamper_evidence is False
    assert policy.immutable is True
    assert backend.tamper_evidence_state().reason == "no_attestation"


def test_postgres_policy_earns_the_claim_from_a_fresh_attestation(tmp_path):
    from audit_core.integrity import utc_now

    path = tmp_path / "chain-head.json"
    path.write_text(json.dumps({"chain_hash": "a" * 64, "observed_at": utc_now()}))
    backend = _bare_postgres_backend(tmp_path, attestation_path=path)
    backend.verify_chain = lambda attestation=None: _report(
        attestation_match=attestation is not None
    )
    assert backend.retention_policy.tamper_evidence is True


def test_postgres_policy_treats_an_unreadable_attestation_as_absent(tmp_path):
    path = tmp_path / "chain-head.json"
    path.write_text("{not json")
    backend = _bare_postgres_backend(tmp_path, attestation_path=path)
    backend.verify_chain = lambda attestation=None: _report(attestation_match=None)
    assert backend.retention_policy.tamper_evidence is False
    assert backend.tamper_evidence_state().reason == "no_attestation"


def test_postgres_policy_drops_the_claim_when_the_chain_cannot_be_walked(tmp_path):
    from audit_core.integrity import utc_now

    path = tmp_path / "chain-head.json"
    path.write_text(json.dumps({"chain_hash": "a" * 64, "observed_at": utc_now()}))
    backend = _bare_postgres_backend(tmp_path, attestation_path=path)

    def _unavailable(attestation=None):
        raise RuntimeError("backend unavailable")

    backend.verify_chain = _unavailable
    assert backend.retention_policy.tamper_evidence is False
    assert backend.tamper_evidence_state().reason == "chain_unreadable"