ops-mason/src/ops_mason/cli.py

"""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("--types-root", type=Path, help="explicit exact-version SecurityCaseType root")
    plan.add_argument(
        "--allow-unimplemented",
        action="store_true",
        help="reference case only: render against an acknowledged type",
    )
    build = families.add_parser(
        "build", help="build bundle from an accepted SecurityCase and its reviewed plan"
    )
    build.add_argument("--case", required=True, type=Path)
    build.add_argument(
        "--plan", required=True, type=Path, help="the reviewed plan from plan --case"
    )
    build.add_argument(
        "--out", type=Path, help="bundle directory (default: builds/case-<id>)"
    )
    build.add_argument(
        "--allow-unimplemented",
        action="store_true",
        help="reference case only: build against an acknowledged type",
    )
    build.add_argument("--netkingdom-root", type=Path, default=None)
    build.add_argument("--types-root", type=Path, help="explicit exact-version SecurityCaseType root")
    structure = families.add_parser(
        "structure-request", help="read-only typed structural evidence request"
    )
    structure.add_argument("--case", required=True, type=Path)
    structure.add_argument("--plan", required=True, type=Path)
    structure.add_argument("--bundle", required=True, type=Path)
    structure.add_argument("--netkingdom-root", type=Path)
    structure.add_argument("--types-root", type=Path, help="explicit exact-version SecurityCaseType root")
    structure.add_argument("--allow-unimplemented", action="store_true")
    return parser


def _build(args) -> int:
    from ops_mason import case_build, case_plan

    nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT
    try:
        case_id = (
            (case_plan.yaml.safe_load(args.case.read_text()) or {})
            .get("spec", {})
            .get("id", "unknown")
        )
        out = args.out or Path("builds") / f"case-{case_id}"
        result = case_build.build_from_case(
            args.case,
            args.plan,
            out,
            nk_root=nk_root,
            types_root=args.types_root,
            allow_unimplemented=args.allow_unimplemented,
        )
    except (
        case_plan.CasePlanError,
        case_build.CaseBuildRefused,
        FileExistsError,
        OSError,
    ) as exc:
        print(
            json.dumps(
                {"ok": False, "error": str(exc), "issues": getattr(exc, "issues", [])}
            ),
            file=sys.stderr,
        )
        return 2
    fails = case_build.validate_with_platform([Path(p) for p in result["ccrs"]])
    print(
        json.dumps(
            {"ok": not fails, "platform_validation": fails or "pass", **result},
            indent=2,
        )
    )
    return 1 if fails else 0


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,
            types_root=args.types_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 == "structure-request":
        from ops_mason import case_structure, case_plan

        try:
            request = case_structure.prepare(
                args.case,
                args.plan,
                args.bundle,
                nk_root=args.netkingdom_root or case_plan.DEFAULT_NK_ROOT,
                types_root=args.types_root,
                allow_unimplemented=args.allow_unimplemented,
            )
        except Exception:
            print(
                json.dumps(
                    {"ok": False, "error": "structural-source-preparation-refused"}
                ),
                file=sys.stderr,
            )
            return 2
        print(
            json.dumps(
                {
                    "source_only": True,
                    "request_digest": request.fingerprint,
                    "request": request.as_dict(),
                },
                indent=2,
            )
        )
        return 0
    if args.family == "plan":
        return _plan(args)
    if args.family == "build":
        return _build(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())