"""Command line helpers for the development audit backend.""" from __future__ import annotations import argparse import json from typing import Any from audit_core.interface import AuditEvent from audit_core.mock_file_backend import MockFileAuditBackend def parse_detail(values: list[str]) -> dict[str, Any]: details: dict[str, Any] = {} for item in values: if "=" not in item: raise SystemExit(f"detail must be key=value: {item}") key, value = item.split("=", 1) details[key] = value return details def emit(args: argparse.Namespace) -> int: details = parse_detail(args.detail) event = AuditEvent( tenant=args.tenant, scope=args.scope, source=args.source, actor=args.actor, action=args.action, resource=args.resource, outcome=args.outcome, reason=args.reason, details=details, ) backend = MockFileAuditBackend(base_dir=args.dir, retention_days=args.retention_days) path = backend.emit(event) print(json.dumps({"ok": True, "path": path, "event_id": event.event_id}, sort_keys=True)) return 0 def cleanup(args: argparse.Namespace) -> int: backend = MockFileAuditBackend(base_dir=args.dir, retention_days=args.retention_days) removed = backend.cleanup_old_files() print(json.dumps({"ok": True, "removed": removed}, sort_keys=True)) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Audit Core development CLI") sub = parser.add_subparsers(dest="command", required=True) emit_parser = sub.add_parser("emit", help="Write one event to the mock file backend.") emit_parser.add_argument("--tenant", default="platform") emit_parser.add_argument("--scope", default="platform-control-plane") emit_parser.add_argument("--source", required=True) emit_parser.add_argument("--actor") emit_parser.add_argument("--action", required=True) emit_parser.add_argument("--resource", required=True) emit_parser.add_argument("--outcome", default="success") emit_parser.add_argument("--reason") emit_parser.add_argument("--detail", action="append", default=[]) emit_parser.add_argument("--dir", default=None) emit_parser.add_argument("--retention-days", type=int, default=None) emit_parser.set_defaults(func=emit) cleanup_parser = sub.add_parser("cleanup", help="Remove mock audit files older than retention.") cleanup_parser.add_argument("--dir", default=None) cleanup_parser.add_argument("--retention-days", type=int, default=None) cleanup_parser.set_defaults(func=cleanup) schema_parser = sub.add_parser( "migrate", help="Apply pending PostgreSQL schema migrations and exit.", ) schema_parser.add_argument( "--to-postgres", dest="destination", default=None, help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.", ) schema_parser.add_argument("--schema", default="audit_core") schema_parser.set_defaults(func=migrate_schema) replay_parser = sub.add_parser( "replay", help="Re-submit a stored event through accept(); must reconcile as duplicate.", ) replay_parser.add_argument("--event-id", required=True) replay_parser.add_argument( "--to-postgres", dest="destination", default=None, help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.", ) replay_parser.add_argument("--schema", default="audit_core") replay_parser.set_defaults(func=replay_event) 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) verify_parser = sub.add_parser( "verify-chain", help="Walk the custody hash chain; exit 0 only if intact.", ) verify_parser.add_argument( "--to-postgres", dest="destination", default=None, help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.", ) verify_parser.add_argument("--schema", default="audit_core") verify_parser.add_argument( "--against", default=None, help="Chain-head attestation JSON. A cited head missing from the live chain is a break.", ) verify_parser.set_defaults(func=verify_chain) attest_parser = sub.add_parser( "attest-chain", help="Write the live chain head to a file outside the database.", ) attest_parser.add_argument( "--to-postgres", dest="destination", default=None, help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.", ) attest_parser.add_argument("--schema", default="audit_core") attest_parser.add_argument("--output", required=True) attest_parser.set_defaults(func=attest_chain) return parser def _postgres_backend(dsn: str | None, schema: str, *, migrate: bool): import os from audit_core.postgres_backend import PostgresAuditBackend return PostgresAuditBackend( dsn if dsn is not None else os.environ.get("AUDIT_CORE_DATABASE_URL") or "", schema=schema, credential_dir=os.environ.get("AUDIT_CORE_CREDENTIAL_DIR"), migrate=migrate, ) def migrate_schema(args: argparse.Namespace) -> int: """Apply pending schema migrations using the current connection. Production runs this as a Job with the migration lease. The runtime Deployment sets AUDIT_CORE_AUTO_MIGRATE=0 so the app role never needs CREATE TABLE. """ backend = _postgres_backend(args.destination, args.schema, migrate=False) try: applied = backend.migrate() finally: backend.close() print(json.dumps({"ok": True, "schema": args.schema, "applied": applied}, sort_keys=True)) return 0 def replay_event(args: argparse.Namespace) -> int: """Reconcile a stored event. A first-acceptance result is a custody defect.""" backend = _postgres_backend(args.destination, args.schema, migrate=False) try: result = backend.replay(args.event_id) except KeyError: print(json.dumps({"ok": False, "error": "not_found", "event_id": args.event_id})) return 1 finally: backend.close() print(json.dumps({ "ok": True, "event_id": args.event_id, "duplicate": result.duplicate, "reference": result.reference, }, sort_keys=True)) return 0 if result.duplicate else 2 def verify_chain(args: argparse.Namespace) -> int: from audit_core.integrity import load_attestation backend = _postgres_backend(args.destination, args.schema, migrate=False) try: cited = load_attestation(args.against) if args.against else None report = backend.verify_chain(cited) finally: backend.close() print(json.dumps(report.as_dict(), sort_keys=True)) return 0 if report.intact else 1 def attest_chain(args: argparse.Namespace) -> int: from audit_core.integrity import write_attestation backend = _postgres_backend(args.destination, args.schema, migrate=False) try: report = backend.verify_chain() body = write_attestation(args.output, report) finally: backend.close() print(json.dumps({"ok": report.intact, "path": args.output, **body}, sort_keys=True)) return 0 if report.intact else 1 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() return args.func(args) if __name__ == "__main__": raise SystemExit(main())