"""Command-line surface for guarded ops-mason builders.""" from __future__ import annotations import argparse import json import sys from collections.abc import Sequence from ops_mason.kubernetes_plane import ( PlaneBundle, PlaneError, apply, preflight, rollback_plan, verify, ) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="ops-mason") families = parser.add_subparsers(dest="family", required=True) plane = families.add_parser("plane", help="guarded Kubernetes security plane") commands = plane.add_subparsers(dest="command", required=True) for name in ("render", "preflight", "verify", "rollback-plan"): command = commands.add_parser(name) command.add_argument("--bundle", required=True) apply_parser = commands.add_parser("apply") apply_parser.add_argument("--bundle", required=True) apply_parser.add_argument("--confirm", required=True, help="exact approved plan id") apply_parser.add_argument( "--expect-digest", required=True, help="exact digest returned by preflight" ) return parser def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) try: bundle = PlaneBundle.load(args.bundle) if args.command == "render": result = bundle.render() elif args.command == "preflight": result = preflight(bundle) elif args.command == "verify": result = verify(bundle) elif args.command == "rollback-plan": result = rollback_plan(bundle) elif args.command == "apply": result = apply( bundle, confirm_plan_id=args.confirm, expected_digest=args.expect_digest, ) else: # pragma: no cover - argparse enforces the command set raise AssertionError(args.command) except (OSError, PlaneError) as exc: print(json.dumps({"ok": False, "error": str(exc)}), file=sys.stderr) return 2 print(json.dumps(result, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": # pragma: no cover raise SystemExit(main())