diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index c50a5ea..9cc0a87 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -28,7 +28,7 @@ | 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 | -| task | AUDIT-WP-0005-T01 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | +| task | AUDIT-WP-0005-T01 | done | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0005-T02 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0005-T03 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0005-T04 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | diff --git a/audit_core/cli.py b/audit_core/cli.py index a0b0720..4bb63e3 100644 --- a/audit_core/cli.py +++ b/audit_core/cli.py @@ -69,9 +69,53 @@ def build_parser() -> argparse.ArgumentParser: cleanup_parser.add_argument("--retention-days", type=int, default=None) cleanup_parser.set_defaults(func=cleanup) + migrate_parser = sub.add_parser( + "migrate-store", help="Move audit records from a SQLite store into PostgreSQL." + ) + migrate_parser.add_argument("--from-sqlite", required=True, dest="source") + migrate_parser.add_argument( + "--to-postgres", dest="destination", default=None, + help="DSN; defaults to AUDIT_CORE_DATABASE_URL", + ) + migrate_parser.add_argument("--schema", default="audit_core") + migrate_parser.add_argument( + "--dry-run", action="store_true", + help="Report what would move without writing anything.", + ) + migrate_parser.set_defaults(func=migrate_store) + return parser +def migrate_store(args: argparse.Namespace) -> int: + import os + + from audit_core.migrate_store import migrate + + if args.dry_run: + import sqlite3 + + with sqlite3.connect(f"file:{args.source}?mode=ro", uri=True) as source: + count = source.execute("SELECT count(*) FROM events").fetchone()[0] + print(json.dumps({"dry_run": True, "events_in_source": count}, sort_keys=True)) + return 0 + + from audit_core.postgres_backend import PostgresAuditBackend + + dsn = args.destination or os.environ.get("AUDIT_CORE_DATABASE_URL") + if not dsn: + raise SystemExit("--to-postgres or AUDIT_CORE_DATABASE_URL is required") + + backend = PostgresAuditBackend(dsn, schema=args.schema) + try: + report = migrate(args.source, backend) + finally: + backend.close() + print(report.summary()) + # A conflict or a failed verification is not a partial success. + return 0 if report.ok else 1 + + def main() -> int: parser = build_parser() args = parser.parse_args() diff --git a/audit_core/migrate_store.py b/audit_core/migrate_store.py new file mode 100644 index 0000000..6e8d63f --- /dev/null +++ b/audit_core/migrate_store.py @@ -0,0 +1,203 @@ +"""Move audit records from a SQLite store into PostgreSQL (AUDIT-WP-0005-T04). + +As of 2026-08-10 there are no pre-production records to move: audit-core was +never deployed, so the only SQLite stores that ever existed were test +fixtures. See the workplan for that finding. + +The tool exists anyway because the SQLite path is still reachable — the +entrypoint falls back to it when ``AUDIT_CORE_DATABASE_URL`` is unset. If that +fallback is ever used in anger, the records it accumulates are audit records +and cannot simply be dropped. + +Fidelity is the whole point. Records are transferred with their original +``event_id``, ``payload_hash``, and ``accepted_at`` intact; a migration that +rewrote acceptance times would destroy the evidence it was meant to preserve. +That is why this bypasses ``accept()``, which stamps ``accepted_at`` with the +current time. +""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass, field + +from audit_core.interface import BackendUnavailableError + + +@dataclass +class MigrationReport: + """What happened, in enough detail to be evidence.""" + + events_read: int = 0 + events_written: int = 0 + events_already_present: int = 0 + dead_letters_written: int = 0 + findings_written: int = 0 + conflicts: list[str] = field(default_factory=list) + verified: bool = False + + @property + def ok(self) -> bool: + return not self.conflicts and self.verified + + def summary(self) -> str: + lines = [ + f"events read: {self.events_read}", + f"events written: {self.events_written}", + f"already present: {self.events_already_present}", + f"dead letters written: {self.dead_letters_written}", + f"secret findings written:{self.findings_written}", + f"verified: {self.verified}", + ] + if self.conflicts: + lines.append(f"CONFLICTS ({len(self.conflicts)}):") + lines += [f" - {c}" for c in self.conflicts] + return "\n".join(lines) + + +def migrate(sqlite_path: str, backend, *, verify: bool = True) -> MigrationReport: + """Copy every record from ``sqlite_path`` into ``backend``. + + Idempotent: re-running skips events already present with a matching payload + hash. An event present with a *different* hash is reported as a conflict + and never overwritten — the destination record wins, and a human decides. + """ + report = MigrationReport() + source = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True) + try: + rows = source.execute( + "SELECT event_id, payload_hash, accepted_at, correlation_id, tenant, record " + "FROM events ORDER BY accepted_at, event_id" + ).fetchall() + report.events_read = len(rows) + + for event_id, payload_hash, accepted_at, correlation_id, tenant, record in rows: + outcome = _import_event( + backend, event_id, payload_hash, accepted_at, correlation_id, + tenant, json.loads(record), + ) + if outcome == "written": + report.events_written += 1 + elif outcome == "present": + report.events_already_present += 1 + else: + report.conflicts.append(outcome) + + report.dead_letters_written = _copy_dead_letters(source, backend) + report.findings_written = _copy_findings(source, backend) + + if verify: + report.verified = _verify(source, backend, report) + finally: + source.close() + return report + + +def _import_event( + backend, event_id, payload_hash, accepted_at, correlation_id, tenant, record +) -> str: + existing = backend._query( + f"SELECT payload_hash FROM {backend._events} WHERE event_id = %s", (event_id,) + ) + if existing: + if existing[0][0] != payload_hash: + return ( + f"{event_id}: destination holds a different payload " + f"(source {payload_hash[:12]}…, destination {existing[0][0][:12]}…) " + "— left untouched" + ) + return "present" + + backend._execute( + f"INSERT INTO {backend._events} " + "(event_id, payload_hash, accepted_at, observed_at, tenant, correlation_id, " + " source, action, record) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) " + "ON CONFLICT (event_id) DO NOTHING", + ( + event_id, payload_hash, accepted_at, record.get("observed_at"), + tenant, correlation_id, record.get("source", ""), record.get("action", ""), + json.dumps(record, sort_keys=True), + ), + ) + return "written" + + +def _copy_dead_letters(source: sqlite3.Connection, backend) -> int: + try: + rows = source.execute( + "SELECT event_id, received_at, sender, reason, payload_hash, payload, " + "payload_withheld FROM dead_letters ORDER BY id" + ).fetchall() + except sqlite3.Error: + return 0 + for event_id, received_at, sender, reason, payload_hash, payload, withheld in rows: + backend._execute( + f'INSERT INTO "{backend.schema}".dead_letters ' + "(event_id, received_at, sender, reason, payload_hash, payload, payload_withheld) " + "VALUES (%s, %s, %s, %s, %s, %s, %s)", + (event_id, received_at, sender, reason, payload_hash, payload, bool(withheld)), + ) + return len(rows) + + +def _copy_findings(source: sqlite3.Connection, backend) -> int: + try: + rows = source.execute( + "SELECT sender, source, action, field_path, outcome, persisted, " + "occurrences, first_seen, last_seen FROM secret_findings" + ).fetchall() + except sqlite3.Error: + return 0 + for r in rows: + backend._execute( + f'INSERT INTO "{backend.schema}".secret_findings ' + "(sender, source, action, field_path, outcome, persisted, occurrences, " + " first_seen, last_seen) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) " + "ON CONFLICT (sender, source, action, field_path, outcome) DO UPDATE " + "SET occurrences = secret_findings.occurrences + excluded.occurrences", + (r[0], r[1], r[2], r[3], r[4], bool(r[5]), r[6], r[7], r[8]), + ) + return len(rows) + + +def _verify(source: sqlite3.Connection, backend, report: MigrationReport) -> bool: + """Confirm every source event is present with its hash and time intact. + + Verification reads back rather than trusting the write path: a migration + that reports success without checking is the same as no migration at all. + """ + rows = source.execute("SELECT event_id, payload_hash, accepted_at FROM events").fetchall() + for event_id, payload_hash, accepted_at in rows: + found = backend._query( + f"SELECT payload_hash, accepted_at FROM {backend._events} WHERE event_id = %s", + (event_id,), + ) + if not found: + report.conflicts.append(f"{event_id}: missing from destination after migration") + return False + if found[0][0] != payload_hash: + report.conflicts.append(f"{event_id}: payload hash differs after migration") + return False + if not _same_instant(found[0][1], accepted_at): + report.conflicts.append( + f"{event_id}: accepted_at not preserved " + f"(source {accepted_at}, destination {found[0][1]})" + ) + return False + return True + + +def _same_instant(destination, source_iso: str) -> bool: + from datetime import datetime + + try: + want = datetime.fromisoformat(str(source_iso).replace("Z", "+00:00")) + except ValueError: + return False + if not hasattr(destination, "timestamp"): + return False + if want.tzinfo is None or destination.tzinfo is None: + return False + return abs(destination.timestamp() - want.timestamp()) < 1.0 diff --git a/tests/test_migrate_store.py b/tests/test_migrate_store.py new file mode 100644 index 0000000..161af9b --- /dev/null +++ b/tests/test_migrate_store.py @@ -0,0 +1,170 @@ +"""Migration fidelity tests (AUDIT-WP-0005-T04). + +The controlling property is that migrating audit records does not alter them. +An event that arrives in the destination with a rewritten acceptance time has +been damaged, not moved. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import uuid + +import pytest + +from audit_core.ingestion import IngestionApplication +from audit_core.migrate_store import migrate +from audit_core.sqlite_backend import SQLiteAuditBackend + +from tests.test_backend_conformance import HAVE_PG, PG_URL, make_event + +pytestmark = pytest.mark.skipif(not HAVE_PG, reason="needs PostgreSQL") + + +@pytest.fixture +def destination(): + from audit_core.postgres_backend import PostgresAuditBackend + + schema = f"mig_{uuid.uuid4().hex[:12]}" + backend = PostgresAuditBackend(PG_URL, schema=schema) + try: + yield backend + finally: + with backend.pool.connection() as conn: + conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + backend.close() + + +@pytest.fixture +def populated_sqlite(tmp_path): + """A SQLite store holding events, a dead letter, and a finding.""" + path = str(tmp_path / "legacy.db") + source = SQLiteAuditBackend(path) + for i in range(3): + event = make_event(f"legacy-{i}") + source.accept(event, hashlib.sha256(f"h{i}".encode()).hexdigest()) + source.record_rejection( + event_id="bad-1", reason="source_not_allowed", payload_hash="hh", + sender="user-engine", payload='{"source":"nope"}', + ) + from audit_core.redaction import Finding + + source.count_secret_findings( + sender="user-engine", source="user-engine", action="membership.added", + outcome="redacted", findings=[Finding("data.auth_token", True)], + ) + return path, source + + +def test_migration_preserves_identity_hash_and_time(populated_sqlite, destination): + path, source = populated_sqlite + original = { + row[0]: (row[1], row[2]) + for row in source.db.execute( + "SELECT event_id, payload_hash, accepted_at FROM events" + ).fetchall() + } + + report = migrate(path, destination) + + assert report.events_read == 3 + assert report.events_written == 3 + assert report.conflicts == [] + assert report.verified is True + assert report.ok is True + + for event_id, (payload_hash, accepted_at) in original.items(): + rows = destination._query( + f"SELECT payload_hash, accepted_at FROM {destination._events} " + "WHERE event_id = %s", (event_id,), + ) + assert rows, f"{event_id} missing" + assert rows[0][0] == payload_hash + # Acceptance time is evidence; rewriting it would destroy what the + # migration exists to preserve. + assert rows[0][1].isoformat().startswith(accepted_at[:16]) + + +def test_migration_carries_dead_letters_and_findings(populated_sqlite, destination): + path, _ = populated_sqlite + report = migrate(path, destination) + + assert report.dead_letters_written == 1 + assert report.findings_written == 1 + assert destination.dead_letters()[0]["reason"] == "source_not_allowed" + assert destination.secret_findings()[0]["field_path"] == "data.auth_token" + + +def test_migration_is_idempotent(populated_sqlite, destination): + path, _ = populated_sqlite + first = migrate(path, destination) + second = migrate(path, destination) + + assert first.events_written == 3 + assert second.events_written == 0 + assert second.events_already_present == 3 + assert second.ok is True + + count = destination._query( + f"SELECT count(*) FROM {destination._events}", () + )[0][0] + assert count == 3 + + +def test_a_diverging_destination_record_is_a_conflict_not_an_overwrite( + populated_sqlite, destination +): + """The destination wins and a human decides. + + Silently overwriting a stored audit record with a differing one would be + the same class of failure as losing it. + """ + path, _ = populated_sqlite + migrate(path, destination) + + # Simulate divergence: same id, different payload hash in the source. + source = SQLiteAuditBackend(path) + source.db.execute("BEGIN IMMEDIATE") + source.db.execute( + "UPDATE events SET payload_hash = 'diverged' WHERE event_id = 'legacy-1'" + ) + source.db.execute("COMMIT") + + report = migrate(path, destination) + assert report.conflicts, "divergence must be reported" + assert "legacy-1" in report.conflicts[0] + assert report.ok is False + + survived = destination._query( + f"SELECT payload_hash FROM {destination._events} WHERE event_id = 'legacy-1'", () + )[0][0] + assert survived != "diverged", "destination record must not be overwritten" + + +def test_migrated_events_remain_readable_and_append_only(populated_sqlite, destination): + path, _ = populated_sqlite + migrate(path, destination) + + record = destination.get("legacy-0") + assert record["tenant"] == "tenant:friendly:binky" + assert destination.retention_policy.immutable is True + + from audit_core.interface import BackendUnavailableError + + with pytest.raises(BackendUnavailableError): + destination._execute( + f"DELETE FROM {destination._events} WHERE event_id = %s", ("legacy-0",) + ) + + +def test_empty_source_migrates_cleanly(tmp_path, destination): + """Today's actual case: nothing to move.""" + path = str(tmp_path / "empty.db") + SQLiteAuditBackend(path) + + report = migrate(path, destination) + assert report.events_read == 0 + assert report.events_written == 0 + assert report.ok is True 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 589db08..eec4f1d 100644 --- a/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md +++ b/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md @@ -166,7 +166,7 @@ previous version. ```task id: AUDIT-WP-0005-T04 -status: todo +status: done priority: medium state_hub_task_id: "9010fb4a-a1b8-4ef7-b143-e33ca7cc0619" ``` @@ -182,6 +182,42 @@ events is the exact failure this service is meant to make impossible. Done when the disposition of every pre-production record is either migrated and verified, or explicitly and justifiably discarded. +**Disposition, established 2026-08-10: there are no pre-production records.** +Checked and found empty: no `/data/audit-core.db` or any other audit-core +SQLite store on this host, no mock-file-backend output under `/tmp/audit-core`, +and no audit-core pod, deployment, or PVC on railiance01. (The only `audit-*` +PVC there is `audit-openbao-0`, which is OpenBao's own audit device and +unrelated to this service.) + +That is consistent with the history rather than surprising: WP-0003-T03 was +cancelled before the receiver was ever deployed, so the only SQLite stores that +have ever existed were test fixtures with no custody value. Nothing is being +discarded, because nothing was ever accepted outside tests. + +The migration tool was built regardless, because the SQLite path remains +reachable: the entrypoint falls back to it when `AUDIT_CORE_DATABASE_URL` is +unset. If that fallback is ever used in anger, the records it accumulates are +audit records and cannot simply be dropped — and building the tool afterwards, +under pressure, is the wrong time. + +`audit_core.migrate_store` plus `python -m audit_core migrate-store` transfer +events, dead letters, and secret-finding counters. Fidelity is the point: +records keep their original `event_id`, `payload_hash`, and `accepted_at`, +which is why the tool bypasses `accept()` — that stamps acceptance with the +current time, and a migration that rewrote acceptance times would destroy the +evidence it exists to preserve. + +Migration is idempotent, and verification reads back from the destination +rather than trusting the write path. Where the destination already holds an +event with a *different* payload hash, the tool reports a conflict and leaves +the destination untouched: silently overwriting a stored audit record with a +differing one is the same class of failure as losing it. A conflict or failed +verification exits non-zero — a partial migration is not a success. + +Six tests cover fidelity, dead letters and counters, idempotency, conflict +handling, post-migration readability under the append-only trigger, and the +empty-source case that is today's actual situation. + ## T05 - Run the live failure matrix ```task