tegwick
3a7d63e18f
VaultDynamicSecret pulls database/creds/* so a rotating lease is not frozen into KV. Runtime sets AUDIT_CORE_AUTO_MIGRATE=0; schema is a Job with the migration lease. Image base is digest-pinned. Namespace and NetworkPolicies are on the cluster; Deployment waits for the attended OpenBao ESO token.
197 lines
6.8 KiB
Python
197 lines
6.8 KiB
Python
"""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)
|
|
|
|
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 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())
|