audit-core/tests/test_ingestion.py

import io
import json
from datetime import datetime, timezone

import pytest

from audit_core.ingestion import IngestionApplication
from audit_core.interface import (
    BackendUnavailableError,
    RetentionPolicy,
    custody_class_satisfies,
)
from audit_core.mock_file_backend import MockFileAuditBackend
from audit_core.sqlite_backend import SQLiteAuditBackend


def invoke(app, payload, *, token="opaque", key="evt-1", method="POST",
           path="/v1/events", body=None, length=None):
    raw = body if body is not None else json.dumps(payload).encode()
    environ = {
        "PATH_INFO": path,
        "REQUEST_METHOD": method,
        "CONTENT_LENGTH": str(len(raw)) if length is None else length,
        "wsgi.input": io.BytesIO(raw),
        "HTTP_AUTHORIZATION": f"Bearer {token}",
    }
    if key is not None:
        environ["HTTP_IDEMPOTENCY_KEY"] = key
    result = {}
    out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
    return result["status"], (json.loads(out) if out else {})


def event(**overrides):
    base = {
        "id": "evt-1",
        "type": "membership.added",
        "source": "user-engine",
        "subject": "membership-1",
        "tenant": "tenant:friendly:binky",
        "correlation_id": "corr-1",
        "occurred_at": "2026-08-09T00:00:00+00:00",
        "data": {"membership_id": "membership-1"},
    }
    base.update(overrides)
    return base


@pytest.fixture
def app(tmp_path):
    return IngestionApplication(SQLiteAuditBackend(str(tmp_path / "events.db")), "opaque")


# --- backend contract (T01) -------------------------------------------------

def test_refuses_a_non_durable_backend():
    """The development file backend must never become the production sink."""
    with pytest.raises(ValueError, match="not durable"):
        IngestionApplication(MockFileAuditBackend(base_dir="/tmp/unused"), "opaque")


def test_readiness_reports_custody_class(app):
    status, body = invoke(app, None, path="/readyz", method="GET", body=b"")
    assert status.startswith("200")
    assert body["durable"] is True
    assert body["custody_class"] == "development"


def test_accepted_events_are_durable_across_reopen(tmp_path):
    path = str(tmp_path / "events.db")
    first = IngestionApplication(SQLiteAuditBackend(path), "opaque")
    assert invoke(first, event())[0].startswith("202")

    reopened = IngestionApplication(SQLiteAuditBackend(path), "opaque")
    status, body = invoke(reopened, event())
    assert status.startswith("200") and body["status"] == "duplicate"


# --- idempotency and conflict (T01, T02) ------------------------------------

def test_accepts_once_and_replays_idempotently(app):
    assert invoke(app, event())[0].startswith("202")
    status, body = invoke(app, event())
    assert status.startswith("200") and body["status"] == "duplicate"


def test_same_id_different_payload_is_a_conflict(app):
    assert invoke(app, event())[0].startswith("202")
    status, body = invoke(app, event(subject="membership-2"))
    assert status.startswith("409")
    assert body["error"] == "event_id_conflict"


# --- authentication (T02) ---------------------------------------------------

def test_rejects_wrong_credential(app):
    assert invoke(app, event(), token="wrong")[0].startswith("401")


def test_non_ascii_credential_is_unauthorized_not_a_crash(app):
    """compare_digest raises TypeError on non-ASCII str; that is a 401."""
    assert invoke(app, event(), token="wröng")[0].startswith("401")


# --- validation (T02, T07) --------------------------------------------------

@pytest.mark.parametrize("payload,expected", [
    (event(source="somewhere-else"), "source_not_allowed"),
    (event(occurred_at="2026-08-09T00:00:00"), "timestamp_missing_timezone"),
    (event(occurred_at="not-a-date"), "invalid_timestamp"),
    (event(tenant=""), "invalid_event"),
])
def test_rejects_bad_events(app, payload, expected):
    status, body = invoke(app, payload)
    assert status.startswith("400")
    assert body["error"] == expected


def test_rejects_mismatched_idempotency_key(app):
    status, body = invoke(app, event(), key="other")
    assert status.startswith("400")
    assert body["error"] == "idempotency_key_mismatch"


def test_rejects_malformed_json(app):
    assert invoke(app, None, body=b"{not json")[0].startswith("400")


def test_rejects_empty_body(app):
    status, body = invoke(app, None, body=b"")
    assert status.startswith("400")
    assert body["error"] == "empty_body"


