audit-core/tests/test_backend_conformance.py

"""Behaviour every durable audit backend must satisfy.

One suite, run against every backend. This is what makes "the Postgres backend
is done" mean something: it is the same contract SQLite already passes, not a
parallel set of tests that happen to be green.

Postgres tests are skipped unless a server is reachable. Point
``AUDIT_CORE_TEST_DATABASE_URL`` at one, or run ``make pg-test-up`` to start a
throwaway container.
"""

from __future__ import annotations

import hashlib
import json
import os
import threading
import uuid

import pytest

from audit_core.interface import (
    AuditEvent,
    BackendUnavailableError,
    EventConflictError,
)
from audit_core.redaction import Finding
from audit_core.sqlite_backend import SQLiteAuditBackend

PG_URL = os.environ.get("AUDIT_CORE_TEST_DATABASE_URL")


def _postgres_available() -> bool:
    if not PG_URL:
        return False
    try:
        import psycopg

        with psycopg.connect(PG_URL, connect_timeout=3):
            return True
    except Exception:
        return False


HAVE_PG = _postgres_available()


@pytest.fixture(params=["sqlite", "postgres"])
def backend(request, tmp_path):
    if request.param == "sqlite":
        yield SQLiteAuditBackend(str(tmp_path / "conformance.db"))
        return
    if not HAVE_PG:
        pytest.skip("no PostgreSQL reachable (set AUDIT_CORE_TEST_DATABASE_URL)")
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    instance = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        yield instance
    finally:
        with instance.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        instance.close()


def make_event(event_id="evt-1", **kw):
    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", "data": {"a": 1}},
        observed_at="2026-08-09T00:00:00+00:00",
    )
    fields.update(kw)
    return AuditEvent(**fields)


def digest(event: AuditEvent) -> str:
    return hashlib.sha256(
        json.dumps(event.as_record(), sort_keys=True).encode()
    ).hexdigest()


# --- the custody contract ---------------------------------------------------

def test_declares_a_retention_policy(backend):
    policy = backend.retention_policy
    assert policy.durable is True
    assert policy.custody_class in ("development", "operational", "archive", "hot_search")
    if policy.custody_class == "operational":
        assert policy.recoverable_days == 30
        assert policy.recoverable_basis == "measured"
        assert policy.recoverable_source
    # A backend claiming tamper evidence must also claim immutability;
    # the reverse is allowed.
    if policy.tamper_evidence:
        assert policy.immutable


def test_accepts_then_reports_duplicate(backend):
    event = make_event()
    first = backend.accept(event, digest(event))
    assert first.duplicate is False
    second = backend.accept(event, digest(event))
    assert second.duplicate is True
    assert second.reference == first.reference


def test_same_id_different_payload_conflicts(backend):
    event = make_event()
    backend.accept(event, digest(event))
    with pytest.raises(EventConflictError):
        backend.accept(event, "a-different-hash")


def test_rejects_an_invalid_event(backend):
    from audit_core.interface import EventValidationError

    with pytest.raises((EventValidationError, ValueError)):
        backend.accept(make_event(tenant=""), "hash")


# --- concurrency ------------------------------------------------------------

