#!/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. With informed_decision{url, token_file} configured (T21), the approval is then presented to the founder: one memo through informed-decision's present_existing intake, carrying the admission YAML. Exit codes: 0 requested (and presented, if configured), 4 refused, 7 requested but the memo was not accepted (resubmitting is idempotent). 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) MEMO_CLASS = "bounded-execution" REVIEW_GROUP = "net-kingdom-admins" def admission_memo(doc: dict, admission_text: str, binding: dict, approval: dict) -> dict: """The informed-decision intake envelope that presents this executor-filed approval (present_existing profile, informed-decision docs/decision-services-alignment-answer.md S1). The packet carries the admission YAML itself; required highlights point at the scope ceiling, the window and budget, and the revert. The intent restates the approval's binding without the actor (informed-decision takes the actor from its registry).""" spec = doc["spec"] approval_id = approval["id"] memo_id = (spec["id"].lower() + "-" + approval_id.split("-")[0])[:63] item = memo_id + "-admission" digest = "sha256:" + X.hashlib.sha256(admission_text.encode("utf-8")).hexdigest() ceiling = "; ".join(f"{c['mode']} {c['scope']}" for c in spec["scope_ceiling"]) budget = spec["budget"] def highlight(n, severity, section, note): return {"id": f"{memo_id}-h{n}", "item_id": item, "severity": severity, "required_ack": True, "locator": {"kind": "admission", "admission_id": spec["id"], "section": section}, "note": note} return { "kind": "informed-decision.memo-intake", "version": 1, "idempotency_key": f"nk-bind-executor.{approval_id}", "memo_class": MEMO_CLASS, "review_group": REVIEW_GROUP, "drafted_by": "net-kingdom bounded executor (service:nk-bind-executor)", "approval_intent": {k: binding[k] for k in ("action", "principal", "purpose", "target")}, "memo": { "id": memo_id, "version": 1, "question": f"Admit {spec['id']}: {spec['title']}?"[:400], "requested_act": "approve", "binding_level": "organizational", "brief": (f"Bounded Execution Admission {spec['id']} (risk {spec['risk_class']}). " f"Executor {spec['executor']['identity']} may act once, inside " f"{spec['window']['not_before']} .. {spec['window']['not_after']}, within: {ceiling}. " f"Budget {budget['wall_time_seconds']} s, {budget['attempts']} attempt(s). " f"Purpose: {spec['purpose']}")[:20000], "binding": { "justification": None, "principal": {"display_name": f"Named reviewer in {REVIEW_GROUP}", "id": "pending-human-session", "identifiers": [], "kind": "person", "role": "approver"}, "target": {"environment": "prod", "id": spec["id"], "kind": "bounded-execution-admission", "label": spec["title"][:200], "requires_new_bind": False}, "terms": None, }, "step_kind": "approve", "packet": [{"item_id": item, "label": f"Admission {spec['id']} (YAML)", "hash": digest}], "highlights": [ highlight(1, "critical", "scope_ceiling", "Approving lets the executor act once, unattended, within exactly this scope ceiling: " + ceiling), highlight(2, "warning", "window_budget", f"Window {spec['window']['not_before']} .. {spec['window']['not_after']}; budget " f"{budget['wall_time_seconds']} s, {budget['attempts']} attempt(s); the executor stops " "and reverts when a check fails or the budget is exhausted."), highlight(3, "warning", "plan_revert", "Each mutating step names its own checks and revert; read the plan before approving."), ], "locale": "en", "approval_id": approval_id, "approval_binding_digest": approval["binding"]["digest"], "sealed": False, }, "packet": [{"item_id": item, "media_type": "text/plain", "content": admission_text}], } 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 if "informed_decision" in cfg: # present it to the founder (T21); the approval stays usable either way idd = cfg["informed_decision"] envelope = admission_memo(doc, pathlib.Path(admission_path).read_text(), binding, created) try: receipt = C.InformedDecision(idd["url"], idd["token_file"], idd.get("ca_file")).submit(envelope) result.update(outcome="presented", memo_id=receipt["memo_id"], effects="approval requested; memo submitted to informed-decision") except Exception as error: # the approval exists; the memo can be resubmitted with the same key result.update(outcome="requested_not_presented", memo_id=envelope["memo"]["id"], reason="memo:" + type(error).__name__) code = 7 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())