def test_rejects_oversized_body(app):
    status, body = invoke(app, None, body=b"x", length=str(512 * 1024))
    assert status.startswith("400")
    assert body["error"] == "payload_too_large"


def test_rejects_truncated_body(app):
    status, body = invoke(app, None, body=b"{}", length="500")
    assert status.startswith("400")
    assert body["error"] == "truncated_body"


# --- routing (T07) ----------------------------------------------------------

@pytest.mark.parametrize("path,method", [
    ("/nope", "POST"),
    ("/nope", "GET"),
    ("/v1/events", "DELETE"),
])
def test_unknown_routes_are_not_found(app, path, method):
    assert invoke(app, event(), path=path, method=method)[0].startswith("404")


def test_healthz_needs_no_credential(app):
    status, _ = invoke(app, None, path="/healthz", method="GET", body=b"", token="wrong")
    assert status.startswith("200")


# --- failure handling (T02) -------------------------------------------------

class _BrokenBackend:
    @property
    def retention_policy(self):
        return RetentionPolicy("archive", 3650, True, True, durable=True)

    def emit(self, event):
        raise BackendUnavailableError("down")

    def accept(self, event, payload_hash):
        raise BackendUnavailableError("down")


class _ExplodingBackend(_BrokenBackend):
    def accept(self, event, payload_hash):
        raise RuntimeError("unexpected")


def test_backend_unavailable_is_retryable_503():
    status, body = invoke(IngestionApplication(_BrokenBackend(), "opaque"), event())
    assert status.startswith("503")
    assert body["error"] == "backend_unavailable"


def test_unexpected_backend_error_still_returns_a_response():
    """No request path may terminate without calling start_response."""
    status, body = invoke(IngestionApplication(_ExplodingBackend(), "opaque"), event())
    assert status.startswith("500")
    assert body["error"] == "internal_error"


# --- concurrency (T01) ------------------------------------------------------

def test_concurrent_duplicates_produce_exactly_one_record(tmp_path):
    """Racing submissions of one event: one acceptance, one custody record.

    This is the assertion the whole service rests on, so it is exercised rather
    than assumed. An earlier single-connection implementation passed every
    serial test while letting two callers both be told they were first.
    """
    import threading

    backend = SQLiteAuditBackend(str(tmp_path / "race.db"))
    app = IngestionApplication(backend, "opaque")
    outcomes: list[str] = []
    lock = threading.Lock()
    barrier = threading.Barrier(16)

    def submit():
        barrier.wait()
        status, body = invoke(app, event())
        with lock:
            outcomes.append(body.get("status", status))

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

    assert outcomes.count("accepted") == 1, outcomes
    assert outcomes.count("duplicate") == 15, outcomes
    stored = backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0]
    assert stored == 1


# --- sender identity binding (T03) ------------------------------------------

from audit_core.senders import SenderIdentity, SenderRegistry  # noqa: E402


def bound_app(tmp_path, **kw):
    identity = SenderIdentity(
        name="user-engine",
        tokens=kw.get("tokens", ("opaque",)),
        sources=frozenset(kw.get("sources", {"user-engine"})),
        tenants=frozenset(kw.get("tenants", {"tenant:friendly:binky"})),
        may_read=kw.get("may_read", False),
        secret_policy=kw.get("secret_policy", "redact"),
        expires_at=kw.get("expires_at"),
    )
    # An unrestricted operator sits alongside the scoped sender. The
    # instance-wide read surfaces — stats, dead letters, secret findings,
    # integrity — carry no tenant key and so require full scope
    # (AUDIT-WP-0008-T04); reading them as the scoped sender is what that task
    # made a 403.
    operator = SenderIdentity(
        name="operator",
        tokens=("operator",),
        sources=frozenset({"user-engine"}),
        tenants=frozenset({"*"}),
        may_write=False,
        may_read=True,
    )
    backend = SQLiteAuditBackend(str(tmp_path / "bound.db"))
    return IngestionApplication(backend, SenderRegistry([identity, operator])), backend


def test_credential_may_not_claim_another_tenant(tmp_path):
    """The property WP-0003 recorded as done but never implemented."""
    app, _ = bound_app(tmp_path)
    assert invoke(app, event())[0].startswith("202")
    status, body = invoke(app, event(tenant="tenant:coulomb"))
    assert status.startswith("400")
    assert body["error"] == "tenant_not_allowed"


def test_credential_may_not_claim_another_source(tmp_path):
    app, _ = bound_app(tmp_path)
    status, body = invoke(app, event(source="issue-core"))
    assert status.startswith("400")
    assert body["error"] == "source_not_allowed"


