diff --git a/Containerfile b/Containerfile index 447613f..2d4a10e 100644 --- a/Containerfile +++ b/Containerfile @@ -3,7 +3,7 @@ RUN useradd --system --uid 10001 --create-home audit-core WORKDIR /app COPY pyproject.toml README.md LICENSE ./ COPY audit_core ./audit_core -RUN pip install --no-cache-dir ".[serve]" +RUN pip install --no-cache-dir ".[serve,postgres]" USER 10001 EXPOSE 8080 CMD ["audit-core-ingest"] diff --git a/Makefile b/Makefile index 08ff44c..d63504a 100644 --- a/Makefile +++ b/Makefile @@ -21,4 +21,22 @@ help: ## Show this help @awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n make \033[36m\033[0m\n"} \ /^[a-zA-Z_-]+:.*?##/ { printf " \033[36m%-24s\033[0m %s\n", $$1, $$2 }' $(MAKEFILE_LIST) -.PHONY: test mock-audit-smoke mock-audit-cleanup help +.PHONY: test test-pg pg-test-up pg-test-down mock-audit-smoke mock-audit-cleanup help + +PG_TEST_CONTAINER ?= ac-pg-test +PG_TEST_PORT ?= 55445 +PG_TEST_URL ?= postgresql://postgres:test@127.0.0.1:$(PG_TEST_PORT)/audit_core + +pg-test-up: ## Start a throwaway PostgreSQL for backend conformance tests + -docker rm -f $(PG_TEST_CONTAINER) 2>/dev/null + docker run -d --name $(PG_TEST_CONTAINER) -e POSTGRES_PASSWORD=test \ + -e POSTGRES_DB=audit_core -p 127.0.0.1:$(PG_TEST_PORT):5432 postgres:16-alpine + @for i in $$(seq 1 40); do \ + docker exec $(PG_TEST_CONTAINER) pg_isready -U postgres >/dev/null 2>&1 && exit 0; \ + sleep 1; done; echo "postgres did not become ready" >&2; exit 1 + +pg-test-down: ## Remove the throwaway PostgreSQL + -docker rm -f $(PG_TEST_CONTAINER) + +test-pg: ## Run the suite including PostgreSQL conformance (needs pg-test-up) + AUDIT_CORE_TEST_DATABASE_URL="$(PG_TEST_URL)" python3 -m pytest -q diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 6085aeb..c50a5ea 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -11,7 +11,7 @@ | workplan | AUDIT-WP-0001 | finished | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md | | workplan | AUDIT-WP-0002 | finished | — | workplans/AUDIT-WP-0002-pluggable-audit-backend.md | | workplan | AUDIT-WP-0003 | finished | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | -| workplan | AUDIT-WP-0004 | proposed | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| workplan | AUDIT-WP-0004 | finished | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | workplan | AUDIT-WP-0005 | proposed | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0001-T01 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md | | task | AUDIT-WP-0001-T02 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md | @@ -24,7 +24,7 @@ | task | AUDIT-WP-0004-T01 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T02 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T03 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | -| task | AUDIT-WP-0004-T04 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| task | AUDIT-WP-0004-T04 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T05 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T06 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T07 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index cc14e28..e51343f 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -414,16 +414,44 @@ def _serve_fallback(app, host: str, port: int, timeout: int) -> None: server.serve_forever() +def build_backend() -> IdempotentAuditBackend: + """Select the custody backend from the environment. + + ``AUDIT_CORE_DATABASE_URL`` selects PostgreSQL (production custody). + Falling back to SQLite is explicit and logged, so a deployment that meant + to use Postgres and lost its URL is visible rather than quietly running on + the wrong store. + """ + url = os.environ.get("AUDIT_CORE_DATABASE_URL") + if url: + from audit_core.postgres_backend import PostgresAuditBackend + + retention = os.environ.get("AUDIT_CORE_RETENTION_DAYS") + log.info("custody backend: postgresql") + return PostgresAuditBackend( + url, + schema=os.environ.get("AUDIT_CORE_DATABASE_SCHEMA", "audit_core"), + retention_days=int(retention) if retention else None, + max_size=int(os.environ.get("AUDIT_CORE_DB_POOL_MAX", "8")), + statement_timeout_ms=int( + os.environ.get("AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS", "30000") + ), + ) + path = os.environ.get("AUDIT_CORE_DATABASE_PATH", "/data/audit-core.db") + log.warning( + "custody backend: sqlite at %s — AUDIT_CORE_DATABASE_URL is unset, so this " + "is not the production store", path, + ) + return SQLiteAuditBackend(path) + + def main() -> None: logging.basicConfig( level=os.environ.get("AUDIT_CORE_LOG_LEVEL", "INFO"), format='{"ts":"%(asctime)s","level":"%(levelname)s","logger":"%(name)s","msg":"%(message)s"}', stream=sys.stdout, ) - backend = SQLiteAuditBackend( - os.environ.get("AUDIT_CORE_DATABASE_PATH", "/data/audit-core.db") - ) - app = IngestionApplication(backend, SenderRegistry.from_env()) + app = IngestionApplication(build_backend(), SenderRegistry.from_env()) serve( app, host=os.environ.get("AUDIT_CORE_HOST", "0.0.0.0"), @@ -431,3 +459,7 @@ def main() -> None: threads=int(os.environ.get("AUDIT_CORE_THREADS", "8")), timeout=int(os.environ.get("AUDIT_CORE_REQUEST_TIMEOUT", "30")), ) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/audit_core/postgres_backend.py b/audit_core/postgres_backend.py new file mode 100644 index 0000000..cf79fb7 --- /dev/null +++ b/audit_core/postgres_backend.py @@ -0,0 +1,426 @@ +"""Durable PostgreSQL audit backend (AUDIT-WP-0005-T01). + +Production custody. Runs on the Railiance shared PostgreSQL platform +(`rapp-postgres`), inside the database-per-consumer boundary fixed by that +repo's ADR-0001. + +Two properties are pushed into the database rather than the application: + +* **Idempotency** — ``INSERT ... ON CONFLICT DO NOTHING RETURNING`` resolves + insert-or-detect in one statement, so two concurrent submissions of the same + event cannot both be told they were first. The SQLite backend learned this + the hard way; see its test. +* **Append-only custody** — a trigger rejects ``UPDATE`` and ``DELETE`` on the + events table, so a leaked runtime credential can add records but cannot + rewrite or erase existing ones. This is what lets ``RetentionPolicy`` declare + ``immutable=True`` honestly; see :meth:`retention_policy` for its limits. +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from typing import Any + +from audit_core.interface import ( + AcceptResult, + AuditEvent, + BackendUnavailableError, + EventConflictError, + EventValidationError, + RetentionPolicy, + validate_event, +) +from audit_core.redaction import Finding + +try: # pragma: no cover - import guard + import psycopg + from psycopg_pool import ConnectionPool +except ImportError as exc: # pragma: no cover + raise ImportError( + "the postgres backend needs psycopg: pip install 'audit-core[postgres]'" + ) from exc + +DEFAULT_SCHEMA = "audit_core" + +# Ordered, append-only. Each entry runs once and is recorded in +# schema_migrations. Never edit a released migration — add a new one. +MIGRATIONS: list[tuple[str, str]] = [ + ( + "0001-events", + """ + CREATE TABLE IF NOT EXISTS {schema}.events ( + event_id text PRIMARY KEY, + payload_hash text NOT NULL, + accepted_at timestamptz NOT NULL DEFAULT now(), + observed_at timestamptz, + tenant text NOT NULL, + correlation_id text, + source text NOT NULL, + action text NOT NULL, + record jsonb NOT NULL + ); + CREATE INDEX IF NOT EXISTS events_correlation_idx + ON {schema}.events (correlation_id); + -- Tenant keying is mandatory per business-app-service-contract 1.3; + -- the index makes per-tenant export and retention tractable. + CREATE INDEX IF NOT EXISTS events_tenant_idx ON {schema}.events (tenant); + """, + ), + ( + "0002-append-only", + """ + CREATE OR REPLACE FUNCTION {schema}.reject_mutation() RETURNS trigger AS $fn$ + BEGIN + RAISE EXCEPTION 'audit events are append-only (attempted %)', TG_OP + USING ERRCODE = 'restrict_violation'; + END; + $fn$ LANGUAGE plpgsql; + + DROP TRIGGER IF EXISTS events_append_only ON {schema}.events; + CREATE TRIGGER events_append_only + BEFORE UPDATE OR DELETE ON {schema}.events + FOR EACH ROW EXECUTE FUNCTION {schema}.reject_mutation(); + """, + ), + ( + "0003-dead-letters", + """ + CREATE TABLE IF NOT EXISTS {schema}.dead_letters ( + id bigserial PRIMARY KEY, + event_id text, + received_at timestamptz NOT NULL DEFAULT now(), + sender text, + reason text NOT NULL, + payload_hash text NOT NULL, + payload text, + payload_withheld boolean NOT NULL DEFAULT false + ); + CREATE INDEX IF NOT EXISTS dead_letters_event_idx + ON {schema}.dead_letters (event_id); + """, + ), + ( + "0004-secret-findings", + """ + CREATE TABLE IF NOT EXISTS {schema}.secret_findings ( + sender text NOT NULL, + source text NOT NULL, + action text NOT NULL, + field_path text NOT NULL, + outcome text NOT NULL, + persisted boolean NOT NULL DEFAULT false, + occurrences bigint NOT NULL DEFAULT 0, + first_seen timestamptz NOT NULL DEFAULT now(), + last_seen timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (sender, source, action, field_path, outcome) + ); + """, + ), +] + +# Rejection reasons whose payload must never be persisted — storing the body of +# an event rejected *for carrying secret-shaped material* would write that +# material into the audit store. +WITHHOLD_PAYLOAD_REASONS = frozenset({"secret_shaped_field"}) + + +class PostgresAuditBackend: + """Audit custody in PostgreSQL.""" + + def __init__( + self, + dsn: str | None = None, + *, + schema: str = DEFAULT_SCHEMA, + retention_days: int | None = None, + min_size: int = 1, + max_size: int = 8, + statement_timeout_ms: int = 30_000, + migrate: bool = True, + ) -> None: + self.dsn = dsn or os.environ.get("AUDIT_CORE_DATABASE_URL") or "" + if not self.dsn: + raise ValueError("a DSN is required (AUDIT_CORE_DATABASE_URL)") + if not schema.isidentifier(): + raise ValueError(f"unsafe schema name: {schema!r}") + self.schema = schema + self.retention_days = retention_days + try: + self.pool = ConnectionPool( + self.dsn, + min_size=min_size, + max_size=max_size, + kwargs={ + "autocommit": True, + # A stalled write must surface as unavailable rather than + # hold a request open indefinitely. + "options": f"-c statement_timeout={int(statement_timeout_ms)}", + }, + open=True, + timeout=10, + ) + except Exception as exc: # psycopg raises a wide family here + raise BackendUnavailableError(f"cannot connect: {exc}") from exc + if migrate: + self.migrate() + + # --- schema ------------------------------------------------------------ + + def migrate(self) -> list[str]: + """Apply pending migrations. Returns the ids applied this call.""" + applied: list[str] = [] + try: + with self.pool.connection() as conn: + conn.execute(f'CREATE SCHEMA IF NOT EXISTS "{self.schema}"') + conn.execute( + f'CREATE TABLE IF NOT EXISTS "{self.schema}".schema_migrations (' + " id text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())" + ) + done = { + row[0] + for row in conn.execute( + f'SELECT id FROM "{self.schema}".schema_migrations' + ).fetchall() + } + for migration_id, body in MIGRATIONS: + if migration_id in done: + continue + conn.execute(body.format(schema=f'"{self.schema}"')) + conn.execute( + f'INSERT INTO "{self.schema}".schema_migrations (id) VALUES (%s)', + (migration_id,), + ) + applied.append(migration_id) + except psycopg.Error as exc: + raise BackendUnavailableError(f"migration failed: {exc}") from exc + return applied + + @property + def _events(self) -> str: + return f'"{self.schema}".events' + + # --- contract ---------------------------------------------------------- + + @property + def retention_policy(self) -> RetentionPolicy: + """Declared custody guarantees. + + ``immutable`` is True because migration 0002 installs a trigger that + rejects UPDATE and DELETE, so no consumer credential can alter a stored + record. It is not a claim against the database owner or a superuser, + who can drop the trigger; ``tamper_evidence`` is correspondingly False, + because nothing here would *prove* they had. Hash-chaining or external + anchoring would be needed for that, and is not implemented. + """ + return RetentionPolicy( + custody_class="archive", + retention_days=self.retention_days, + immutable=True, + tamper_evidence=False, + durable=True, + ) + + def emit(self, event: AuditEvent) -> str: + return self.accept(event, payload_hash=_record_hash(event)).reference + + def accept(self, event: AuditEvent, payload_hash: str) -> AcceptResult: + validate_event(event) + reference = f"audit:{event.event_id}" + details = event.details if isinstance(event.details, dict) else {} + record = event.as_record() + try: + with self.pool.connection() as conn: + inserted = conn.execute( + f""" + INSERT INTO {self._events} + (event_id, payload_hash, observed_at, tenant, correlation_id, + source, action, record) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (event_id) DO NOTHING + RETURNING event_id + """, + ( + event.event_id, + payload_hash, + _timestamp(event.observed_at), + event.tenant, + str(details.get("correlation_id") or "") or None, + event.source, + event.action, + json.dumps(record, sort_keys=True), + ), + ).fetchone() + if inserted is not None: + return AcceptResult(duplicate=False, reference=reference) + existing = conn.execute( + f"SELECT payload_hash FROM {self._events} WHERE event_id = %s", + (event.event_id,), + ).fetchone() + except psycopg.Error as exc: + raise BackendUnavailableError(str(exc)) from exc + + if existing is None: + raise BackendUnavailableError("event disappeared during accept") + if existing[0] != payload_hash: + raise EventConflictError( + f"event_id {event.event_id} already held with a different payload" + ) + return AcceptResult(duplicate=True, reference=reference) + + # --- operator surface -------------------------------------------------- + + def get(self, event_id: str) -> dict | None: + rows = self._query( + f"SELECT record, accepted_at FROM {self._events} WHERE event_id = %s", + (event_id,), + ) + if not rows: + return None + return {"accepted_at": _iso(rows[0][1]), **rows[0][0]} + + def by_correlation(self, correlation_id: str, limit: int = 100) -> list[dict]: + rows = self._query( + f"SELECT record, accepted_at FROM {self._events} " + "WHERE correlation_id = %s ORDER BY accepted_at, event_id LIMIT %s", + (correlation_id, int(limit)), + ) + return [{"accepted_at": _iso(at), **rec} for rec, at in rows] + + def replay(self, event_id: str) -> AcceptResult: + """Re-submit a stored event through :meth:`accept`. + + Reconciliation, not re-creation: replaying an event already in custody + must return ``duplicate=True`` against the same record. If it were to + report a first acceptance, the store would be producing a second + custody record for one source event — the exact failure the whole + idempotency design exists to prevent. + """ + rows = self._query( + f"SELECT record, payload_hash FROM {self._events} WHERE event_id = %s", + (event_id,), + ) + if not rows: + raise KeyError(event_id) + record, payload_hash = rows[0] + return self.accept(_event_from_record(record), payload_hash) + + def record_rejection( + self, + *, + event_id: str | None, + reason: str, + payload_hash: str, + sender: str | None = None, + payload: str | None = None, + ) -> None: + withheld = reason in WITHHOLD_PAYLOAD_REASONS + self._execute( + f'INSERT INTO "{self.schema}".dead_letters ' + "(event_id, sender, reason, payload_hash, payload, payload_withheld) " + "VALUES (%s, %s, %s, %s, %s, %s)", + (event_id, sender, reason, payload_hash, + None if withheld else payload, withheld), + ) + + def dead_letters(self, limit: int = 100) -> list[dict]: + rows = self._query( + "SELECT event_id, received_at, sender, reason, payload_hash, payload, " + f'payload_withheld FROM "{self.schema}".dead_letters ' + "ORDER BY id DESC LIMIT %s", + (int(limit),), + ) + return [ + { + "event_id": r[0], "received_at": _iso(r[1]), "sender": r[2], + "reason": r[3], "payload_hash": r[4], "payload": r[5], + "payload_withheld": bool(r[6]), + } + for r in rows + ] + + def count_secret_findings( + self, *, sender: str, source: str, action: str, outcome: str, findings + ) -> None: + for finding in findings: + self._execute( + f'INSERT INTO "{self.schema}".secret_findings ' + "(sender, source, action, field_path, outcome, persisted, occurrences) " + "VALUES (%s, %s, %s, %s, %s, %s, 1) " + "ON CONFLICT (sender, source, action, field_path, outcome) DO UPDATE " + "SET occurrences = secret_findings.occurrences + 1, last_seen = now()", + (sender, source, action, finding.path, outcome, + bool(getattr(finding, "in_persisted_data", False))), + ) + + def secret_findings(self, limit: int = 100) -> list[dict]: + rows = self._query( + "SELECT sender, source, action, field_path, outcome, persisted, " + f'occurrences, first_seen, last_seen FROM "{self.schema}".secret_findings ' + "ORDER BY occurrences DESC, last_seen DESC LIMIT %s", + (int(limit),), + ) + return [ + { + "sender": r[0], "source": r[1], "action": r[2], "field_path": r[3], + "outcome": r[4], "persisted": bool(r[5]), "occurrences": r[6], + "first_seen": _iso(r[7]), "last_seen": _iso(r[8]), + } + for r in rows + ] + + # --- lifecycle --------------------------------------------------------- + + def health(self) -> None: + self._query("SELECT 1", ()) + + def close(self) -> None: + self.pool.close() + + def _query(self, sql: str, params: tuple) -> list: + try: + with self.pool.connection() as conn: + return conn.execute(sql, params).fetchall() + except psycopg.Error as exc: + raise BackendUnavailableError(str(exc)) from exc + + def _execute(self, sql: str, params: tuple) -> None: + try: + with self.pool.connection() as conn: + conn.execute(sql, params) + except psycopg.Error as exc: + raise BackendUnavailableError(str(exc)) from exc + + +def _timestamp(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + + +def _iso(value: Any) -> str | None: + if isinstance(value, datetime): + return value.astimezone(timezone.utc).replace(microsecond=0).isoformat() + return value + + +def _event_from_record(record: dict) -> AuditEvent: + return AuditEvent( + source=record["source"], action=record["action"], resource=record["resource"], + outcome=record["outcome"], tenant=record["tenant"], scope=record["scope"], + actor=record.get("actor"), reason=record.get("reason"), + details=record.get("details") or {}, event_id=record["event_id"], + observed_at=record["observed_at"], schema_version=record["schema_version"], + ) + + +def _record_hash(event: AuditEvent) -> str: + import hashlib + + return hashlib.sha256( + json.dumps(event.as_record(), sort_keys=True).encode("utf-8") + ).hexdigest() diff --git a/docs/audit-backend-contract.md b/docs/audit-backend-contract.md index 8abd4bd..5e88fcc 100644 --- a/docs/audit-backend-contract.md +++ b/docs/audit-backend-contract.md @@ -296,6 +296,12 @@ Read them at `GET /v1/secret-findings` (requires the read privilege): "first_seen": "...", "last_seen": "..."}]} ``` +`occurrences` counts **transmissions, not stored events**: a retry resubmitting +the same secret-shaped field increments it again, even though the event +reconciles as a duplicate and produces no second custody record. That is +deliberate — the number measures how often the sender emitted the field, which +is the behaviour being optimized away. + `persisted` distinguishes a field that reached the stored record from one that sat elsewhere in the envelope and was dropped by normalization anyway. A healthy sender trends to zero occurrences; a non-empty list is a backlog item diff --git a/pyproject.toml b/pyproject.toml index 7b7ea47..f95c5fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,8 @@ dev = ["pytest"] # Production serving. Without this the entrypoint falls back to a threaded # wsgiref server, which is bounded but not a production server (AUDIT-WP-0004-T06). serve = ["waitress>=3.0"] +# Production custody (AUDIT-WP-0005). SQLite remains the development backend. +postgres = ["psycopg[binary,pool]>=3.2"] [project.scripts] audit-core-ingest = "audit_core.ingestion:main" diff --git a/tests/test_backend_conformance.py b/tests/test_backend_conformance.py new file mode 100644 index 0000000..ae5ead7 --- /dev/null +++ b/tests/test_backend_conformance.py @@ -0,0 +1,278 @@ +"""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", "archive", "hot_search") + # 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() + + +# --- postgres-specific guarantees ------------------------------------------- + +pg_only = pytest.mark.skipif(not HAVE_PG, reason="needs PostgreSQL") + + +@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() diff --git a/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md b/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md index 73dd0b4..589db08 100644 --- a/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md +++ b/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md @@ -49,7 +49,7 @@ requirement and consumes it; it does not implement it here. ```task id: AUDIT-WP-0005-T01 -status: todo +status: done priority: high state_hub_task_id: "b1601d0b-922a-40f7-92c0-ea06af6c4468" ``` @@ -76,6 +76,41 @@ Done when the backend passes the same contract tests as the existing backends, concurrent duplicate submissions produce exactly one record, and a database restart mid-write does not produce an acknowledged-but-absent event. +Done 2026-08-10, built and verified against PostgreSQL 16 locally in Docker — +the Railiance cluster was not needed for any of it. + +`tests/test_backend_conformance.py` is a single suite run against every +backend, so "the Postgres backend is done" means it satisfies the same +contract SQLite already does rather than having its own tests that happen to +be green. It skips cleanly when no server is reachable; `make pg-test-up` and +`make test-pg` run it. Suite 50 -> 71. + +`RetentionPolicy` declares `immutable=True`, and that is earned: migration +0002 installs a trigger rejecting UPDATE and DELETE on the events table, so a +leaked runtime credential can append but cannot rewrite or erase the trail. +This materially narrows the residual risk ADR-0001 §5 called out — a leaked +credential could previously forge the audit record. `tamper_evidence` stays +False, because nothing here would *prove* a database owner had dropped the +trigger; hash-chaining or external anchoring would be needed and is not +implemented. + +Migrations are ordered, recorded in `schema_migrations`, and idempotent. +Replay reconciles rather than duplicating — the piece deferred out of +WP-0004-T05 — and is tested to leave exactly one custody record. + +Backend selection is by `AUDIT_CORE_DATABASE_URL`; falling back to SQLite logs +a warning, so a deployment that lost its URL is visible rather than quietly +running on the wrong store. End-to-end smoke through waitress against Postgres +confirmed accept, duplicate, cross-tenant refusal, read/write privilege +separation, visible redaction, secret-finding counters, and readiness +reporting `custody_class=archive`. + +Also fixed: `audit_core.ingestion` had no `__main__` guard, so `python -m +audit_core.ingestion` silently did nothing. + +Not covered here: behaviour across a real database failover, which needs the +cluster and belongs to T05. + ## T02 - Provision storage through the platform lane ```task