diff --git a/bounded-executor/cli.py b/bounded-executor/cli.py index 5d9cae6..6ba704a 100644 --- a/bounded-executor/cli.py +++ b/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): diff --git a/bounded-executor/clients.py b/bounded-executor/clients.py index d7be2e7..0e1b064 100644 --- a/bounded-executor/clients.py +++ b/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: diff --git a/bounded-executor/tests/test_approval_request.py b/bounded-executor/tests/test_approval_request.py new file mode 100644 index 0000000..69d68da --- /dev/null +++ b/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()