#!/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] 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 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") 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=C.ApprovalEngine(cfg["approval_engine"]["url"], cfg["approval_engine"]["token_file"], cfg["approval_engine"].get("ca_file")), 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 journey_only(cfg: dict, logins: list) -> dict: o, j = cfg["openbao"], cfg["journey"] 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("--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.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())