From 146f4e1dd38d4d35d3400aff652e27623773a2da Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 11:36:06 +0200 Subject: [PATCH] fix: repair chained imports and track blocked evidence dependencies Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1046e-465c-7d43-8c4c-d10f0a1feadd --- SCOPE.md | 2 +- WORK-RECORDS.md | 20 +- audit_core/cli.py | 6 +- audit_core/ingestion.py | 15 +- audit_core/integrity.py | 23 +++ audit_core/migrate_store.py | 174 ++++++++++++------ audit_core/postgres_backend.py | 50 ++++- audit_core/sqlite_backend.py | 7 +- docs/cross-repo-evidence-admission.md | 166 +++++++++++++++++ docs/integrity.md | 9 +- docs/operator-runbook.md | 24 +++ docs/stream-completeness.md | 13 ++ intakes/intakes.md | 49 ++--- tests/test_backend_conformance.py | 25 +++ tests/test_integrity.py | 14 ++ tests/test_migrate_store.py | 112 +++++++++++ tests/test_stream_findings.py | 25 +++ .../AUDIT-WP-0012-chained-sqlite-import.md | 65 +++++++ ...DIT-WP-0013-cross-repo-sender-admission.md | 79 ++++++++ ...T-WP-0014-execution-attribution-receipt.md | 42 +++++ ...IT-WP-0015-independent-evidence-returns.md | 44 +++++ 21 files changed, 869 insertions(+), 95 deletions(-) create mode 100644 docs/cross-repo-evidence-admission.md create mode 100644 workplans/AUDIT-WP-0012-chained-sqlite-import.md create mode 100644 workplans/AUDIT-WP-0013-cross-repo-sender-admission.md create mode 100644 workplans/AUDIT-WP-0014-execution-attribution-receipt.md create mode 100644 workplans/AUDIT-WP-0015-independent-evidence-returns.md diff --git a/SCOPE.md b/SCOPE.md index 38ee8b7..e3b575a 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -33,7 +33,7 @@ effect of its own. - Retention profiles, search and export entitlements, dead-letter and secret-finding surfaces. -**Evidence-role obligations under v0.7 §9.6** *(largely not yet built — see +**Evidence-role obligations under v0.7 §9.6** *(implemented — see `workplans/AUDIT-WP-0009-evidence-role-conformance.md`)* - Recording, per source, whether its evidence is **load-bearing** or diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 72ff918..4b43620 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -21,6 +21,10 @@ | workplan | AUDIT-WP-0009 | finished | — | workplans/AUDIT-WP-0009-evidence-role-conformance.md | | workplan | AUDIT-WP-0010 | finished | — | workplans/AUDIT-WP-0010-tenant-engine-sender-admission.md | | workplan | AUDIT-WP-0011 | finished | — | workplans/AUDIT-WP-0011-informed-decision-memo-posed.md | +| workplan | AUDIT-WP-0012 | finished | — | workplans/AUDIT-WP-0012-chained-sqlite-import.md | +| workplan | AUDIT-WP-0013 | blocked | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| workplan | AUDIT-WP-0014 | blocked | — | workplans/AUDIT-WP-0014-execution-attribution-receipt.md | +| workplan | AUDIT-WP-0015 | blocked | — | workplans/AUDIT-WP-0015-independent-evidence-returns.md | | task | ADHOC-2026-09-14-T01 | done | — | workplans/ADHOC-2026-09-14.md | | task | ADHOC-2026-09-24-T01 | done | — | workplans/ADHOC-2026-09-24.md | | task | AUDIT-WP-0001-T01 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md | @@ -81,9 +85,21 @@ | task | AUDIT-WP-0010-T04 | done | — | workplans/AUDIT-WP-0010-tenant-engine-sender-admission.md | | task | AUDIT-WP-0010-T05 | done | — | workplans/AUDIT-WP-0010-tenant-engine-sender-admission.md | | task | AUDIT-WP-0011-T01 | done | — | workplans/AUDIT-WP-0011-informed-decision-memo-posed.md | +| task | AUDIT-WP-0012-T01 | done | — | workplans/AUDIT-WP-0012-chained-sqlite-import.md | +| task | AUDIT-WP-0012-T02 | done | — | workplans/AUDIT-WP-0012-chained-sqlite-import.md | +| task | AUDIT-WP-0012-T03 | done | — | workplans/AUDIT-WP-0012-chained-sqlite-import.md | +| task | AUDIT-WP-0013-T01 | done | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| task | AUDIT-WP-0013-T02 | wait | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| task | AUDIT-WP-0013-T03 | wait | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| task | AUDIT-WP-0013-T04 | done | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| task | AUDIT-WP-0013-T05 | wait | — | workplans/AUDIT-WP-0013-cross-repo-sender-admission.md | +| task | AUDIT-WP-0014-T01 | done | — | workplans/AUDIT-WP-0014-execution-attribution-receipt.md | +| task | AUDIT-WP-0014-T02 | wait | — | workplans/AUDIT-WP-0014-execution-attribution-receipt.md | +| task | AUDIT-WP-0015-T01 | wait | — | workplans/AUDIT-WP-0015-independent-evidence-returns.md | +| task | AUDIT-WP-0015-T02 | wait | — | workplans/AUDIT-WP-0015-independent-evidence-returns.md | | intake | AUDIT-IN-0001 | closed | — | intakes/intakes.md | -| intake | AUDIT-IN-0002 | open | — | intakes/intakes.md | +| intake | AUDIT-IN-0002 | closed | — | intakes/intakes.md | | intake | AUDIT-IN-0003 | closed | — | intakes/intakes.md | -| intake | AUDIT-IN-0004 | open | — | intakes/intakes.md | +| intake | AUDIT-IN-0004 | closed | — | intakes/intakes.md | | intake | AUDIT-IN-0005 | closed | — | intakes/intakes.md | | intake | AUDIT-IN-0006 | accepted | — | intakes/intakes.md | diff --git a/audit_core/cli.py b/audit_core/cli.py index e455211..831e8b1 100644 --- a/audit_core/cli.py +++ b/audit_core/cli.py @@ -101,6 +101,10 @@ def build_parser() -> argparse.ArgumentParser: help="DSN; defaults to AUDIT_CORE_DATABASE_URL", ) migrate_parser.add_argument("--schema", default="audit_core") + migrate_parser.add_argument( + "--source-id", default=None, + help="Stable import identity; retain it if the SQLite file moves (defaults to its absolute path).", + ) migrate_parser.add_argument( "--dry-run", action="store_true", help="Report what would move without writing anything.", @@ -233,7 +237,7 @@ def migrate_store(args: argparse.Namespace) -> int: backend = PostgresAuditBackend(dsn, schema=args.schema) try: - report = migrate(args.source, backend) + report = migrate(args.source, backend, source_id=args.source_id) finally: backend.close() print(report.summary()) diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index 7daf3c9..dfd3ee4 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -340,7 +340,7 @@ class IngestionApplication: obligation; audit-core supplies the numerator and says nothing about what the answer means. """ - query = parse_qs(environ.get("QUERY_STRING", "")) + query = parse_qs(environ.get("QUERY_STRING", ""), keep_blank_values=True) source = (query.get("source") or [""])[0] if not source: return self._json( @@ -371,6 +371,13 @@ class IngestionApplication: return self._json( start_response, HTTPStatus.BAD_REQUEST, {"error": "tenant_required"} ) + instances = query.get("stream_instance_id") + if instances is not None and (len(instances) != 1 or not instances[0].strip()): + return self._json( + start_response, HTTPStatus.BAD_REQUEST, + {"error": "invalid_stream_instance_id"}, + ) + stream_instance_id = instances[0] if instances else None counter = getattr(self.backend, "event_counts", None) if not callable(counter): return self._json( @@ -378,7 +385,10 @@ class IngestionApplication: {"error": "reconciliation_not_supported"}, ) try: - counts = counter(source, since, until, tenant) + if stream_instance_id is None: + counts = counter(source, since, until, tenant) + else: + counts = counter(source, since, until, tenant, stream_instance_id=stream_instance_id) except BackendUnavailableError as exc: log.error("reconciliation failed: %s", exc) return self._json( @@ -391,6 +401,7 @@ class IngestionApplication: "since": since, "until": until, "counts": counts, + **({"stream_instance_id": stream_instance_id} if instances else {}), # Said in the response because this is the number most likely to # be quoted out of context in someone else's conformance argument. "means": ( diff --git a/audit_core/integrity.py b/audit_core/integrity.py index 41f2fa5..1403fd3 100644 --- a/audit_core/integrity.py +++ b/audit_core/integrity.py @@ -64,6 +64,29 @@ class ChainReport: } +def order_chain_rows(rows: Iterable[ChainRow]) -> list[ChainRow]: + """Follow stored links, independently of historical acceptance timestamps. + + Keep unreachable/forked rows in the result so verify_rows reports a break + rather than silently dropping evidence outside the genesis chain. + """ + remaining = list(rows) + successors: dict[str, list[ChainRow]] = {} + for row in remaining: + successors.setdefault(row.chain_prev, []).append(row) + ordered: list[ChainRow] = [] + used: set[str] = set() + previous = GENESIS + while len(successors.get(previous, [])) == 1: + row = successors[previous][0] + if row.event_id in used: + break + ordered.append(row) + used.add(row.event_id) + previous = row.chain_hash + return ordered + [row for row in remaining if row.event_id not in used] + + def verify_rows( rows: Iterable[ChainRow], *, diff --git a/audit_core/migrate_store.py b/audit_core/migrate_store.py index 6e8d63f..ff572ea 100644 --- a/audit_core/migrate_store.py +++ b/audit_core/migrate_store.py @@ -18,9 +18,13 @@ current time. from __future__ import annotations +import hashlib import json import sqlite3 from dataclasses import dataclass, field +from pathlib import Path + +from audit_core.integrity import CHAIN_LOCK_KEY, chain_link from audit_core.interface import BackendUnavailableError @@ -56,110 +60,172 @@ class MigrationReport: return "\n".join(lines) -def migrate(sqlite_path: str, backend, *, verify: bool = True) -> MigrationReport: +def migrate( + sqlite_path: str, backend, *, verify: bool = True, source_id: str | None = None +) -> 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 + Idempotent: re-running skips events with matching hash, time and record. + An event present with different historical evidence is reported as a conflict and never overwritten — the destination record wins, and a human decides. """ + import psycopg + report = MigrationReport() + # Callers moving a source to a new path must retain its explicit source_id. + source_id = source_id or str(Path(sqlite_path).resolve()) + source_id = hashlib.sha256(source_id.encode()).hexdigest() source = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True) try: + source.execute("BEGIN") # One consistent source snapshot, including counters. 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) + # Same serialization boundary as native accept(), including side-table + # receipts. A retry cannot double-count after an interrupted import. + with backend.pool.connection() as conn, conn.transaction(): + conn.execute("SELECT pg_advisory_xact_lock(%s)", (CHAIN_LOCK_KEY,)) + for event_id, payload_hash, accepted_at, correlation_id, tenant, record in rows: + outcome = _import_event( + backend, conn, 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) + report.dead_letters_written = _copy_dead_letters(source, backend, conn, source_id, report) + report.findings_written = _copy_findings(source, backend, conn, source_id, report) if verify: report.verified = _verify(source, backend, report) + if not backend.verify_chain().intact: + report.conflicts.append("destination hash chain is not intact") + report.verified = False + except psycopg.Error as exc: + raise BackendUnavailableError("SQLite import transaction failed") from exc finally: source.close() return report def _import_event( - backend, event_id, payload_hash, accepted_at, correlation_id, tenant, record + backend, conn, 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,) - ) + existing = conn.execute( + f"SELECT payload_hash, accepted_at, record FROM {backend._events} WHERE event_id = %s", + (event_id,), + ).fetchone() 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" - ) + if existing[0] != payload_hash: + return f"{event_id}: destination holds a different payload — left untouched" + if not _same_instant(existing[1], accepted_at) or existing[2] != record: + return f"{event_id}: destination holds different historical evidence — left untouched" return "present" - backend._execute( + previous = backend._chain_head(conn) + conn.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", + " source, action, record, chain_prev, chain_hash) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)", ( 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), + json.dumps(record, sort_keys=True), previous, + chain_link(previous, payload_hash, event_id), ), ) 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: +def _fingerprint(value) -> str: + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def _receipt(backend, conn, source_id, kind, record_id): + return conn.execute( + f'SELECT content_hash, occurrences FROM "{backend.schema}".import_receipts ' + "WHERE source_id = %s AND kind = %s AND record_id = %s", + (source_id, kind, record_id), + ).fetchone() + + +def _save_receipt(backend, conn, source_id, kind, record_id, content_hash, occurrences=0): + conn.execute( + f'INSERT INTO "{backend.schema}".import_receipts ' + "(source_id, kind, record_id, content_hash, occurrences) VALUES (%s,%s,%s,%s,%s) " + "ON CONFLICT (source_id, kind, record_id) DO UPDATE " + "SET content_hash = excluded.content_hash, occurrences = excluded.occurrences", + (source_id, kind, record_id, content_hash, occurrences), + ) + + +def _copy_dead_letters(source, backend, conn, source_id, report) -> int: + if not source.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='dead_letters'").fetchone(): return 0 - for event_id, received_at, sender, reason, payload_hash, payload, withheld in rows: - backend._execute( + rows = source.execute( + "SELECT id, event_id, received_at, sender, reason, payload_hash, payload, " + "payload_withheld FROM dead_letters ORDER BY id" + ).fetchall() + written = 0 + for row_id, *data in rows: + record_id = str(row_id) + fingerprint = _fingerprint(data) + receipt = _receipt(backend, conn, source_id, "dead-letter", record_id) + if receipt: + if receipt[0] != fingerprint: + report.conflicts.append(f"dead-letter {record_id}: source changed after import") + continue + conn.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)), + (*data[:-1], bool(data[-1])), ) - return len(rows) + _save_receipt(backend, conn, source_id, "dead-letter", record_id, fingerprint) + written += 1 + return written -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: +def _copy_findings(source, backend, conn, source_id, report) -> int: + if not source.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='secret_findings'").fetchone(): return 0 + rows = source.execute( + "SELECT sender, source, action, field_path, outcome, persisted, " + "occurrences, first_seen, last_seen FROM secret_findings" + ).fetchall() + written = 0 for r in rows: - backend._execute( + record_id = _fingerprint(r[:5]) + fingerprint = _fingerprint(r) + receipt = _receipt(backend, conn, source_id, "finding", record_id) + if receipt and receipt[0] == fingerprint: + continue + delta = r[6] - (receipt[1] if receipt else 0) + if delta < 0: + report.conflicts.append(f"finding {record_id}: source occurrence count decreased") + continue + conn.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]), + "SET occurrences = secret_findings.occurrences + excluded.occurrences, " + "persisted = secret_findings.persisted OR excluded.persisted, " + "first_seen = LEAST(secret_findings.first_seen, excluded.first_seen), " + "last_seen = GREATEST(secret_findings.last_seen, excluded.last_seen)", + (r[0], r[1], r[2], r[3], r[4], bool(r[5]), delta, r[7], r[8]), ) - return len(rows) + _save_receipt(backend, conn, source_id, "finding", record_id, fingerprint, r[6]) + written += 1 + return written def _verify(source: sqlite3.Connection, backend, report: MigrationReport) -> bool: @@ -200,4 +266,4 @@ def _same_instant(destination, source_iso: str) -> bool: return False if want.tzinfo is None or destination.tzinfo is None: return False - return abs(destination.timestamp() - want.timestamp()) < 1.0 + return destination == want diff --git a/audit_core/postgres_backend.py b/audit_core/postgres_backend.py index 5eac569..964ad72 100644 --- a/audit_core/postgres_backend.py +++ b/audit_core/postgres_backend.py @@ -45,6 +45,7 @@ from audit_core.integrity import ( chain_link, evaluate_tamper_evidence, load_attestation, + order_chain_rows, verify_rows, ) from audit_core.redaction import Finding @@ -164,6 +165,21 @@ MIGRATIONS: list[tuple[str, str]] = [ ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS chain_prev text; """, ), + ( + "0007-import-receipts", + """ + CREATE INDEX IF NOT EXISTS events_chain_prev_idx ON {schema}.events (chain_prev); + CREATE INDEX IF NOT EXISTS events_chain_hash_idx ON {schema}.events (chain_hash); + CREATE TABLE IF NOT EXISTS {schema}.import_receipts ( + source_id text NOT NULL, + kind text NOT NULL, + record_id text NOT NULL, + content_hash text NOT NULL, + occurrences bigint NOT NULL DEFAULT 0, + PRIMARY KEY (source_id, kind, record_id) + ); + """, + ), ] # Rejection reasons whose payload must never be persisted. Storing the body of @@ -380,11 +396,7 @@ class PostgresAuditBackend: try: with self.pool.connection() as conn, conn.transaction(): conn.execute("SELECT pg_advisory_xact_lock(%s)", (CHAIN_LOCK_KEY,)) - head = conn.execute( - f"SELECT chain_hash FROM {self._events} " - "ORDER BY accepted_at DESC, event_id DESC LIMIT 1" - ).fetchone() - previous = head[0] if head and head[0] else GENESIS + previous = self._chain_head(conn) link = chain_link(previous, payload_hash, event.event_id) inserted = conn.execute( f""" @@ -445,7 +457,8 @@ class PostgresAuditBackend: return [{"accepted_at": _iso(at), **rec} for rec, at in rows] def event_counts( - self, source: str, since: str, until: str, tenant: str | None = None + self, source: str, since: str, until: str, tenant: str | None = None, + *, stream_instance_id: str | None = None ) -> list[dict]: """Per-class counts for one source over a bounded window. @@ -462,6 +475,10 @@ class PostgresAuditBackend: if tenant is not None: sql += " AND tenant = %s" params += (tenant,) + if stream_instance_id is not None: + sql += " AND jsonb_typeof(record->'details'->'data'->'stream_instance_id') = 'string'" + sql += " AND record->'details'->'data'->>'stream_instance_id' = %s" + params += (stream_instance_id,) sql += " GROUP BY action ORDER BY action" return [{"class": action, "count": count} for action, count in self._query(sql, params)] @@ -566,7 +583,7 @@ class PostgresAuditBackend: (), ) return verify_rows( - [ + order_chain_rows([ ChainRow( event_id=r[0], payload_hash=r[1], @@ -575,7 +592,7 @@ class PostgresAuditBackend: accepted_at=_iso(r[4]), ) for r in rows - ], + ]), attestation=attestation, ) @@ -634,6 +651,23 @@ class PostgresAuditBackend: except Exception as exc: # pool/libpq can raise outside psycopg.Error raise BackendUnavailableError(str(exc)) from exc + def _chain_head(self, conn) -> str: + """Find the unreferenced link while holding CHAIN_LOCK_KEY. + + accepted_at is historical evidence, not insertion order: an imported + record may predate a record already held without rewriting its chain. + """ + heads = conn.execute( + f"SELECT e.chain_hash FROM {self._events} e WHERE NOT EXISTS (" + f"SELECT 1 FROM {self._events} successor " + "WHERE successor.chain_prev = e.chain_hash) LIMIT 2" + ).fetchall() + if len(heads) == 1: + return heads[0][0] + if heads or conn.execute(f"SELECT EXISTS (SELECT 1 FROM {self._events})").fetchone()[0]: + raise BackendUnavailableError("audit chain has no unique head") + return GENESIS + def _backfill_chain(self, conn) -> None: """Fill chain columns on rows accepted before migration 0006. diff --git a/audit_core/sqlite_backend.py b/audit_core/sqlite_backend.py index 846ec54..e4ed522 100644 --- a/audit_core/sqlite_backend.py +++ b/audit_core/sqlite_backend.py @@ -220,7 +220,8 @@ class SQLiteAuditBackend: return [{"accepted_at": at, **json.loads(rec)} for rec, at in rows] def event_counts( - self, source: str, since: str, until: str, tenant: str | None = None + self, source: str, since: str, until: str, tenant: str | None = None, + *, stream_instance_id: str | None = None ) -> list[dict]: """Per-class counts for one source over a bounded window. @@ -238,6 +239,10 @@ class SQLiteAuditBackend: if tenant is not None: sql += " AND tenant = ?" params += (tenant,) + if stream_instance_id is not None: + sql += " AND json_type(record, '$.details.data.stream_instance_id') = 'text'" + sql += " AND json_extract(record, '$.details.data.stream_instance_id') = ?" + params += (stream_instance_id,) sql += " GROUP BY action ORDER BY action" return [{"class": action, "count": count} for action, count in self._query(sql, params)] diff --git a/docs/cross-repo-evidence-admission.md b/docs/cross-repo-evidence-admission.md new file mode 100644 index 0000000..05bbeb2 --- /dev/null +++ b/docs/cross-repo-evidence-admission.md @@ -0,0 +1,166 @@ +# Cross-repo evidence admission requirements + +Reviewed 2026-10-04 against audit-core source, intakes, inbox and owner files. +Work: AUDIT-WP-0013 (senders), AUDIT-WP-0014 (execution receipts), and +AUDIT-WP-0015 (independent evidence). These are source and metadata commitments; +activation requires the relevant owner delivery and approved live change. + +## Flex-auth decision records — AUDIT-WP-0013-T02/T03 + +Owner file: access-engine/workplans/FLEX-WP-0031-decision-record-emission.md, +reviewed at `257a83e997d7d30d3ace2b4524fab47cdfb7a6de`. T03's durable outbox +is done. Do not keep registration waiting on an outbox that already exists. +Audit-core accepts custody admission per AUDIT-IN-0006; it does not decide how +authorization effects are released. + +Flex-auth must supply six final non-secret registration payloads and exact +namespace/pod ingress selectors: ops-warden, tenant-engine, user-engine, +secrets-engine, informed-decision-t03 and informed-decision-sitting. Each +identity permits its exact `flex-auth.` source, write true/read false, +`tenants: ["*"]` with the existing deny-before-tenant-creation justification, +load-bearing and explicitly selected `secret_policy: redact` in the protected +registry. No wildcard source match and no credential in the payload. + +The emitter actually uses `flex-auth.decision.` (ClassPrefix in +internal/emission/outbox.go). Final heartbeat keys must therefore be +`flex-auth.decision.deny`, `.redact`, `.not_applicable`, `.audit_only`, each at +86400 seconds; shorthand effect names would monitor the wrong stored class. +Heartbeat wire type is `audit-core.heartbeat`, with `data.class` set to the +exact vouched-for class, independent non-empty correlation/id and offset-aware +occurred_at. Each per-pin canonical `emission_cadence` must name the exact +source and classes; flex-auth supplies allow's expected rate, window and +threshold. Audit-core will not choose them for the source. Counts compare +committed outbox entries, not all released decisions, by identical class/window. + +The remaining custody owner is railiance-platform; ops-warden must supply a usable sender-mint routing entry. +The current audit-core-senders entry exists but is draft, unresolvable and +non-executable, names ops-mason, and describes an old in-cluster generation +path; it is not the reviewed attended per-pin mint lane. Correct/promote the +existing routing record or return a concrete owner lane, without creating a +second competing registry. Use `warden route` before any credential +request. An attended seed/delivery needs the founder's terminal and per-change +go. Then audit-core reviews the payloads, prepares exact scope/ingress, +and verifies accepted/duplicate writes, wrong source/tenant/read refusals, +reconciliation, per-class missing heartbeats and formerly-valid bearer +revocation. Source and owner revisions plus sanitized receipts are required. + +Gate House GH-WP-0004-T12 owes the final decision for FLEX-DEC-2026-018. +Message `715cf1f2-55dd-4d2d-9b7b-c5dffa0181e5` is explicitly a reading, not +that decision: current text admits no load-bearing failure-path exception. +Audit-core retains the non-atomic/non-conformant description for the restrictive +classes released uncommitted. It will not repurpose attributive +`completeness_trade`, add a permissive schema field ahead of the ruling, or +claim atomicity. The final ruling blocks an exception/conformance declaration; +it does not block preparing transport custody for honestly described evidence. + +## Qonto collector mapping — AUDIT-WP-0013-T04/T05 + +Accepted bounded receiver admission/mapping commitment for QONTO-WP-0006-T02. +Qonto source reviewed at `bf8a9ede6704523d4744d63bf481ca8b195d86f4`. +rapp-qonto owns the collector runtime; its finished RAPP-QONTO-WP-0003 is +historical rollout evidence, not current collector acceptance. qonto-assistant +owns the source declaration, class inventory, scope and process counters. + +| Source class | Proposed receiver type | Required preserved metadata | +| --- | --- | --- | +| audit.allow | audit.allow | stream_id, stream_instance_id, stream_sequence, event_class and sanitized identity/egress context | +| audit.deny | audit.deny | same metadata; rare load-bearing, no expected-rate monitor | +| audit.heartbeat | audit-core.heartbeat | original event_class audit.heartbeat, timestamp/reason, instance and sequence, both source/window transition counts and window bounds | + +The adapter must supply every field of docs/event-envelope.md. Stable event id +and Idempotency-Key must derive deterministically from the original instance +and sequence so retries cannot mint new accepted records. Tenant and subject +must come from an owner-approved binding, never inferred from a pod name or a +convenience tenant. Preserve the original heartbeat event_class in data; do +not rename Qonto's emitted class in its documentation. + +A heartbeat can vouch for the quiet deny class through `data.class=audit.deny`. +Its assertion is no deny transitions in the window, not proof that there were +no allow transitions. qonto-assistant must approve that mapping and the exact +per-class heartbeat registration with its canonical declaration and actual +configured interval (default 86400). If another vouched-for class is needed, +review fan-out and sequence/count accounting explicitly. Sender-level +`evidence_kind` has one value: use the source owner's explicit load-bearing +admission for this mixed stream, retaining attributive allow and load-bearing +deny/heartbeat in the per-class canonical inventory. No blanket atomicity or +completeness claim follows from admission. + +Audit-core now implements `GET /v1/reconciliation`'s optional +`stream_instance_id`, selecting exact stored `data.stream_instance_id` within +an authorized source/tenant and acceptance-time window. Duplicates count once; +restarts never share counts accidentally. This is source-tested and requires a +reviewed receiver release before live use. It does not turn source-time windows +into acceptance-time windows. The collector/observer must define bounded lag, +watermark/sequence boundaries and the comparable accepted-count derivation; +it must distinguish late arrival from lost evidence. Sender own counts are +available with read false. Independent kings-guard archive/sequence inspection +needs its separately scoped read identity/feed, not a writer upgraded to read +all tenants. No observer access scope is invented here. + +Unblock inputs from rapp-qonto: collector source/image/config/Deployment pins, +metadata-only sample envelopes, exact namespace/pod selectors, authoritative +tenant binding, stable instance/restart/scale-to-zero semantics, retry/drop and +lag behavior, redacted success/failure receipts and actual accepted receiver +records. From qonto-assistant: reviewed mapping and per-class cadence, +classification, retention/access/export/secret policy. Audit-core then prepares +registration and checks acceptance, retries, retained instance/sequence, +wrong-scope refusals and accepted-count derivation. King's Guard KG-WP-0005-T03 +subsequently owns independent missing-heartbeat, sequence-gap and count-divergence +acceptance. T02's sender proof never waits on Qonto T03 observer acceptance. +No bank keys or bank API traffic are needed for these receipts. + +## Execution-attribution receipts — AUDIT-WP-0014 + +Proposal reviewed in net-kingdom/canon/standards/playbook-capability-contract_v0.2.md +at `45a189a39167eef62709df0e857309f7ddc59029` (NK-WP-0040-T02). +Audit-core accepts the evidence-holder role; Railiance is the proposed emitter +and must confirm its exact emitting component and sender source. NetKingdom +owns the canonical receipt schema and validator. Audit-core checks shape and +credential scope, stores the event and chains it. Custody does not authenticate +the historical actor chain or decide whether decision_refs authorize execution. + +Before implementation, NetKingdom must return a versioned schema, artifact +path/digest, unknown/absent semantics and positive/negative fixtures. All listed +fields must be represented: initiator issuer/subject, optional delegation, +executing workload, runtime principal, tenant, environment, run id, source +commit, artifact digest, target, action, decision refs and time interval with +clock source and optional established precision bound. Unknown attribution stays +explicit `unknown`; delegation is absent when none; a time bound is absent when +not established. No bearer or role response may be embedded. + +Railiance must supply the exact wire event class and sender, authoritative +scope, stable retry/id binding to run_id, classification, emission guarantee, +retention/access/export and secret policy, plus emitter revision and metadata +fixtures. Audit-core will wrap the receipt in its existing eight-field envelope, +validate the canonical data schema before accept, enforce declared tenant/source, +refuse malformed/secret content, retain original unknowns, and prove duplicate +and conflict behavior. Cross-check receipt tenant with envelope tenant; require +ordered offset-aware start/end if that is the canonical rule. No independent +local receipt schema is shipped ahead of the owner schema, and generic JSON +acceptance is not advertised as receipt-schema validation. + +## Independent chain-head custody — AUDIT-WP-0015-T01 + +RESOURCE-WP-0002-T06 is done for encrypted logical database copies. That +terminal task is not an unfinished chain-head copy commitment. The 2026-09-15 +AUDIT-WP-0009-T12 receipt proves the in-cluster attestor only. +resource-control and rapp-postgres must name the current offsite operator/owner +record, return a destination outside cluster and Barman failure domains on the +existing Nextcloud + age lane, and supply metadata-only upload and independent +readback receipts for the same chain-head digest, event id/count and observed_at. +State cadence, freshness/failure handling, access separation and retrieval test. +No Nextcloud credential belongs in audit-core. Without this return, the control +remains a database-owner boundary and carries no cluster-owner survival claim. + +## Refreshed E2 and asm-t08 — AUDIT-WP-0015-T02 + +Whitehat commit `7b5bc1e444cf57ae5d9ba1dac0c0189c0833e278` and message +`81bebdd8-5ca7-47b3-aafe-1d00ff8be122` leave WHITEHAT-WP-0006-T03 wait. +The 2026-08-22 E2 run is historical and its -01/-02/-03 engagement ids are +terminal. Whitehat is not requesting a run. A future run needs a distinct +engagement, plane admission and rules-of-engagement approvals; audit-core +creates none of those during this review. The alternative is access-engine's +confirmation of a flex-auth tenant-A identity with no tenant-B authority. +asm-t08's surface owner owes component revision, identities, window and technique. +These remain owner inputs; no field or authorization is fabricated. Original +AUDIT-WP-0008-T05 remains done for its historical accepted scope. diff --git a/docs/integrity.md b/docs/integrity.md index dac23c2..c301887 100644 --- a/docs/integrity.md +++ b/docs/integrity.md @@ -22,8 +22,13 @@ duplicate accept must not mint a second link. A conflict must not advance the head. Concurrent first-accepts of different events take an advisory lock (`CHAIN_LOCK_KEY`) so they cannot fork. -Verify walks `ORDER BY accepted_at, event_id`, recomputes every link, -and reports the first event id whose stored hashes do not match. A +PostgreSQL verification follows `chain_prev` from genesis, recomputes every +link, and retains orphaned or forked rows as breaks. The next append selects +the unique unreferenced head under the same advisory lock. This separates +chain insertion order from historical `accepted_at`: importing older SQLite +events extends the existing chain without rewriting timestamps or links. +SQLite's development chain still walks `ORDER BY accepted_at, event_id`. +Verification reports the first event id whose stored hashes do not match. A break is a **custody defect**, not a sender error. Retrying the event will not repair it. diff --git a/docs/operator-runbook.md b/docs/operator-runbook.md index f31dbe2..552ddfa 100644 --- a/docs/operator-runbook.md +++ b/docs/operator-runbook.md @@ -290,3 +290,27 @@ The historical SQLite importer has five PostgreSQL test failures tracked by AUDIT-IN-0004. It is outside the live receiver path; do not use it as a recovery procedure until that record closes. The release test result must state this exclusion explicitly, never describe the entire suite as green. + + +### Historical SQLite import + +`python -m audit_core migrate-store --from-sqlite --source-id ` +uses PostgreSQL custody with schema migration `0007-import-receipts`. Run this +with the migration owner's database authority, not a runtime writer identity. +Keep `source-id` unchanged across file relocation and retries; its default is +the absolute file path. Replacing a source at that path requires a new identity. +No source content is copied into the receipt table: it holds hashed source and +row identities, content fingerprints and previously imported finding counts. + +Import reads one SQLite snapshot and writes under the native append advisory +lock. Historical events extend the stored link chain with original IDs, hashes, +records and exact acceptance timestamps. Existing differing evidence is reported +as a conflict and retained; no trigger is disabled and no existing chain link +is rewritten. Retries skip imported dead letters and add only the incremental +source finding count. Decreasing counts or changed historical dead letters are +conflicts. Verification checks events and the resulting chain independently. + +The new PostgreSQL receiver and attestor follow links rather than sorting by +historical timestamps. Deploy that code to all PostgreSQL append/verify workers +before importing historical rows into a populated destination; older workers +still use timestamp order. Source tests do not authorize a production import. diff --git a/docs/stream-completeness.md b/docs/stream-completeness.md index 8621c5c..03696f6 100644 --- a/docs/stream-completeness.md +++ b/docs/stream-completeness.md @@ -114,6 +114,19 @@ The response carries a `means` field stating that agreement proves neither completeness nor that any event occurred — because this is the number most likely to be quoted out of context in someone else's conformance argument. + +The optional `stream_instance_id` query parameter restricts counts to the exact +string stored at `details.data.stream_instance_id`. The response echoes that +filter; no filter preserves the existing aggregate response. Empty or repeated +instance parameters are refused. Source and tenant authorization still apply, +and duplicate deliveries count once. Both SQLite and PostgreSQL implement it. +These are **acceptance-time** windows (`since` inclusive, `until` exclusive), +not the producer's occurrence-time or sequence windows. A collector/observer +must account for drain lag before comparing source-window counters, retain +`stream_sequence`, and never compare a restarted process's reset counters with +another instance or a lifetime total. Instance isolation alone does not detect +sequence gaps or determine whether an instance is active. + ## Declared emission cadence `AUDIT-WP-0009-T05`. The contract is info-tech-canon's EmissionCadence diff --git a/intakes/intakes.md b/intakes/intakes.md index 44e41ea..ccdd5fc 100644 --- a/intakes/intakes.md +++ b/intakes/intakes.md @@ -24,7 +24,8 @@ description: > deny/redact/not_applicable without a durable record when the outbox commit fails and counts them as released_uncommitted, and for the token lane. created: '2026-09-23' -updated: '2026-09-24' +updated: '2026-10-04' +work: AUDIT-WP-0013-T02 outcome: accepted-with-corrections-registration-deferred resolution: '(1) One sender per pin: accepted, since each pin can reconcile only what it committed. Correction: may_read false, not true. A writer counts its @@ -54,6 +55,9 @@ resolution: '(1) One sender per pin: accepted, since each pin can reconcile only openbao-attended-exec.py, as for tenant-engine (AUDIT-WP-0010-T02). Registration entries are deferred until the flex-auth outbox exists (FLEX-WP-0031-T03), at flex-auth''s own request.' +follow_up: 'AUDIT-WP-0013-T02/T03: FLEX-WP-0031-T03 outbox is already done. + Await final per-pin payloads, catalog and attended custody; final GH-WP-0004-T12 + ruling remains separate from transport. Requirements: docs/cross-repo-evidence-admission.md.' recorded_in: intakes/intakes.md state_hub_intake_id: "01a0d1e0-e10c-714e-9485-9dd841dafbbb" ``` @@ -208,7 +212,7 @@ state_hub_intake_id: "01a08b76-05b9-7a53-8359-9cda50755f28" id: AUDIT-IN-0002 kind: intake title: 'Register tenant-engine as an attributive audit-core sender' -status: open +status: closed origin: cross-repo origin_ref: TEN-WP-0011-T04 priority: high @@ -232,26 +236,16 @@ description: > Requested: sender registration + token lane, or a correction if the envelope needs a field this engine is not sending. created: '2026-08-29' -updated: '2026-09-10' -outcome: registered-with-a-blocking-correction -resolution: 'Registered attributive with the declared completeness_trade - recorded receiver-side, tenants ["*"] justified per sender, source pinned - exact, ingress added — all inert pending the token - (docs/tenant-engine-source-registration.md, AUDIT-WP-0010 T01/T03/T04). - Taking the correction the intake invited: the envelope does not match. - envelope_for sends five required fields under other names (event_id, action, - resource, observed_at, details) and omits correlation_id entirely, so - normalize() raises invalid_event and every event would 400 and dead-letter. - Because the drain treats 400 as terminal, the outbox row is marked handled - while audit-core holds only a dead letter — lost on both sides, silently, - presenting as a working integration. normalize() is not being relaxed to - accept the alternate spellings: guessing which sender key means which stored - field would make the mapping audit-core''s rather than the sender''s, and - correlation_id cannot be synthesized at all. Root cause is audit-core''s — - the wire envelope was documented nowhere a sender could read it, and - schema_version audit-core.event.v1alpha1 selects nothing here. Contract now - published at docs/event-envelope.md. Stays OPEN until tenant-engine corrects - the emitter and T05 records live evidence.' +updated: '2026-10-04' +closed: '2026-09-15' +work: AUDIT-WP-0010-T05 +outcome: promoted +resolution: 'Reconciled 2026-10-04 against AUDIT-WP-0010-T05: completed + 2026-09-15, native tenant-engine Job accepted 202 then duplicate 200 with + intact 60-event chain. Receipt docs/evidence/2026-09-15-tenant-engine-accept.json. + This supersedes the historical envelope/input wait. Ongoing producer Deployment + drain wiring remains tenant-engine-owned and is not re-opened as audit-core + sender admission.' state_hub_intake_id: "01a04d8f-b1c1-746a-8007-15221ebf0a48" ``` @@ -316,14 +310,21 @@ state_hub_intake_id: "01a04d8f-be67-75ed-a221-d50f99dbb78e" id: AUDIT-IN-0004 kind: intake title: Repair SQLite import into chained PostgreSQL custody -status: open +status: closed origin: residual origin_ref: AUDIT-WP-0005 priority: medium owner: audit-core requested_by: codex created: '2026-09-11' -updated: '2026-09-11' +updated: '2026-10-04' +closed: '2026-10-04' +work: AUDIT-WP-0012 +outcome: repaired-and-postgres-verified +resolution: 'Snapshot import shares native chain lock, follows stored links without + rewriting historical times/links, and uses transactional replay receipts for + dead letters and incremental findings. Full PostgreSQL suite passes without + migration exclusions. Source completion only; no native import performed.' description: > The 2026-09-11 receiver release rehearsal enables real PostgreSQL tests. Five tests in tests/test_migrate_store.py fail because _import_event omits diff --git a/tests/test_backend_conformance.py b/tests/test_backend_conformance.py index bc33bc5..57fcb78 100644 --- a/tests/test_backend_conformance.py +++ b/tests/test_backend_conformance.py @@ -488,3 +488,28 @@ def test_a_stale_mounted_credential_fails_closed(tmp_path): finally: admin.execute(f"DROP ROLE IF EXISTS {role}") admin.close() + + +def test_instance_reconciliation_counts_only_stored_records_in_window(backend): + for event_id, instance, source, tenant in [ + ('a-1', 'a', 'user-engine', 'tenant:friendly:binky'), + ('a-2', 'a', 'user-engine', 'tenant:friendly:binky'), + ('b-1', 'b', 'user-engine', 'tenant:friendly:binky'), + ('sibling-source', 'a', 'other-source', 'tenant:friendly:binky'), + ('sibling-tenant', 'a', 'user-engine', 'tenant:other'), + ('numeric-instance', 1, 'user-engine', 'tenant:friendly:binky'), + ('text-instance', '1', 'user-engine', 'tenant:friendly:binky'), + ]: + event = make_event(event_id, source=source, tenant=tenant, + details={'data': {'stream_instance_id': instance, 'stream_sequence': 1}}) + backend.accept(event, digest(event)) + backend.accept(event, digest(event)) + assert backend.event_counts('user-engine', '2020-01-01T00:00:00+00:00', + '2030-01-01T00:00:00+00:00', 'tenant:friendly:binky', + stream_instance_id='a') == [{'class': 'membership.added', 'count': 2}] + assert backend.event_counts('user-engine', '2020-01-01T00:00:00+00:00', + '2020-01-02T00:00:00+00:00', 'tenant:friendly:binky', + stream_instance_id='a') == [] + assert backend.event_counts('user-engine', '2020-01-01T00:00:00+00:00', + '2030-01-01T00:00:00+00:00', 'tenant:friendly:binky', + stream_instance_id='1') == [{'class': 'membership.added', 'count': 1}] diff --git a/tests/test_integrity.py b/tests/test_integrity.py index d7c8b89..ae30a17 100644 --- a/tests/test_integrity.py +++ b/tests/test_integrity.py @@ -342,3 +342,17 @@ def test_postgres_policy_drops_the_claim_when_the_chain_cannot_be_walked(tmp_pat backend.verify_chain = _unavailable assert backend.retention_policy.tamper_evidence is False assert backend.tamper_evidence_state().reason == "chain_unreadable" + + +def test_link_order_retains_forks_and_unreachable_rows_as_chain_breaks(): + from audit_core.integrity import ChainRow, order_chain_rows, verify_rows + + first = ChainRow('first', 'a', chain_link(GENESIS, 'a', 'first'), GENESIS) + second = ChainRow('second', 'b', chain_link(first.chain_hash, 'b', 'second'), first.chain_hash) + assert verify_rows(order_chain_rows([second, first])).intact + fork = ChainRow('fork', 'c', chain_link(first.chain_hash, 'c', 'fork'), first.chain_hash) + orphan = ChainRow('orphan', 'd', chain_link('absent', 'd', 'orphan'), 'absent') + for extra in (fork, orphan): + ordered = order_chain_rows([second, extra, first]) + assert len(ordered) == 3 + assert not verify_rows(ordered).intact diff --git a/tests/test_migrate_store.py b/tests/test_migrate_store.py index 161af9b..879b295 100644 --- a/tests/test_migrate_store.py +++ b/tests/test_migrate_store.py @@ -168,3 +168,115 @@ def test_empty_source_migrates_cleanly(tmp_path, destination): assert report.events_read == 0 assert report.events_written == 0 assert report.ok is True + + +def test_historical_import_extends_existing_chain_and_preserves_attestation( + populated_sqlite, destination +): + path, _ = populated_sqlite + event = make_event('native-before-import') + destination.accept(event, 'native-hash') + attestation = destination.attest_chain() + old_link = destination._query( + f"SELECT chain_hash FROM {destination._events} WHERE event_id = %s", + (event.event_id,), + )[0][0] + + report = migrate(path, destination) + assert report.ok + assert destination.verify_chain(attestation).intact + assert destination.verify_chain(attestation).attestation_match + assert destination._query( + f"SELECT chain_hash FROM {destination._events} WHERE event_id = %s", + (event.event_id,), + )[0][0] == old_link + destination.accept(make_event('native-after-import'), 'after-hash') + assert destination.verify_chain(attestation).intact + + +def test_replaying_side_tables_does_not_duplicate_evidence(populated_sqlite, destination): + path, _ = populated_sqlite + migrate(path, destination, source_id='stable-source') + replay = migrate(path, destination, source_id='stable-source') + assert replay.ok + assert replay.dead_letters_written == replay.findings_written == 0 + assert len(destination.dead_letters()) == 1 + assert destination.secret_findings()[0]['occurrences'] == 1 + + +def test_incremental_finding_import_adds_only_new_occurrences(populated_sqlite, destination): + from audit_core.redaction import Finding + + path, source = populated_sqlite + migrate(path, destination) + source.count_secret_findings( + sender='user-engine', source='user-engine', action='membership.added', + outcome='redacted', findings=[Finding('data.auth_token', True)], + ) + assert migrate(path, destination).ok + assert migrate(path, destination).ok + assert destination.secret_findings()[0]['occurrences'] == 2 + + +def test_import_and_native_accept_share_one_chain_lock(populated_sqlite, destination): + from concurrent.futures import ThreadPoolExecutor + + path, _ = populated_sqlite + with ThreadPoolExecutor(max_workers=3) as executor: + futures = [executor.submit(migrate, path, destination) for _ in range(2)] + futures.append(executor.submit(destination.accept, make_event('concurrent'), 'native')) + results = [future.result(timeout=20) for future in futures] + assert all(result.ok for result in results[:2]) + assert sum(result.events_written for result in results[:2]) == 3 + assert destination.verify_chain().intact + assert destination.verify_chain().events == 4 + assert len(destination.dead_letters()) == 1 + assert destination.secret_findings()[0]['occurrences'] == 1 + + +def test_matching_hash_does_not_hide_a_different_acceptance_time(populated_sqlite, destination): + path, source = populated_sqlite + migrate(path, destination) + source.db.execute("UPDATE events SET accepted_at = '2020-01-01T00:00:00+00:00' WHERE event_id = 'legacy-1'") + report = migrate(path, destination) + assert not report.ok + assert any('different historical evidence' in conflict for conflict in report.conflicts) + assert destination.verify_chain().intact + + +def test_failed_import_rolls_back_events_and_replay_receipts( + populated_sqlite, destination, monkeypatch +): + import audit_core.migrate_store as importer + import psycopg + from audit_core.interface import BackendUnavailableError + + path, _ = populated_sqlite + original = importer._copy_findings + def fail_after_side_tables(*args): + original(*args) + raise psycopg.OperationalError('injected import failure') + monkeypatch.setattr(importer, '_copy_findings', fail_after_side_tables) + with pytest.raises(BackendUnavailableError): + migrate(path, destination) + assert destination.verify_chain().events == 0 + assert destination.dead_letters() == [] + assert destination.secret_findings() == [] + assert destination._query(f'SELECT count(*) FROM "{destination.schema}".import_receipts', ())[0][0] == 0 + monkeypatch.setattr(importer, '_copy_findings', original) + assert migrate(path, destination).ok + + +def test_source_relocation_with_stable_id_preserves_side_table_replay( + populated_sqlite, destination, tmp_path +): + path, _ = populated_sqlite + # Backup captures WAL-backed source data too. + import sqlite3 + moved = str(tmp_path / 'moved.db') + with sqlite3.connect(path) as source, sqlite3.connect(moved) as copy: + source.backup(copy) + migrate(path, destination, source_id='same-source') + replay = migrate(moved, destination, source_id='same-source') + assert replay.ok + assert replay.events_written == replay.dead_letters_written == replay.findings_written == 0 diff --git a/tests/test_stream_findings.py b/tests/test_stream_findings.py index f3b83ca..d60757c 100644 --- a/tests/test_stream_findings.py +++ b/tests/test_stream_findings.py @@ -270,3 +270,28 @@ def test_the_overlay_may_shorten_a_heartbeat_interval_but_not_lengthen_it(tmp_pa "AUDIT_CORE_SENDERS": json.dumps(base), "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), }) + + +def test_reconciliation_separates_process_instances_and_keeps_scope_checks(app): + for event_id, instance in [('a-1', 'a'), ('a-2', 'a'), ('b-1', 'b')]: + emit(app, event_id, 'audit.deny', data={'stream_instance_id': instance, 'stream_sequence': 1}) + # A delivery retry remains one accepted event. + assert emit(app, 'a-1', 'audit.deny', data={'stream_instance_id': 'a', 'stream_sequence': 1})[0].startswith('200') + base = ('/v1/reconciliation?source=approval-engine&tenant=tenant:platform' + '&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00') + status, body = invoke(app, base + '&stream_instance_id=a') + assert status.startswith('200') + assert body['stream_instance_id'] == 'a' + assert body['counts'] == [{'class': 'audit.deny', 'count': 2}] + assert 'stream_sequence' not in body + assert invoke(app, base + '&stream_instance_id=absent')[1]['counts'] == [] + assert invoke(app, base.replace('tenant:platform', 'tenant:other') + '&stream_instance_id=a')[0].startswith('403') + assert invoke(app, base.replace('source=approval-engine', 'source=other') + '&stream_instance_id=a')[0].startswith('403') + + +@pytest.mark.parametrize('suffix', ['&stream_instance_id=', '&stream_instance_id=a&stream_instance_id=b']) +def test_reconciliation_refuses_ambiguous_instance_filters(app, suffix): + status, body = invoke(app, '/v1/reconciliation?source=approval-engine&tenant=tenant:platform' + '&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00' + suffix) + assert status.startswith('400') + assert body['error'] == 'invalid_stream_instance_id' diff --git a/workplans/AUDIT-WP-0012-chained-sqlite-import.md b/workplans/AUDIT-WP-0012-chained-sqlite-import.md new file mode 100644 index 0000000..ac17c4c --- /dev/null +++ b/workplans/AUDIT-WP-0012-chained-sqlite-import.md @@ -0,0 +1,65 @@ +--- +id: AUDIT-WP-0012 +type: workplan +title: "Repair historical SQLite import into PostgreSQL custody" +domain: infotech +repo: audit-core +status: finished +owner: codex +topic_slug: audit-evidence-dependencies +created: "2026-10-04" +updated: "2026-10-04" +state_hub_workstream_id: "0d3658fe-bbe0-560c-8814-c367a93bb844" +--- + +# AUDIT-WP-0012 — Repair historical SQLite import into PostgreSQL custody + +Repair the open residual AUDIT-IN-0004. Source completion only: coordinating deployment of link-order append/verify workers is required before any historical production import; no deployment or native import is performed by this plan. + +## Preserve historical events in the existing destination chain + +```task +id: AUDIT-WP-0012-T01 +status: done +priority: high +state_hub_task_id: "63d39a39-cae8-5773-a451-dfd78e78eac0" +``` + +AUDIT-IN-0004. Import takes a SQLite snapshot and the same PostgreSQL advisory lock as native append. PostgreSQL follows stored links instead of historical accepted_at order; existing links and exact timestamps remain untouched. No trigger is disabled. New migration 0007 adds indexes and side-table import receipts without editing released migrations. Conflicting historical evidence is reported, never overwritten. CLI exposes stable --source-id across source relocation. + +## Make dead-letter and finding replay idempotent + +```task +id: AUDIT-WP-0012-T02 +status: done +priority: high +state_hub_task_id: "f310efd2-4b8c-5ee6-902f-4ad51a5ce5ec" +``` + +Transactional receipts identify source/row fingerprints. Replays skip dead letters, increment findings only by newly observed source counts, and report changed historic rows or decreasing counts as conflicts. Imported evidence and its receipt commit together. + +## Verify populated destinations, concurrency and complete PostgreSQL suite + +```task +id: AUDIT-WP-0012-T03 +status: done +priority: high +state_hub_task_id: "2ef142c6-ba31-5cea-97c7-618af2f834cb" +``` + +Disposable local PostgreSQL at 127.0.0.1:55446, database audit_core; no production import. Final full suite: 322 passed, zero skips and no migration exclusions. New regression coverage: pre-existing attestation survives older imports, exact acceptance-time conflict, native append after import, concurrent native/import/replay, side-table replay and incremental counts, forks/orphans, and instance count boundaries. Rollback of events/side tables/receipts and stable source identity after file relocation also pass. Optional PyYAML/jsonschema checks were enabled in the repo venv; PYTHONPATH=/home/worsch/infospace-bench/src supplies the existing local canon dependency. + + +## Review decision + +State Hub decision `8a896602-b6c6-4ecf-af81-2504e3e123dc` records the +historical-time versus stored-link ordering rule and the owner evidence bounds. +Validation command: + +```bash +PYTHONPATH=/home/worsch/infospace-bench/src \ +AUDIT_CORE_TEST_DATABASE_URL=postgresql://postgres:test@127.0.0.1:55446/audit_core \ +.venv/bin/python -m pytest -q -rs +``` + +The database and password above are disposable local test fixtures. diff --git a/workplans/AUDIT-WP-0013-cross-repo-sender-admission.md b/workplans/AUDIT-WP-0013-cross-repo-sender-admission.md new file mode 100644 index 0000000..0ef5418 --- /dev/null +++ b/workplans/AUDIT-WP-0013-cross-repo-sender-admission.md @@ -0,0 +1,79 @@ +--- +id: AUDIT-WP-0013 +type: workplan +title: "Complete flex-auth and Qonto sender admission" +domain: infotech +repo: audit-core +status: blocked +owner: codex +topic_slug: audit-evidence-dependencies +created: "2026-10-04" +updated: "2026-10-04" +state_hub_workstream_id: "9e400de2-6ba1-520e-9b25-94aa2470e528" +--- + +# AUDIT-WP-0013 — Complete flex-auth and Qonto sender admission + +All independent local receiver work in this plan is complete. Remaining tasks require named owner inputs or attended custody delivery; the workplan is blocked. Requirements are metadata-only and do not authorize deployment, credentials or bank traffic. + +## Review current owners and publish admission requirements + +```task +id: AUDIT-WP-0013-T01 +status: done +priority: high +state_hub_task_id: "6f9505d8-81ea-5406-b693-0781b254b38d" +``` + +Reviewed AUDIT-IN-0006, FLEX-WP-0031, QONTO-WP-0006 and RAPP-QONTO-WP-0003. Accepted bounded custody/mapping commitments in docs/cross-repo-evidence-admission.md. Flex-auth T03 outbox is already done; transport registration must not wait on it or on downstream drain/observer completion. Exact prefixed heartbeat classes, source-owned rate thresholds and mixed Qonto class semantics are recorded. + +## Admit six flex-auth pin identities and verify custody + +```task +id: AUDIT-WP-0013-T02 +status: wait +priority: high +blocking_reason: "flex-auth must return final per-pin cadence/scope/selector payloads; ops-warden routing catalog and railiance-platform attended seed/delivery are not supplied." +blocked_on: message-from:flex-auth +state_hub_task_id: "fff70081-0d6e-5e16-917a-c7cc3204ad53" +``` + +AUDIT-IN-0006 / FLEX-WP-0031-T02. Unblock with the six exact source registrations, explicit tenant/secret policies, emission_cadence including source-chosen allow thresholds, namespace/pod selectors, a governed usable catalog lane and attended custody receipts. The current audit-core-senders route is draft/unresolvable and describes an old ops-mason path; correct or replace that owner routing before attended mint. Then prepare scoped overlays and verify acceptance/retry/refusal/count/heartbeat/revocation behavior. may_read stays false. Requirements and owner split are in docs/cross-repo-evidence-admission.md; nothing is activated by this task. + +## Resolve the load-bearing failure-path declaration + +```task +id: AUDIT-WP-0013-T03 +status: wait +priority: high +blocking_reason: "Gate House GH-WP-0004-T12 must supply the final decision and any canonical failure-path declaration shape." +blocked_on: message-from:gate-house +state_hub_task_id: "310e28d2-849d-549a-8c4b-d82c23ebad93" +``` + +Message 715cf1f2-55dd-4d2d-9b7b-c5dffa0181e5 is a reading, expressly not a ruling. Keep restrictive released_uncommitted classes non-atomic/non-conformant under current text. Do not treat attributive completeness_trade as an exception. This blocks conformance/exception declaration, not custody preparation in T02. Require the decision ID, applicable text/schema revision, classes, failure condition and detection obligations before implementing a new field. + +## Implement receiver process-instance accepted counts + +```task +id: AUDIT-WP-0013-T04 +status: done +priority: high +state_hub_task_id: "cd7133db-7ce9-5e83-9914-b0e8ecf044a3" +``` + +Optional stream_instance_id on GET /v1/reconciliation selects exact retained details.data.stream_instance_id on both backends, preserving source/tenant permissions and bounded acceptance-time windows. Duplicate retries count once; empty/repeated filters are refused. Aggregate response is unchanged without the filter. API and both-backend regressions pass. This addresses one receiver gap for QONTO-WP-0006-T02, not runtime or observer acceptance. + +## Admit the Qonto collector and verify accepted-class mapping + +```task +id: AUDIT-WP-0013-T05 +status: wait +priority: high +blocking_reason: "rapp-qonto must supply the actual collector binding, scope/selectors, mapping/lag rules and receiver receipts; qonto-assistant must approve classification and per-class cadence." +blocked_on: message-from:rapp-qonto +state_hub_task_id: "76b59808-abc3-5b8d-b31a-e3fb611f2cbe" +``` + +QONTO-WP-0006-T02. Audit-core owns admission and review of accepted mapping; rapp-qonto owns runtime delivery. Required inputs and proposed audit.allow/audit.deny/audit.heartbeat mapping are in docs/cross-repo-evidence-admission.md. Retain original stream instance, sequence and source counts. Approve deterministic retry IDs, heartbeat vouched-for class, authoritative tenant and acceptance-time/sequence-window comparison. Supply pinned collector/runtime artifacts, sanitized success/failure receipts and actual accepted records. Neither source stdout nor finished RAPP-QONTO-WP-0003 closes this gate. KG-WP-0005-T03 observer acceptance follows sender proof, never precedes it. + diff --git a/workplans/AUDIT-WP-0014-execution-attribution-receipt.md b/workplans/AUDIT-WP-0014-execution-attribution-receipt.md new file mode 100644 index 0000000..b6f45eb --- /dev/null +++ b/workplans/AUDIT-WP-0014-execution-attribution-receipt.md @@ -0,0 +1,42 @@ +--- +id: AUDIT-WP-0014 +type: workplan +title: "Validate canonical execution-attribution receipts on ingest" +domain: infotech +repo: audit-core +status: blocked +owner: codex +topic_slug: audit-evidence-dependencies +created: "2026-10-04" +updated: "2026-10-04" +state_hub_workstream_id: "687311f9-8830-5aac-b541-0c4f0413c9e6" +--- + +# AUDIT-WP-0014 — Validate canonical execution-attribution receipts on ingest + +The owner field list is a proposal and its schema is not supplied. Local contract review is complete; implementation cannot proceed honestly until the canonical owner and emitter return the requirements named below. + +## Establish the holder and emitter boundary + +```task +id: AUDIT-WP-0014-T01 +status: done +priority: high +state_hub_task_id: "67ea59cb-b74b-5792-bc11-cdd30dfaf25b" +``` + +NK-WP-0040-T02 / message 48eb1101-a577-40c0-bbaa-39df02198155. Audit-core accepts evidence custody; Railiance is the proposed emitter and must confirm its exact source/component. NetKingdom owns the canonical schema and validator. Published field/unknown/absence/envelope requirements in docs/cross-repo-evidence-admission.md. Audit-core does not validate authorization truth or reconstruct an unknown actor chain. + +## Implement and test canonical receipt validation + +```task +id: AUDIT-WP-0014-T02 +status: wait +priority: high +blocking_reason: "NetKingdom NK-WP-0040-T02 must return the versioned receipt schema, validator and fixtures; Railiance must confirm its emitting source and wire binding." +blocked_on: message-from:net-kingdom +state_hub_task_id: "33efb421-8dd7-5311-8447-9d6ca00992ba" +``` + +Return artifact paths/digests, unknown/optional semantics and positive/negative fixtures covering every proposed receipt field. Railiance returns source/type, run-id/idempotency binding, scope, classification/cadence and custody policies. Then enforce canonical data validation and tenant equality before acceptance; test malformed content, secret withholding, duplicate/conflict and unchanged unknowns. Generic JSON storage is not canonical receipt validation. No local substitute schema or default actor is fabricated. + diff --git a/workplans/AUDIT-WP-0015-independent-evidence-returns.md b/workplans/AUDIT-WP-0015-independent-evidence-returns.md new file mode 100644 index 0000000..0254a5e --- /dev/null +++ b/workplans/AUDIT-WP-0015-independent-evidence-returns.md @@ -0,0 +1,44 @@ +--- +id: AUDIT-WP-0015 +type: workplan +title: "Obtain independent chain-head and refreshed tenancy evidence" +domain: infotech +repo: audit-core +status: blocked +owner: codex +topic_slug: audit-evidence-dependencies +created: "2026-10-04" +updated: "2026-10-04" +state_hub_workstream_id: "ad0685fd-77cc-5415-afae-87ea7226b195" +--- + +# AUDIT-WP-0015 — Obtain independent chain-head and refreshed tenancy evidence + +Track unmet independent evidence returns without relabeling historically completed implementation tasks as incomplete. Both tasks require owner evidence beyond audit-core; no local implementation can substitute for it. + +## Obtain independent offsite chain-head readback + +```task +id: AUDIT-WP-0015-T01 +status: wait +priority: high +blocking_reason: "resource-control and rapp-postgres must name a current chain-head custody owner/task and return offsite upload/readback receipts with cadence and failure handling." +blocked_on: message-from:resource-control +state_hub_task_id: "5fa2bfe4-df47-58a1-9240-80e2ad224617" +``` + +AUDIT-WP-0009-T12 proves in-cluster attestation only. Finished RESOURCE-WP-0002-T06 proves database logical-copy recovery, not a chain-head copy operation. Require the independent Nextcloud + age destination outside cluster/Barman domains, matching head digest/event/count/observed_at readback, cadence/freshness/failure owner and retrieval test. Requirements in docs/cross-repo-evidence-admission.md. Do not add Nextcloud credentials to audit-core or claim cluster-owner resilience before this return. + +## Record refreshed E2 and asm-t08 owner prerequisites + +```task +id: AUDIT-WP-0015-T02 +status: wait +priority: high +blocking_reason: "access-engine must confirm a tenant-A flex-auth identity without tenant-B authority, or Whitehat must supply a separately approved future engagement; asm-t08 surface owner inputs remain missing." +blocked_on: message-from:access-engine +state_hub_task_id: "e8a9cd4f-0267-524a-8ddd-5e1cec33e3af" +``` + +WHITEHAT-WP-0006-T03 stays wait. Prior -01/-02/-03 engagements are terminal; Whitehat requested no run. The alternative confirmation is access-engine-owned. asm-t08 needs its surface owner to name component revision, identities, window and technique. Preserve historical AUDIT-WP-0008-T05 completion and exact whitehat-security commit 7b5bc1e444cf57ae5d9ba1dac0c0189c0833e278. No engagement is minted, live run approved or packet sent. Requirements are recorded in docs/cross-repo-evidence-admission.md. +