audit-core/audit_core/cli.py

"""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())