def test_rotation_accepts_both_tokens(tmp_path):
    """Rotation must not need a delivery gap."""
    app, _ = bound_app(tmp_path, tokens=("current", "next"))
    assert invoke(app, event(), token="current")[0].startswith("202")
    assert invoke(app, event(id="evt-2"), key="evt-2", token="next")[0].startswith("202")
    assert invoke(app, event(id="evt-3"), key="evt-3", token="retired")[0].startswith("401")


def test_expired_sender_is_unauthorized_at_http_boundary(tmp_path):
    app, _ = bound_app(
        tmp_path,
        expires_at=datetime(2000, 1, 1, tzinfo=timezone.utc),
    )
    status, body = invoke(app, event())
    assert status.startswith("401")
    assert body["error"] == "unauthorized"


def test_sender_credential_cannot_read_the_trail_back(tmp_path):
    app, _ = bound_app(tmp_path, may_read=False)
    assert invoke(app, event())[0].startswith("202")
    status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
    assert status.startswith("403")
    assert body["error"] == "read_forbidden"


# --- operator read surface (T05) --------------------------------------------

def test_lookup_by_event_id_and_correlation(app):
    assert invoke(app, event())[0].startswith("202")
    assert invoke(app, event(id="evt-2", correlation_id="corr-1"), key="evt-2")[0].startswith("202")

    status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
    assert status.startswith("200")
    assert body["event_id"] == "evt-1"
    assert body["tenant"] == "tenant:friendly:binky"

    status, body = invoke_query(app, "correlation_id=corr-1")
    assert status.startswith("200")
    assert {e["event_id"] for e in body["events"]} == {"evt-1", "evt-2"}


def invoke_query(app, query, token="opaque"):
    environ = {
        "PATH_INFO": "/v1/events",
        "REQUEST_METHOD": "GET",
        "QUERY_STRING": query,
        "CONTENT_LENGTH": "0",
        "wsgi.input": io.BytesIO(b""),
        "HTTP_AUTHORIZATION": f"Bearer {token}",
    }
    result = {}
    out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
    return result["status"], (json.loads(out) if out else {})


def test_unknown_event_id_is_not_found(app):
    status, _ = invoke(app, None, path="/v1/events/nope", method="GET", body=b"")
    assert status.startswith("404")


def test_correlation_lookup_requires_a_correlation_id(app):
    status, body = invoke_query(app, "")
    assert status.startswith("400")
    assert body["error"] == "correlation_id_required"


def test_rejected_events_appear_as_dead_letters(app):
    assert invoke(app, event(source="issue-core"))[0].startswith("400")
    status, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"")
    assert status.startswith("200")
    entry = body["dead_letters"][0]
    assert entry["reason"] == "source_not_allowed"
    assert entry["event_id"] == "evt-1"
    assert entry["payload"] is not None


def test_secret_rejection_withholds_the_payload(tmp_path):
    """Storing the body of an event rejected for carrying secret-shaped
    material would write that material into the audit store."""
    app, _ = bound_app(tmp_path, secret_policy="reject", may_read=True)
    assert invoke(app, event(data={"password": "hunter2"}))[0].startswith("400")
    _, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"", token="operator")
    entry = body["dead_letters"][0]
    assert entry["reason"] == "secret_shaped_field"
    assert entry["payload_withheld"] is True
    assert entry["payload"] is None
    assert entry["payload_hash"]


# --- redaction policy (T04) -------------------------------------------------

def test_default_policy_redacts_and_accepts(app):
    """Default is redact: losing the whole audit record over one field is
    worse than storing it with that field masked."""
    status, _ = invoke(app, event(data={"membership_id": "m-1", "auth_token": "s3cret"}))
    assert status.startswith("202")

    _, record = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
    assert record["details"]["data"]["auth_token"] == "[redacted]"
    assert record["details"]["data"]["membership_id"] == "m-1"
    # The record must admit it was modified.
    assert record["details"]["redaction"]["policy"] == "redact"
    assert record["details"]["redaction"]["paths"] == ["data.auth_token"]


def test_reject_policy_is_available_per_sender(tmp_path):
    app, _ = bound_app(tmp_path, secret_policy="reject")
    status, body = invoke(app, event(data={"password": "x"}))
    assert status.startswith("400")
    assert body["error"] == "secret_shaped_field"


