tegwick
36445ae679
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e75a-fc5c-7913-9dba-9846210c766d
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""Command-line surface for guarded ops-mason builders."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
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)
|
|
if name == "preflight":
|
|
command.add_argument("--readiness-repo", type=Path)
|
|
|
|
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"
|
|
)
|
|
apply_parser.add_argument("--readiness-repo", type=Path)
|
|
apply_parser.add_argument("--activation", choices=("APPROVED", "BREAK_GLASS"), default="APPROVED")
|
|
apply_parser.add_argument("--break-glass-reason")
|
|
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, readiness_repo=args.readiness_repo)
|
|
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,
|
|
readiness_repo=args.readiness_repo,
|
|
activation=args.activation,
|
|
break_glass_reason=args.break_glass_reason,
|
|
)
|
|
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())
|