tegwick
a3e97c8bd4
--allow-unimplemented (acknowledged type only, plan marked reference_case) lets the accepted Forgejo case render; reviewed now requires a live survey. WP-0005 lane applied and custody moved by the founder. keycape-factor-issuer filled from net-kingdom, rotation gap attributed to railiance-platform. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 327203@bnt-lap001 Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0
106 lines
4.1 KiB
Python
106 lines
4.1 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")
|
|
|
|
plan = families.add_parser("plan", help="construction plan derived from a SecurityCase")
|
|
plan.add_argument("--case", required=True, type=Path)
|
|
plan.add_argument("--out", type=Path, help="write the plan here (default: stdout)")
|
|
plan.add_argument("--no-live", action="store_true",
|
|
help="skip the live OpenBao survey (marks the plan's survey source 'none')")
|
|
plan.add_argument("--netkingdom-root", type=Path, default=None)
|
|
plan.add_argument("--allow-unimplemented", action="store_true",
|
|
help="reference case only: render against an acknowledged type")
|
|
return parser
|
|
|
|
|
|
def _plan(args) -> int:
|
|
from ops_mason import case_plan
|
|
|
|
nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT
|
|
try:
|
|
doc, collisions, rendered = case_plan.plan_from_case(
|
|
args.case, nk_root=nk_root, structure=case_plan.Structure() if args.no_live else None,
|
|
allow_unimplemented=args.allow_unimplemented)
|
|
except case_plan.CasePlanError as exc:
|
|
print(json.dumps({"ok": False, "error": str(exc), "issues": exc.issues}), file=sys.stderr)
|
|
return 2
|
|
if args.out:
|
|
args.out.write_text(doc)
|
|
else:
|
|
sys.stdout.write(doc)
|
|
print(json.dumps({"ok": not collisions, "digest": rendered["digest"], "collisions": collisions}), file=sys.stderr)
|
|
return 1 if collisions else 0
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = _parser().parse_args(argv)
|
|
if args.family == "plan":
|
|
return _plan(args)
|
|
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())
|