def test_nested_and_listed_secrets_are_redacted(app):
    payload = event(data={"items": [{"api_secret": "a"}, {"ok": 1}], "n": {"private_key": "k"}})
    assert invoke(app, payload)[0].startswith("202")
    _, record = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
    data = record["details"]["data"]
    assert data["items"][0]["api_secret"] == "[redacted]"
    assert data["items"][1]["ok"] == 1
    assert data["n"]["private_key"] == "[redacted]"
    assert set(record["details"]["redaction"]["paths"]) == {
        "data.items[0].api_secret", "data.n.private_key",
    }


def test_findings_are_counted_by_path_for_both_outcomes(tmp_path):
    """Counters name the field to fix, not just a total."""
    app, backend = bound_app(tmp_path, may_read=True)
    for i in range(3):
        invoke(app, event(id=f"e{i}", data={"auth_token": "x"}), key=f"e{i}")

    strict, _ = bound_app(tmp_path, secret_policy="reject")
    invoke(strict, event(id="r1", data={"auth_token": "x"}), key="r1")

    _, body = invoke(app, None, path="/v1/secret-findings", method="GET", body=b"", token="operator")
    rows = {(r["outcome"], r["field_path"]): r for r in body["secret_findings"]}
    assert rows[("redacted", "data.auth_token")]["occurrences"] == 3
    assert rows[("redacted", "data.auth_token")]["action"] == "membership.added"
    assert rows[("redacted", "data.auth_token")]["source"] == "user-engine"
    assert rows[("redacted", "data.auth_token")]["persisted"] is True
    assert rows[("rejected", "data.auth_token")]["occurrences"] == 1


def test_counters_survive_restart(tmp_path):
    """The counters drive a fix in the sending service; that work outlives a
    pod restart, so they are durable rather than in-memory."""
    path = str(tmp_path / "counters.db")
    first = IngestionApplication(SQLiteAuditBackend(path), "opaque")
    invoke(first, event(data={"auth_token": "x"}))

    reopened = IngestionApplication(SQLiteAuditBackend(path), "opaque")
    _, body = invoke(reopened, None, path="/v1/secret-findings", method="GET", body=b"")
    assert body["secret_findings"][0]["occurrences"] == 1


# --- deployment guards and counters (WP-0005-T03) ---------------------------

def test_required_custody_class_refuses_a_development_backend(tmp_path):
    """Losing AUDIT_CORE_DATABASE_URL must fail to start, not silently
    downgrade custody to the development store."""
    backend = SQLiteAuditBackend(str(tmp_path / "dev.db"))
    with pytest.raises(ValueError, match="does not meet the required"):
        IngestionApplication(backend, "opaque", require_custody_class="operational")
    with pytest.raises(ValueError, match="does not meet the required"):
        IngestionApplication(backend, "opaque", require_custody_class="archive")


def test_operational_and_archive_alias_for_one_deploy():
    """A mixed rollout must start: new backend + old require, and the reverse."""

    class _Operational(_BrokenBackend):
        @property
        def retention_policy(self):
            return RetentionPolicy(
                custody_class="operational",
                retention_days=None,
                immutable=True,
                tamper_evidence=False,
                durable=True,
                recoverable_days=30,
                recoverable_source="cited",
                recoverable_basis="measured",
            )

    IngestionApplication(_Operational(), "opaque", require_custody_class="archive")
    IngestionApplication(_Operational(), "opaque", require_custody_class="operational")


def test_custody_class_alias_is_not_development():
    assert custody_class_satisfies("operational", "archive")
    assert custody_class_satisfies("archive", "operational")
    assert not custody_class_satisfies("development", "operational")
    assert not custody_class_satisfies("development", "archive")
    assert custody_class_satisfies("development", "development")


def test_readiness_reports_recovery_fields_for_operational_backend():
    class _Operational(_BrokenBackend):
        @property
        def retention_policy(self):
            return RetentionPolicy(
                custody_class="operational",
                retention_days=None,
                immutable=True,
                tamper_evidence=False,
                durable=True,
                recoverable_days=30,
                recoverable_source="resource-control/data/capability/platform-audit-storage.json",
                recoverable_basis="measured",
            )

        def health(self):
            return None

    status, body = invoke(
        IngestionApplication(_Operational(), "opaque"),
        None, path="/readyz", method="GET", body=b"",
    )
    assert status.startswith("200")
    assert body["custody_class"] == "operational"
    assert body["durable"] is True
    assert body["recoverable_days"] == 30
    assert body["recoverable_basis"] == "measured"
    assert "platform-audit-storage" in body["recoverable_source"]


