net-kingdom/bounded-executor/cli.py

#!/usr/bin/env python3
"""Run one admission (NK-WP-0056-T13): the entry point of the rapp-bounded-executor Job.

  python3 bounded-executor/cli.py --admission FILE --approval-id ID --config FILE [--result FILE]

The config (JSON, non-secret) names endpoints, token files and the pilot's targets:

  executor_identity, runtime_principal, tenant, environment, approver_subjects{name: subject_id},
  trust_satisfied[...]                      trust states the operator attests for this run (recorded)
  approval_engine{url, token_file, ca_file?}
  audit_core{url, token_file, ca_file?}
  openbao{addr, k8s_role, jwt_file, ca_file?}  bind write role (CCR-2026-0058); jwt_file: audience-openbao token
  lldap{url, ca_file?, set_password: "/app/lldap_set_password"}
  kubernetes{namespace, external_secret, deployment}
  bind{kv, user}, journey{kv, k8s_role, config{...}}  journey read role (CCR-2026-0059): one login per lane

Exit codes: 0 completed, 3 reverted, 4 refused before any effect, 5 stopped / unknown / revert failed / budget,
6 consumed but not in custody. The result printed is value-free.

  python3 bounded-executor/cli.py --journey-only --config FILE [--result FILE]

  python3 bounded-executor/cli.py --request --admission FILE --config FILE [--result FILE]

Request (NK-WP-0056-T18): file the approval for one admission as the executor itself (approval-engine requires
binding.actor == the token's sub). Binding: action bounded-execution, actor = executor_identity, principal = the
first listed approver's subject, target {admission_id, admission_digest}, validity = the admission window,
human_control on, required_count = number of listed approvers. Prints the approval id; consumes nothing.
Exit codes: 0 requested, 4 refused.

approval_engine.client{token_url, client_id, k8s_role, kv, host?} selects the executor's own KeyCape client
(SecurityCaseType oidc-service-client, case nk-bind-executor-approval): the client secret is read once through its
own OpenBao lane (k8s_role, field CLIENT_SECRET), that login is revoked at once, and every approval call exchanges a
fresh token for exactly its scope. approval_engine.token_file remains for fixtures.

Journey-only (NK-WP-0056-T10): read-only, no admission. Logs in with the journey role only (journey.k8s_role, the
openbao-audience token), runs the synthetic journey once and prints its value-free receipt. No write role, no
Kubernetes or directory client, no approval or audit call. Needs only openbao{addr, jwt_file, ca_file?} and
journey{kv, k8s_role, config}. Exit codes: 0 pass, 1 journey failed (fail closed), 4 refused before the journey.
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).resolve().parent


def _load(name):
    spec = importlib.util.spec_from_file_location(name, HERE / f"{name}.py")
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


X = _load("executor")
OPS = _load("operations")
C = _load("clients")
EXIT = {"completed": 0, "reverted": 3, "refused": 4}


def approval_carrier(cfg: dict, logins: list):
    """The approval-engine client: the executor's own KeyCape client when configured, else a token file."""
    ae, o = cfg["approval_engine"], cfg["openbao"]
    if "client" not in ae:
        return C.ApprovalEngine(ae["url"], ae["token_file"], ae.get("ca_file"))
    c = ae["client"]
    lane = OPS.OpenBao(o["addr"], k8s_role=c["k8s_role"], jwt_file=o["jwt_file"], ca_file=o.get("ca_file"))
    logins.append(lane)
    data, _ = lane.read(c["kv"])
    lane.revoke_self()  # one read; the role's token is not kept for the run
    if set(data) != {"CLIENT_SECRET"} or not isinstance(data["CLIENT_SECRET"], str) or not data["CLIENT_SECRET"]:
        raise OPS.OperationError("approval client kv shape")
    secret = data["CLIENT_SECRET"]
    tokens = C.KeyCapeClientCredentials(c["token_url"], c["client_id"], lambda: secret, host=c.get("host"),
                                        ca_file=c.get("ca_file"))
    return C.ApprovalEngine(ae["url"], token_provider=tokens.token, ca_file=ae.get("ca_file"))


