NK-WP-0056-T18 (part): executor's own approval client and request phase

KeyCapeClientCredentials: client_secret_basic exchange, one scope per token,
public Host for the in-cluster URL, no redirects, bounded response checks.
ApprovalEngine takes a per-scope token provider (create/read/consume) and
can create. cli.py --request files the approval for one admission as
service:nk-bind-executor (principal = first approver's subject, target
admission id+digest, validity = window, human_control, required_count =
approvers) and verifies the created object. The client secret comes through
its own OpenBao lane (oidc-service-client case), read once, login revoked at
once, kept in memory only. Refusals before any call: other executor,
unmapped approver, closed window, shared role. 56 executor tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5502@bnt-lap001
Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196

bounded-executor/cli.py

@@ -19,6 +19,19 @@ Exit codes: 0 completed, 3 reverted, 4 refused before any effect, 5 stopped / un
 
   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
@@ -50,12 +63,39 @@ 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
 
@@ -79,8 +119,7 @@ def build(cfg: dict, logins: list):
     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")),
+    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")),
@@ -91,6 +130,53 @@ def build(cfg: dict, logins: list):
     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
@@ -125,11 +211,18 @@ def exit_code(result: dict) -> int:
 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):

bounded-executor/clients.py

@@ -6,6 +6,7 @@ Errors carry status codes only, never response bodies.
 
 from __future__ import annotations
 
+import base64
 import json
 import pathlib
 import ssl
@@ -18,20 +19,83 @@ class ClientError(Exception):
     pass
 
 
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):  # credentials never follow a redirect
+        return None
+
+
+def _opener(ca_file: str | None):
+    context = ssl.create_default_context(cafile=ca_file) if ca_file else ssl.create_default_context()
+    return urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect(),
+                                       urllib.request.HTTPSHandler(context=context)).open
+
+
+class KeyCapeClientCredentials:
+    """client_credentials exchange at KeyCape (src/internal/server/oidc/token.go: client_secret_basic, every
+    requested scope must be allowed for the client). A fresh token per call and scope; nothing is cached or logged,
+    as secrets-engine's approval consumer does. `secret` is a callable so the value stays in its owner's frame.
+
+    `host` sends the public Host (and X-Forwarded-*) when the token URL is the in-cluster service, the way the
+    synthetic journey reaches KeyCape (sso-mfa/k8s/synthetic-journey/journey.py host_map)."""
+
+    def __init__(self, token_url: str, client_id: str, secret, *, host: str | None = None,
+                 ca_file: str | None = None, timeout: float = 15):
+        parsed = urllib.parse.urlsplit(token_url)
+        if parsed.scheme not in ("https", "http") or not parsed.netloc or parsed.query or parsed.fragment:
+            raise ClientError("token url")
+        if not callable(secret):
+            raise ClientError("secret provider")
+        self.token_url, self.client_id, self.secret, self.host = token_url, client_id, secret, host
+        self.open = _opener(ca_file)
+        self.timeout = timeout
+
+    def token(self, scope: str) -> str:
+        if not scope or " " in scope:
+            raise ClientError("one scope per token")
+        basic = base64.b64encode((urllib.parse.quote(self.client_id, safe="") + ":"
+                                  + urllib.parse.quote(self.secret(), safe="")).encode()).decode()
+        headers = {"Authorization": "Basic " + basic, "Content-Type": "application/x-www-form-urlencoded",
+                   "Accept": "application/json"}
+        if self.host:
+            headers.update({"Host": self.host, "X-Forwarded-Host": self.host, "X-Forwarded-Proto": "https"})
+        body = urllib.parse.urlencode({"grant_type": "client_credentials", "scope": scope}).encode()
+        request = urllib.request.Request(self.token_url, method="POST", headers=headers, data=body)
+        try:
+            with self.open(request, timeout=self.timeout) as response:
+                data = json.loads(response.read(1 << 16))
+        except urllib.error.HTTPError as error:
+            error.close()
+            raise ClientError(f"token {error.code}") from None
+        token = data.get("access_token")
+        if (not isinstance(token, str) or not token or len(token) > 16384 or any(c.isspace() for c in token)
+                or str(data.get("token_type", "")).lower() != "bearer"
+                or not isinstance(data.get("expires_in"), int) or not 0 < data["expires_in"] <= 900):
+            raise ClientError("token response")
+        return token
+
+
 class _Http:
-    def __init__(self, base: str, token_file: str, ca_file: str | None = None, timeout: float = 15):
+    def __init__(self, base: str, token_file: str | None = None, ca_file: str | None = None, timeout: float = 15,
+                 token_provider=None):
         parsed = urllib.parse.urlsplit(base)
         if parsed.scheme not in ("https", "http") or not parsed.netloc or parsed.query or parsed.fragment:
             raise ClientError("base url")
+        if (token_file is None) == (token_provider is None):
+            raise ClientError("exactly one of token_file and token_provider")
         self.base = base.rstrip("/")
         self.token_file = token_file
-        context = ssl.create_default_context(cafile=ca_file) if ca_file else ssl.create_default_context()
-        self.open = urllib.request.build_opener(urllib.request.ProxyHandler({}),
-                                                urllib.request.HTTPSHandler(context=context)).open
+        self.token_provider = token_provider
+        self.open = _opener(ca_file)
         self.timeout = timeout
 
-    def call(self, method: str, path: str, body: dict | None = None, headers: dict | None = None) -> tuple[int, dict]:
-        head = {"Authorization": "Bearer " + pathlib.Path(self.token_file).read_text().strip(),
+    def bearer(self, scope: str | None) -> str:
+        if self.token_provider is not None:
+            return self.token_provider(scope)
+        return pathlib.Path(self.token_file).read_text().strip()
+
+    def call(self, method: str, path: str, body: dict | None = None, headers: dict | None = None,
+             scope: str | None = None) -> tuple[int, dict]:
+        head = {"Authorization": "Bearer " + self.bearer(scope),
                 "Accept": "application/json", **(headers or {})}
         data = None
         if body is not None:
@@ -48,10 +112,20 @@ class _Http:
 
 
 class ApprovalEngine:
-    """GET /v1/approvals/{id} and POST /v1/approvals/{id}/consume (approval-engine api.py)."""
+    """POST /v1/approvals, GET /v1/approvals/{id} and POST /v1/approvals/{id}/consume (approval-engine api.py).
+    Each call asks for exactly the scope it needs (approval:create, approval:read, approval:consume)."""
 
-    def __init__(self, base: str, token_file: str, ca_file: str | None = None):
-        self.http = _Http(base, token_file, ca_file)
+    def __init__(self, base: str, token_file: str | None = None, ca_file: str | None = None, token_provider=None):
+        self.http = _Http(base, token_file, ca_file, token_provider=token_provider)
+
+    def create(self, binding: dict, validity: dict, required_count: int) -> dict:
+        """File the approval request as the actor itself. human_control is always on and pdp_path off (pilot)."""
+        status, data = self.http.call("POST", "/v1/approvals", {
+            "binding": binding, "validity": validity, "required_count": required_count,
+            "human_control": True, "pdp_path": False}, scope="approval:create")
+        if status != 201:
+            raise ClientError(f"create {status}")
+        return data
 
     @staticmethod
     def _path(approval_id: str) -> str:
@@ -60,11 +134,12 @@ class ApprovalEngine:
         return "/v1/approvals/" + urllib.parse.quote(approval_id, safe="")
 
     def get(self, approval_id: str) -> dict:
-        return self.http.call("GET", self._path(approval_id))[1]
+        return self.http.call("GET", self._path(approval_id), scope="approval:read")[1]
 
     def consume(self, approval_id: str, request_digest: str, decision_id: str) -> dict:
         return self.http.call("POST", self._path(approval_id) + "/consume",
-                              {"request_digest": request_digest, "decision_id": decision_id})[1]
+                              {"request_digest": request_digest, "decision_id": decision_id},
+                              scope="approval:consume")[1]
 
 
 class AuditCore:

bounded-executor/tests/test_approval_request.py

@@ -0,0 +1,256 @@
+"""The executor's own approval client and request phase (NK-WP-0056-T18). One fake server plays OpenBao, KeyCape and
+approval-engine; no live service."""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import copy
+import importlib.util
+import io
+import json
+import pathlib
+import sys
+import tempfile
+import threading
+import unittest
+import urllib.parse
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import yaml
+
+HERE = pathlib.Path(__file__).resolve().parent
+
+
+def load(name):
+    spec = importlib.util.spec_from_file_location(name, HERE.parent / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+CLI = load("cli")
+C, OPS, X = CLI.C, CLI.OPS, CLI.X
+ADMISSION = X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml"
+SECRET = "synthetic-client-secret/with:specials"
+CLIENT_KV = "platform/workloads/keycape/nk-bind-executor-approval"
+APPROVAL_ROLE = "workload-kv-read-keycape-nk-bind-executor-approval-bounded-executor"
+
+
+class World:
+    def __init__(self):
+        self.requests = []          # (method, path, headers, body)
+        self.revoked = []
+        self.token_status = 200
+        self.token_body = None
+        self.create_status = 201
+        self.create_override = {}
+        self.kv = {"CLIENT_SECRET": SECRET}
+
+
+def serve(w: World):
+    class H(BaseHTTPRequestHandler):
+        def _send(self, code, body, headers=None):
+            raw = json.dumps(body).encode()
+            self.send_response(code)
+            for k, v in (headers or {}).items():
+                self.send_header(k, v)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+        def _body(self):
+            return self.rfile.read(int(self.headers.get("Content-Length") or 0))
+
+        def do_GET(self):
+            w.requests.append(("GET", self.path, dict(self.headers), None))
+            if self.path == "/v1/sys/health":
+                return self._send(200, {"initialized": True})
+            if self.path == "/v1/platform/data/workloads/keycape/nk-bind-executor-approval":
+                return self._send(200, {"data": {"data": w.kv, "metadata": {"version": 1}}})
+            self._send(404, {})
+
+        def do_POST(self):
+            raw = self._body()
+            w.requests.append(("POST", self.path, dict(self.headers), raw))
+            if self.path == "/v1/auth/kubernetes/login":
+                role = json.loads(raw)["role"]
+                return self._send(200, {"auth": {"client_token": f"tok-{role}", "lease_duration": 900}})
+            if self.path == "/v1/auth/token/revoke-self":
+                w.revoked.append(self.headers.get("X-Vault-Token"))
+                return self._send(204, {})
+            if self.path == "/token":
+                if w.token_status == 302:
+                    return self._send(302, {}, {"Location": "http://elsewhere.invalid/token"})
+                if w.token_status != 200:
+                    return self._send(w.token_status, {"error": "invalid_scope"})
+                scope = urllib.parse.parse_qs(raw.decode())["scope"][0]
+                return self._send(200, w.token_body or {"access_token": "at-" + scope.replace(":", "-"),
+                                                        "token_type": "Bearer", "expires_in": 900})
+            if self.path == "/v1/approvals":
+                body = json.loads(raw)
+                created = {"id": "apr-1", "status": "requested",
+                           "binding": {**body["binding"], "digest": "sha256:" + "ab" * 32, "human_control": True},
+                           "validity": body["validity"], **w.create_override}
+                return self._send(w.create_status, created)
+            self._send(404, {})
+
+        def log_message(self, *a):
+            return
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), H)
+    threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()
+    return server
+
+
+class Base(unittest.TestCase):
+    def setUp(self):
+        self.w = World()
+        self.server = serve(self.w)
+        self.addCleanup(self.server.shutdown)
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.root = pathlib.Path(tmp.name)
+        (self.root / "jwt").write_text("synthetic-openbao-audience-jwt\n")
+        self.addr = f"http://127.0.0.1:{self.server.server_port}"
+
+    def cfg(self, **approval_client):
+        return {
+            "executor_identity": "service:nk-bind-executor",
+            "approver_subjects": {"admin": "uid=admin,ou=people,dc=netkingdom,dc=local"},
+            "openbao": {"addr": self.addr, "k8s_role": "net-kingdom-bind-executor-svc-authelia-write",
+                        "jwt_file": str(self.root / "jwt")},
+            "journey": {"kv": "x", "k8s_role": "net-kingdom-bind-executor-syn-journey-read", "config": {}},
+            "approval_engine": {"url": self.addr, "client": {
+                "token_url": self.addr + "/token", "client_id": "nk-bind-executor-approval",
+                "k8s_role": APPROVAL_ROLE, "kv": CLIENT_KV, "host": "kc.coulomb.social", **approval_client}},
+        }
+
+    def calls(self, path):
+        return [r for r in self.w.requests if r[1] == path]
+
+
+class KeyCapeExchange(Base):
+    def client(self):
+        return C.KeyCapeClientCredentials(self.addr + "/token", "nk-bind-executor-approval", lambda: SECRET,
+                                          host="kc.coulomb.social")
+
+    def test_client_secret_basic_one_scope_public_host(self):
+        self.assertEqual(self.client().token("approval:create"), "at-approval-create")
+        _, _, headers, raw = self.calls("/token")[0]
+        user, _, password = base64.b64decode(headers["Authorization"].split()[1]).decode().partition(":")
+        self.assertEqual((urllib.parse.unquote(user), urllib.parse.unquote(password)),
+                         ("nk-bind-executor-approval", SECRET))
+        self.assertEqual(urllib.parse.parse_qs(raw.decode()),
+                         {"grant_type": ["client_credentials"], "scope": ["approval:create"]})
+        self.assertEqual((headers["Host"], headers["X-Forwarded-Proto"]), ("kc.coulomb.social", "https"))
+
+    def test_refusals(self):
+        with self.assertRaises(C.ClientError):
+            self.client().token("approval:read approval:create")  # one scope per token
+        self.w.token_status = 400
+        with self.assertRaisesRegex(C.ClientError, "token 400"):
+            self.client().token("approval:read")
+        self.w.token_status = 302  # credentials never follow a redirect
+        with self.assertRaises(C.ClientError):
+            self.client().token("approval:read")
+        self.w.token_status = 200
+        for body in ({"access_token": "a", "token_type": "Bearer", "expires_in": 3600},
+                     {"access_token": "a", "token_type": "mac", "expires_in": 900},
+                     {"access_token": "a b", "token_type": "Bearer", "expires_in": 900}):
+            self.w.token_body = body
+            with self.subTest(body=body), self.assertRaisesRegex(C.ClientError, "token response"):
+                self.client().token("approval:read")
+
+
+class ScopesPerCall(Base):
+    def test_each_approval_call_asks_for_exactly_its_scope(self):
+        seen = []
+        ae = C.ApprovalEngine(self.addr, token_provider=lambda scope: seen.append(scope) or "t")
+        ae.create({"action": "a"}, {}, 1)
+        with contextlib.suppress(C.ClientError):
+            ae.get("apr-1")
+        with contextlib.suppress(C.ClientError):
+            ae.consume("apr-1", "sha256:" + "0" * 64, "run-1")
+        self.assertEqual(seen, ["approval:create", "approval:read", "approval:consume"])
+        body = json.loads(self.calls("/v1/approvals")[0][3])
+        self.assertEqual((body["human_control"], body["pdp_path"], body["required_count"]), (True, False, 1))
+
+    def test_exactly_one_token_source(self):
+        with self.assertRaises(C.ClientError):
+            C.ApprovalEngine(self.addr)
+        with self.assertRaises(C.ClientError):
+            C.ApprovalEngine(self.addr, token_file="/x", token_provider=lambda s: "t")
+
+
+class RequestPhase(Base):
+    def run_cli(self, cfg, admission=ADMISSION):
+        (self.root / "cfg.json").write_text(json.dumps(cfg))
+        with contextlib.redirect_stdout(io.StringIO()) as out:
+            code = CLI.main(["--request", "--admission", str(admission), "--config", str(self.root / "cfg.json")])
+        return code, json.loads(out.getvalue()), out.getvalue()
+
+    def admission_with(self, mutate):
+        doc = yaml.safe_load(ADMISSION.read_text())
+        mutate(doc["spec"])
+        path = self.root / "admission.yaml"
+        path.write_text(yaml.safe_dump(doc, sort_keys=False))
+        return path
+
+    def test_files_the_approval_as_the_executor(self):
+        code, result, text = self.run_cli(self.cfg())
+        self.assertEqual((code, result["outcome"], result["approval_id"]), (0, "requested", "apr-1"))
+        doc = X.bea.load(ADMISSION)
+        body = json.loads(self.calls("/v1/approvals")[0][3])
+        self.assertEqual(body["binding"], {
+            "action": "bounded-execution", "actor": "service:nk-bind-executor",
+            "principal": "uid=admin,ou=people,dc=netkingdom,dc=local",
+            "purpose": f"{doc['spec']['id']}: {doc['spec']['title']}"[:200],
+            "target": {"admission_id": doc["spec"]["id"], "admission_digest": X.bea.admission_digest(doc)}})
+        self.assertEqual(body["validity"], {"not_before": doc["spec"]["window"]["not_before"],
+                                            "expires_at": doc["spec"]["window"]["not_after"]})
+        self.assertEqual((body["human_control"], body["pdp_path"], body["required_count"]), (True, False, 1))
+        # the approval token was exchanged for exactly approval:create, with the bearer from KeyCape
+        self.assertEqual(self.calls("/v1/approvals")[0][2]["Authorization"], "Bearer at-approval-create")
+        # the approval lane logged in once, read the secret and was revoked; no other lane logged in
+        logins = [json.loads(r[3])["role"] for r in self.calls("/v1/auth/kubernetes/login")]
+        self.assertEqual(logins, [APPROVAL_ROLE])
+        self.assertIn("tok-" + APPROVAL_ROLE, self.w.revoked)
+        self.assertNotIn(SECRET, text)
+
+    def test_refused_before_anything_is_sent(self):
+        cases = {
+            "another executor": (self.cfg(), self.admission_with(
+                lambda s: s["executor"].update(identity="service:other"))),
+            "unmapped approver": ({**self.cfg(), "approver_subjects": {}}, ADMISSION),
+            "closed window": (self.cfg(), self.admission_with(lambda s: s["window"].update(
+                not_before="2026-01-01T09:00:00Z", not_after="2026-01-01T11:00:00Z"))),
+            "shared role": (self.cfg(k8s_role="net-kingdom-bind-executor-syn-journey-read"), ADMISSION),
+        }
+        for label, (cfg, admission) in cases.items():
+            self.w.requests.clear()
+            with self.subTest(label):
+                code, result, _ = self.run_cli(cfg, admission)
+                self.assertEqual((code, result["outcome"]), (4, "refused"))
+                self.assertEqual([r for r in self.w.requests if r[1] != "/v1/sys/health"], [])
+
+    def test_refused_when_the_created_approval_differs(self):
+        for override in ({"status": "approved"}, {"binding": {"actor": "service:other"}}):
+            self.w.create_override = override
+            with self.subTest(override=override):
+                code, result, _ = self.run_cli(self.cfg())
+                self.assertEqual((code, result["outcome"]), (4, "refused"))
+
+    def test_bad_client_kv_shape_is_refused_and_revoked(self):
+        self.w.kv = {"CLIENT_SECRET": SECRET, "EXTRA": "x"}
+        code, result, text = self.run_cli(self.cfg())
+        self.assertEqual((code, result["reason"]), (4, "request:OperationError"))
+        self.assertIn("tok-" + APPROVAL_ROLE, self.w.revoked)
+        self.assertEqual(self.calls("/v1/approvals"), [])
+        self.assertNotIn(SECRET, text)
+
+
+if __name__ == "__main__":
+    unittest.main()