def test_counters_track_each_outcome(tmp_path):
    app, _ = bound_app(tmp_path, may_read=True)
    invoke(app, event())                                   # accepted
    invoke(app, event())                                   # duplicate
    invoke(app, event(subject="other"))                    # conflict
    invoke(app, event(id="e2", tenant="tenant:coulomb"), key="e2")  # rejected
    invoke(app, event(), token="nope")                     # unauthorized

    _, body = invoke(app, None, path="/v1/stats", method="GET", body=b"", token="operator")
    counts = body["counts"]
    assert counts["accepted"] == 1
    assert counts["duplicate"] == 1
    assert counts["conflict"] == 1
    assert counts["rejected"] == 1
    assert counts["unauthorized"] == 1
    assert body["since"]


def test_stats_require_the_read_privilege(tmp_path):
    app, _ = bound_app(tmp_path, may_read=False)
    status, _ = invoke(app, None, path="/v1/stats", method="GET", body=b"")
    assert status.startswith("403")


# --- read is tenant-scoped (AUDIT-WP-0008-T04) ------------------------------

def two_tenant_app(tmp_path):
    """A seeded store, plus a reader scoped to one of the two tenants.

    The writer is unrestricted so both tenants exist; the reader is bound to
    ``binky`` only. Both applications share one backend, which is the point —
    the boundary has to hold in the read path, not in the store.
    """
    backend = SQLiteAuditBackend(str(tmp_path / "scoped.db"))
    writer = SenderIdentity(
        name="seeder", tokens=("seed",), sources=frozenset({"user-engine"}),
        tenants=frozenset({"*"}), may_read=True,
    )
    reader = SenderIdentity(
        name="scoped-reader", tokens=("scoped",), sources=frozenset({"user-engine"}),
        tenants=frozenset({"tenant:friendly:binky"}), may_read=True,
    )
    seed = IngestionApplication(backend, SenderRegistry([writer]))
    assert invoke(seed, event(), token="seed")[0].startswith("202")
    assert invoke(
        seed, event(id="evt-2", tenant="tenant:coulomb"), key="evt-2", token="seed"
    )[0].startswith("202")
    return IngestionApplication(backend, SenderRegistry([reader])), seed


def test_scoped_reader_cannot_fetch_another_tenants_event(tmp_path):
    """The defect AUDIT-WP-0008 found: may_read was the only gate."""
    scoped, _ = two_tenant_app(tmp_path)

    status, body = invoke(scoped, None, path="/v1/events/evt-1", method="GET",
                          body=b"", token="scoped")
    assert status.startswith("200")
    assert body["tenant"] == "tenant:friendly:binky"

    status, body = invoke(scoped, None, path="/v1/events/evt-2", method="GET",
                          body=b"", token="scoped")
    assert status.startswith("404")
    assert body["error"] == "not_found"


def test_cross_tenant_refusal_is_indistinguishable_from_absence(tmp_path):
    """403 here would confirm the event exists and is someone else's."""
    scoped, _ = two_tenant_app(tmp_path)
    present = invoke(scoped, None, path="/v1/events/evt-2", method="GET",
                     body=b"", token="scoped")
    absent = invoke(scoped, None, path="/v1/events/evt-nope", method="GET",
                    body=b"", token="scoped")
    assert present == absent


def test_correlation_lookup_is_filtered_not_refused(tmp_path):
    """One correlation legitimately spans tenants; serve the readable slice."""
    scoped, seed = two_tenant_app(tmp_path)

    status, body = invoke_query(scoped, "correlation_id=corr-1", token="scoped")
    assert status.startswith("200")
    assert {e["event_id"] for e in body["events"]} == {"evt-1"}

    status, body = invoke_query(seed, "correlation_id=corr-1", token="seed")
    assert {e["event_id"] for e in body["events"]} == {"evt-1", "evt-2"}


@pytest.mark.parametrize(
    "path", ["/v1/dead-letters", "/v1/stats", "/v1/secret-findings", "/v1/integrity"]
)
def test_unscoped_surfaces_require_full_tenant_scope(tmp_path, path):
    """Not tenant-keyed, so they cannot be filtered — refuse instead."""
    scoped, seed = two_tenant_app(tmp_path)

    status, body = invoke(scoped, None, path=path, method="GET", body=b"",
                          token="scoped", key=None)
    assert status.startswith("403")
    assert body["error"] == "full_tenant_scope_required"

    status, _ = invoke(seed, None, path=path, method="GET", body=b"",
                       token="seed", key=None)
    assert status.startswith("200")