def distinct_roles(cfg: dict) -> None:
    roles = [cfg["openbao"].get("k8s_role"), cfg.get("journey", {}).get("k8s_role"),
             cfg.get("approval_engine", {}).get("client", {}).get("k8s_role")]
    roles = [r for r in roles if r]
    if len(roles) != len(set(roles)):
        raise ValueError("every lane must use its own OpenBao role")


def build(cfg: dict, logins: list):
    """`logins` collects every OpenBao client as soon as it exists, so the caller revokes them all, even when a
    later step of setup fails."""
    o = cfg["openbao"]
    if cfg["journey"]["k8s_role"] == o["k8s_role"]:
        raise ValueError("the journey read and the bind write must use separate roles")
    distinct_roles(cfg)

    OPS.wait_reachable(o["addr"], ca_file=o.get("ca_file"))  # new pod: NetworkPolicy may not be applied yet

    def login(role: str):
        client = OPS.OpenBao(o["addr"], k8s_role=role, jwt_file=o["jwt_file"], ca_file=o.get("ca_file"))
        logins.append(client)
        return client

    bao = login(o["k8s_role"])
    journey_bao = login(cfg["journey"]["k8s_role"])
    directory = OPS.Directory(cfg["lldap"]["url"], [cfg["lldap"].get("set_password", "/app/lldap_set_password")],
                              ca_file=cfg["lldap"].get("ca_file"))
    kube = OPS.Kube()
    k = cfg["kubernetes"]
    operations, checks = OPS.pilot_registry(
        bao=bao, journey_bao=journey_bao, directory=directory, delivery=OPS.KubeDelivery(kube, k["namespace"], k["external_secret"]),
        restart=OPS.KubeRestart(kube, k["namespace"], k["deployment"]), bind_kv=cfg["bind"]["kv"],
        bind_user=cfg["bind"]["user"], journey_kv=cfg["journey"]["kv"], journey_config=cfg["journey"]["config"])
    trusted = set(cfg.get("trust_satisfied", []))

    def alert(kind: str, info: dict) -> None:
        print(json.dumps({"alert": kind, **info}), file=sys.stderr, flush=True)

    deps = X.Deps(carrier=approval_carrier(cfg, logins),
                  operations=operations, checks=checks, trust=lambda state: state in trusted,
                  audit=C.AuditCore(cfg["audit_core"]["url"], cfg["audit_core"]["token_file"],
                                    cfg["audit_core"].get("ca_file")),
                  alert=alert)
    config = X.Config(executor_identity=cfg["executor_identity"], runtime_principal=cfg["runtime_principal"],
                      tenant=cfg["tenant"], environment=cfg["environment"],
                      approver_subjects=dict(cfg["approver_subjects"]))
    return config, deps


def request_binding(doc: dict, cfg: dict) -> tuple[dict, dict, int]:
    """The approval binding for one admission. Refuses an invalid admission, another executor, an unmapped
    approver or a closed window, before anything is sent."""
    report = X.bea.evaluate(doc, json.loads(X.bea.SCHEMA.read_text()))
    if not report["valid"]:
        raise ValueError("admission invalid")
    spec = doc["spec"]
    if spec["executor"]["identity"] != cfg["executor_identity"]:
        raise ValueError("admission names another executor")
    approvers = spec["approval"]["approvers"]
    principal = cfg["approver_subjects"][approvers[0]]
    closes = X.dt.datetime.fromisoformat(spec["window"]["not_after"].replace("Z", "+00:00"))
    if X.dt.datetime.now(X.dt.UTC) >= closes:
        raise ValueError("admission window closed")
    binding = {"action": X.ACTION, "actor": cfg["executor_identity"], "principal": principal,
               "purpose": f"{spec['id']}: {spec['title']}"[:200],
               "target": {"admission_id": spec["id"], "admission_digest": X.bea.admission_digest(doc)}}
    validity = {"not_before": spec["window"]["not_before"], "expires_at": spec["window"]["not_after"]}
    return binding, validity, len(approvers)