def test_concurrent_accept_yields_exactly_one_first(backend):
    """The assertion the service rests on, checked per backend.

    An early SQLite implementation passed every serial test while telling two
    concurrent callers they were both first, so this is not theoretical.
    """
    event = make_event("race-1")
    payload_hash = digest(event)
    results: list = []
    lock = threading.Lock()
    barrier = threading.Barrier(12)

    def submit():
        barrier.wait()
        try:
            outcome = backend.accept(event, payload_hash).duplicate
        except Exception as exc:  # recorded, not swallowed
            outcome = type(exc).__name__
        with lock:
            results.append(outcome)

    threads = [threading.Thread(target=submit) for _ in range(12)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert results.count(False) == 1, results
    assert results.count(True) == 11, results


# --- durability and reads ---------------------------------------------------

def test_lookup_by_event_id_and_correlation(backend):
    backend.accept(make_event("e1"), "h1")
    backend.accept(make_event("e2"), "h2")

    record = backend.get("e1")
    assert record["event_id"] == "e1"
    assert record["tenant"] == "tenant:friendly:binky"
    assert record["accepted_at"]

    assert backend.get("missing") is None
    assert {e["event_id"] for e in backend.by_correlation("corr-1")} == {"e1", "e2"}


def test_dead_letters_withhold_secret_payloads(backend):
    backend.record_rejection(
        event_id="e9", reason="secret_shaped_field", payload_hash="h",
        sender="user-engine", payload='{"password":"hunter2"}',
    )
    backend.record_rejection(
        event_id="e8", reason="source_not_allowed", payload_hash="h2",
        sender="user-engine", payload='{"source":"nope"}',
    )
    entries = {d["event_id"]: d for d in backend.dead_letters()}
    assert entries["e9"]["payload"] is None
    assert entries["e9"]["payload_withheld"] is True
    assert entries["e8"]["payload"] is not None
    assert entries["e8"]["payload_withheld"] is False


def test_secret_findings_count_per_path(backend):
    findings = [Finding("data.auth_token", True)]
    for _ in range(3):
        backend.count_secret_findings(
            sender="user-engine", source="user-engine",
            action="membership.added", outcome="redacted", findings=findings,
        )
    row = backend.secret_findings()[0]
    assert row["field_path"] == "data.auth_token"
    assert row["occurrences"] == 3
    assert row["persisted"] is True


def test_health_passes_on_a_live_backend(backend):
    backend.health()


def test_chain_links_and_verify_is_clean(backend):
    from audit_core.integrity import GENESIS

    first = make_event("chain-1")
    second = make_event("chain-2")
    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
    assert report.head != GENESIS
    replay = backend.accept(first, digest(first))
    assert replay.duplicate is True
    assert backend.verify_chain().events == report.events


# --- postgres-specific guarantees -------------------------------------------

pg_only = pytest.mark.skipif(not HAVE_PG, reason="needs PostgreSQL")


@pg_only
def test_rewritten_payload_fails_verify_postgres():
    """Superuser rewrite is the evidence the trigger never gave us."""
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        event = make_event("break-1")
        backend.accept(event, digest(event))
        other = make_event("break-2")
        backend.accept(other, digest(other))
        assert backend.verify_chain().intact is True
        with backend.pool.connection() as conn:
            conn.execute(f'ALTER TABLE "{schema}".events DISABLE TRIGGER events_append_only')
            conn.execute(
                f'UPDATE "{schema}".events SET payload_hash = %s WHERE event_id = %s',
                ("deadbeef" * 8, "break-2"),
            )
            conn.execute(f'ALTER TABLE "{schema}".events ENABLE TRIGGER events_append_only')
        report = backend.verify_chain()
        assert report.intact is False
        assert report.first_break == "break-2"
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_replay_reconciles_rather_than_duplicating():
    """Replay must never mint a second custody record for one source event."""
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        event = make_event("replay-1")
        backend.accept(event, digest(event))

        outcome = backend.replay("replay-1")
        assert outcome.duplicate is True

        rows = backend._query(
            f'SELECT count(*) FROM "{schema}".events WHERE event_id = %s', ("replay-1",)
        )
        assert rows[0][0] == 1

        with pytest.raises(KeyError):
            backend.replay("never-stored")
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_stored_events_are_append_only():
    """The basis for declaring immutable=True.

    Without this, a leaked runtime credential could rewrite or erase the audit
    trail — and the retention policy would be claiming a guarantee it does not
    have.
    """
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        event = make_event("immutable-1")
        backend.accept(event, digest(event))
        assert backend.retention_policy.immutable is True

        with pytest.raises(BackendUnavailableError):
            backend._execute(
                f'UPDATE "{schema}".events SET tenant = %s WHERE event_id = %s',
                ("tenant:coulomb", "immutable-1"),
            )
        with pytest.raises(BackendUnavailableError):
            backend._execute(
                f'DELETE FROM "{schema}".events WHERE event_id = %s', ("immutable-1",)
            )
        assert backend.get("immutable-1")["tenant"] == "tenant:friendly:binky"
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_migrations_are_idempotent_and_recorded():
    from audit_core.postgres_backend import MIGRATIONS, PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema, migrate=False)
    try:
        applied = backend.migrate()
        assert applied == [m[0] for m in MIGRATIONS]
        assert backend.migrate() == []  # second call is a no-op
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_connects_from_a_brokered_libpq_environment(monkeypatch):
    """The credential broker injects PG* vars into the child process rather
    than handing over a DSN, so an empty conninfo must work.

    This keeps the credential out of audit-core's configuration entirely.
    """
    import urllib.parse

    from audit_core.postgres_backend import PostgresAuditBackend

    parsed = urllib.parse.urlparse(PG_URL)
    monkeypatch.setenv("PGHOST", parsed.hostname or "127.0.0.1")
    monkeypatch.setenv("PGPORT", str(parsed.port or 5432))
    monkeypatch.setenv("PGUSER", parsed.username or "postgres")
    monkeypatch.setenv("PGPASSWORD", parsed.password or "")
    monkeypatch.setenv("PGDATABASE", (parsed.path or "/postgres").lstrip("/"))
    monkeypatch.delenv("AUDIT_CORE_DATABASE_URL", raising=False)

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(schema=schema)
    try:
        event = make_event("brokered-1")
        assert backend.accept(event, digest(event)).duplicate is False
        assert backend.get("brokered-1")["event_id"] == "brokered-1"
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