def run_request(admission_path: str, cfg: dict) -> tuple[dict, int]:
    logins: list = []
    try:
        doc = X.bea.load(pathlib.Path(admission_path))
        binding, validity, required = request_binding(doc, cfg)
        distinct_roles(cfg)
        OPS.wait_reachable(cfg["openbao"]["addr"], ca_file=cfg["openbao"].get("ca_file"))
        created = approval_carrier(cfg, logins).create(binding, validity, required)
        got = created.get("binding") or {}
        if (got.get("actor") != binding["actor"] or got.get("target") != binding["target"]
                or created.get("status") != "requested" or not created.get("id")):
            raise ValueError("created approval does not match the request")
        result = {"mode": "request", "outcome": "requested", "approval_id": created["id"],
                  "binding_digest": got.get("digest"), "admission_id": binding["target"]["admission_id"],
                  "admission_digest": binding["target"]["admission_digest"], "required_count": required,
                  "effects": "approval requested"}
        code = 0
    except Exception as error:  # nothing was consumed; at most an unapproved request exists
        result = {"mode": "request", "outcome": "refused", "reason": "request:" + type(error).__name__}
        code = 4
    finally:
        for client in logins:
            client.revoke_self()
    return result, code


def journey_only(cfg: dict, logins: list) -> dict:
    o, j = cfg["openbao"], cfg["journey"]
    OPS.wait_reachable(o["addr"], ca_file=o.get("ca_file"))  # new pod: NetworkPolicy may not be applied yet
    client = OPS.OpenBao(o["addr"], k8s_role=j["k8s_role"], jwt_file=o["jwt_file"], ca_file=o.get("ca_file"))
    logins.append(client)
    receipt = OPS.journey_from_kv(client, j["kv"], j["config"])
    return {"mode": "journey-only", "outcome": "pass" if receipt["result"] == "pass" else "fail",
            "effects": False, "journey": receipt}


def run_journey_only(cfg: dict) -> tuple[dict, int]:
    logins: list = []
    try:
        result = journey_only(cfg, logins)
        code = 0 if result["outcome"] == "pass" else 1
    except Exception as error:  # nothing was changed: the journey only reads and signs in
        result = {"mode": "journey-only", "outcome": "refused", "reason": "setup:" + type(error).__name__,
                  "effects": False}
        code = 4
    finally:
        for client in logins:
            client.revoke_self()
    return result, code


def exit_code(result: dict) -> int:
    if result.get("reason") in ("receipt_not_in_custody", "audit_unavailable") and result["outcome"] != "refused":
        return 6
    return EXIT.get(result["outcome"], 5)


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
    parser.add_argument("--journey-only", action="store_true", help="read-only journey run, no admission (T10)")
    parser.add_argument("--request", action="store_true", help="file the approval for --admission (T18)")
    parser.add_argument("--admission")
    parser.add_argument("--approval-id")
    parser.add_argument("--config", required=True)
    parser.add_argument("--result", help="also write the result here")
    args = parser.parse_args(argv)
    if args.request:
        if args.journey_only or args.approval_id or not args.admission:
            parser.error("--request takes --admission and --config only")
        result, code = run_request(args.admission, json.loads(pathlib.Path(args.config).read_text()))
        _emit(result, args.result)
        return code
    if args.journey_only == bool(args.admission or args.approval_id):
        parser.error("give either --journey-only, or --admission with --approval-id")
    if not args.journey_only and not (args.admission and args.approval_id):
        parser.error("--admission and --approval-id go together")
    cfg = json.loads(pathlib.Path(args.config).read_text())
    if args.journey_only:
        result, code = run_journey_only(cfg)
        _emit(result, args.result)
        return code
    admission = X.bea.load(pathlib.Path(args.admission))
    logins: list = []
    try:
        config, deps = build(cfg, logins)
        result = X.Run(admission, args.approval_id, config, deps).execute()
    except Exception as error:  # setup failed before preflight: nothing was consumed
        result = {"outcome": "refused", "reason": "setup:" + type(error).__name__, "run_id": None, "effects": False}
    finally:
        for client in logins:
            client.revoke_self()
    _emit(result, args.result)
    return exit_code(result)


def _emit(result: dict, path: str | None) -> None:
    text = json.dumps(result, indent=2, sort_keys=True)
    if path:
        pathlib.Path(path).write_text(text + "\n")
    print(text)


if __name__ == "__main__":
    sys.exit(main())