def test_missing_connection_information_is_a_clear_error(monkeypatch):
    for var in ("AUDIT_CORE_DATABASE_URL", "PGHOST", "PGUSER"):
        monkeypatch.delenv(var, raising=False)
    try:
        from audit_core.postgres_backend import PostgresAuditBackend
    except ImportError:
        pytest.skip("psycopg is not installed")

    with pytest.raises(ValueError, match="no connection information"):
        PostgresAuditBackend()


# --- rotating credentials from a mounted directory (T02) --------------------

def _write_credentials(directory, *, user, password, url):
    import urllib.parse

    parsed = urllib.parse.urlparse(url)
    (directory / "username").write_text(user)
    (directory / "password").write_text(password)
    (directory / "host").write_text(parsed.hostname or "127.0.0.1")
    (directory / "port").write_text(str(parsed.port or 5432))
    (directory / "dbname").write_text((parsed.path or "/postgres").lstrip("/"))


@pg_only
def test_rotating_a_mounted_credential_needs_no_restart(tmp_path):
    """The property env vars cannot provide.

    A dynamic lease changes while the pod runs. Reading the credential at
    connection time rather than at startup is what makes that rotation
    invisible to senders — no restart, no delivery gap.
    """
    import psycopg

    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"rot_{uuid.uuid4().hex[:10]}"
    role = f"lease_{uuid.uuid4().hex[:8]}"
    admin = psycopg.connect(PG_URL, autocommit=True)
    directory = tmp_path / "db"
    directory.mkdir()
    backend = None
    try:
        admin.execute(f"CREATE ROLE {role} LOGIN PASSWORD 'first' SUPERUSER")
        _write_credentials(directory, user=role, password="first", url=PG_URL)

        backend = PostgresAuditBackend(
            "", schema=schema, credential_dir=str(directory), min_size=1, max_size=2
        )
        event = make_event("rot-1")
        assert backend.accept(event, digest(event)).duplicate is False

        # Rotate: new password in the database, new password in the mount.
        # The process is never restarted and the object is never rebuilt.
        admin.execute(f"ALTER ROLE {role} PASSWORD 'second'")
        (directory / "password").write_text("second")

        # Existing pooled sessions stay authenticated after a password change,
        # so they must be terminated for this to test anything at all — without
        # this the pool keeps serving on the old connections and an
        # implementation that read the credential once at startup would pass.
        admin.execute(
            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
            "WHERE usename = %s AND pid <> pg_backend_pid()", (role,)
        )
        backend.pool.check()          # discard the dead connections and reopen

        event2 = make_event("rot-2")
        assert backend.accept(event2, digest(event2)).duplicate is False
        assert backend.get("rot-1")["event_id"] == "rot-1"
    finally:
        if backend:
            with backend.pool.connection() as conn:
                conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
            backend.close()
        admin.execute(
            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
            "WHERE usename = %s AND pid <> pg_backend_pid()", (role,)
        )
        admin.execute(f"DROP ROLE IF EXISTS {role}")
        admin.close()


@pg_only
def test_a_stale_mounted_credential_fails_closed(tmp_path):
    """A revoked lease must surface as unavailable, not as silent success."""
    import psycopg

    from audit_core.postgres_backend import PostgresAuditBackend

    role = f"lease_{uuid.uuid4().hex[:8]}"
    admin = psycopg.connect(PG_URL, autocommit=True)
    directory = tmp_path / "db"
    directory.mkdir()
    try:
        admin.execute(f"CREATE ROLE {role} LOGIN PASSWORD 'only' SUPERUSER")
        _write_credentials(directory, user=role, password="wrong-password", url=PG_URL)
        with pytest.raises((BackendUnavailableError, Exception)):
            PostgresAuditBackend("", credential_dir=str(directory), migrate=True)
    finally:
        admin.execute(f"DROP ROLE IF EXISTS {role}")
        admin.close()