From 625ad652b9abe69f967bb56d02d9e5ceff8339e8 Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 3 Nov 2026 23:14:29 +0100 Subject: [PATCH] NK-WP-0056-T19: executor approval exercise; parse approval-engine's stored assurance object The exercise runs the real request phase, KeyCape client credentials, preflight and consume against KeyCape's pinned image and approval-engine's source with synthetic keys. It found that approval-engine stores an entry's assurance as the verified token's JSON object (KeyCape assuranceClaim), so the executor's comparison with a bare "aal2" refused every real founder approval. entry_is_aal2 now requires level aal2/aal3 with mfa true and fails closed otherwise. Fakes use the real shape. The request-phase tests freeze the clock inside the example admission's window (they had expired on 2026-11-03T11:00Z). Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 523928@bnt-lap001 Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350 --- bounded-executor/README.md | 6 + bounded-executor/executor.py | 17 +- .../exercise_approval_identity.py | 396 ++++++++++++++++++ bounded-executor/synthetic_pilot.py | 3 +- .../tests/test_approval_request.py | 18 + bounded-executor/tests/test_executor.py | 24 +- 6 files changed, 458 insertions(+), 6 deletions(-) create mode 100644 bounded-executor/exercise_approval_identity.py diff --git a/bounded-executor/README.md b/bounded-executor/README.md index 6bad9ab..1bc755d 100644 --- a/bounded-executor/README.md +++ b/bounded-executor/README.md @@ -48,6 +48,12 @@ without guessing. runs the synthetic journey once and prints its value-free receipt. Exit codes: `0` pass, `1` fail closed, `4` refused. `synthetic_journey_only.py` proves it on the T08 stack: it passes when healthy and fails closed on a broken bind. +`exercise_approval_identity.py` (T19) runs the executor's real request phase, approval client, preflight and consume +against KeyCape's pinned image and approval-engine's source. It uses synthetic keys and writes a value-free receipt: +`python3 bounded-executor/exercise_approval_identity.py --receipt FILE` (defaults `~/key-cape`, `~/approval-engine`). +It found that approval-engine stores an entry's assurance as the token's JSON object, not a bare `aal2`. The executor +now parses that object (`entry_is_aal2`). + `cli.py` (T13) is the Job's entry point: one admission, one approval id, a non-secret config. Exit codes: `0` completed, `3` reverted, `4` refused before any effect, `5` stopped, unknown or failed revert, `6` consumed but not in custody. `Containerfile` builds the image, which `.forgejo/workflows/bounded-executor-image.yaml` publishes by digest. diff --git a/bounded-executor/executor.py b/bounded-executor/executor.py index 98c749b..8ad8120 100644 --- a/bounded-executor/executor.py +++ b/bounded-executor/executor.py @@ -106,6 +106,20 @@ class Deps: sleep: Callable[[float], None] = time.sleep +def entry_is_aal2(assurance) -> bool: + """An approval entry's assurance as approval-engine stores it: the approver token's verified `assurance` claim, + serialized as JSON (approval-engine api.py, POST .../entries). KeyCape mints {"level", "methods", "mfa", "source", + "at"} (token.go assuranceClaim). The pilot requires level aal2 with MFA verified in that login; aal3 also passes. + Anything else, including a bare "aal2" string that no real carrier produces, fails closed (NK-WP-0056-T19).""" + if isinstance(assurance, str): + try: + assurance = json.loads(assurance) + except ValueError: + return False + return (isinstance(assurance, dict) and assurance.get("level") in ("aal2", "aal3") + and assurance.get("mfa") is True) + + def iso(t: dt.datetime) -> str: return t.astimezone(dt.UTC).isoformat(timespec="seconds").replace("+00:00", "Z") @@ -177,7 +191,8 @@ class Run: raise Refused("approval_no_entries") for entry in entries: # aal2 is recorded, not enforced, by approval-engine today: check every entry if (entry.get("subject_id") not in allowed or str(entry.get("principal_type", "")).lower() != "human" - or entry.get("assurance") != "aal2" or entry.get("subject_id") == self.cfg.executor_identity): + or not entry_is_aal2(entry.get("assurance")) + or entry.get("subject_id") == self.cfg.executor_identity): raise Refused("approval_entry") self.approver_subject = entries[0]["subject_id"] window_end = bea.parse_time(spec["window"]["not_after"]) diff --git a/bounded-executor/exercise_approval_identity.py b/bounded-executor/exercise_approval_identity.py new file mode 100644 index 0000000..ab5226c --- /dev/null +++ b/bounded-executor/exercise_approval_identity.py @@ -0,0 +1,396 @@ +#!/usr/bin/env python3 +"""Executor approval exercise against the real services (NK-WP-0056-T19). + +Disposable KeyCape on the pinned image -> the executor's own client -> approval-engine's real source. +Follows secrets-engine `tools/exercise_approval_identity.py`. Synthetic keys and credentials only. No OpenBao, no +cluster, no live service. + +The executor-side code under test is the real code: `cli.run_request` (request phase), `clients.KeyCapeClientCredentials`, +`clients.ApprovalEngine`, and `executor.Run.preflight` and `consume`. Only the OpenBao lane that delivers the client +secret is replaced: the generated value is handed straight to the client. The T18 tests cover that lane. + +KeyCape runs `client-from-case` output for case `nk-bind-executor-approval` and the `approval-engine-operator` client +from key-cape's example registry. The operator client stands in for the revoker and for a service that tries to bind a +human-control approval. approval-engine runs in-process (WSGI) with its JWT authenticator against KeyCape's JWKS. + +Checks: create, approve, consume, idempotent retry, a different digest, revoke-before-consume, wrong scopes, and the +executor's own refusals. The founder's aal2 entry is seeded through approval-engine's store with the exact assurance +object KeyCape mints (token.go assuranceClaim) and that approval-engine's API persists. That is a limitation: there is +no human PKCE login here. + +Run: python3 bounded-executor/exercise_approval_identity.py --receipt FILE +Writes a value-free receipt (no secret, token or generated value; asserted). Exit 0 when every check passes. +""" +from __future__ import annotations + +import argparse +import base64 +import datetime as dt +import hashlib +import importlib.util +import json +import os +import pathlib +import secrets +import socket +import subprocess +import sys +import tempfile +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from unittest.mock import patch +from wsgiref.simple_server import WSGIRequestHandler, make_server + +import yaml +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +KEYCAPE_IMAGE = ("forgejo.coulomb.social/coulomb/key-cape@sha256:" + "9e082dca78f65a4af3e92d3de3705b99e49ca9b298d9562ad0d37511c4306f8d") # sso-mfa/k8s/keycape/deployment.yaml +HERE = pathlib.Path(__file__).resolve().parent +REPO = HERE.parent +CASE = REPO / "security-case-types/oidc-service-client/examples/nk-bind-executor-approval.yaml" +EXECUTOR = "service:nk-bind-executor" +APPROVER_SUB = "synthetic-founder-sub" +OPERATOR = "approval-engine-operator" +UNDER_TEST = ("cli.py", "clients.py", "executor.py") + + +def _load(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +cli = _load("bounded_executor_cli", HERE / "cli.py") +X, C, OPS = cli.X, cli.C, cli.OPS +keycape_client = _load("keycape_client", REPO / "sso-mfa/k8s/keycape/keycape-client.py") + + +def run(args, timeout=60): + result = subprocess.run(args, capture_output=True, timeout=timeout) + if result.returncode: + raise RuntimeError("contained command failed: " + args[0]) + return result.stdout.decode().strip() + + +def commit(path: pathlib.Path) -> str: + return run(["git", "-C", str(path), "rev-parse", "HEAD"]) + + +def private_write(path: pathlib.Path, value: str) -> None: + with path.open("x") as out: + os.chmod(path, 0o600) + out.write(value) + + +def free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def keycape_assurance(level: str) -> str: + """The assurance object KeyCape mints for a browser login (token.go assuranceClaim), stored by approval-engine's + API as json.dumps(identity.assurance, sort_keys=True) (api.py, POST .../entries).""" + mfa = level != "aal1" + return json.dumps({"level": level, "methods": ["pwd", "otp"] if mfa else ["pwd"], "mfa": mfa, + "source": "key-cape", "at": int(time.time())}, sort_keys=True) + + +def admission(admission_id: str) -> dict: + """The pilot example pinned to the current files, with a window around now (synthetic_pilot.admission).""" + doc = X.bea.load(REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml") + spec = doc["spec"] + spec["id"] = admission_id + now = dt.datetime.now(dt.UTC).replace(microsecond=0) + spec["window"] = {"not_before": (now - dt.timedelta(minutes=5)).isoformat().replace("+00:00", "Z"), + "not_after": (now + dt.timedelta(hours=2)).isoformat().replace("+00:00", "Z")} + files = [{"path": f["path"], "sha256": hashlib.sha256((REPO / f["path"]).read_bytes()).hexdigest()} + for f in spec["plan"]["source"]["files"]] + spec["plan"]["source"]["files"] = files + pins = {f["path"]: f["sha256"] for f in files} + for check in spec["checks"]: + check["sha256"] = pins[check["ref"]] + spec["plan_digest"] = X.bea.plan_digest(doc) + if not X.bea.evaluate(doc, json.loads(X.bea.SCHEMA.read_text()))["valid"]: + raise RuntimeError("synthetic admission invalid") + return doc + + +def expect(fn, *errors): + """Run fn and return the exception it raised; fail if it raised nothing or something else.""" + try: + fn() + except errors as error: + return error + raise AssertionError("expected a refusal") + + +def exercise(args) -> dict: + sys.path.insert(0, str(args.approval_engine_source)) + from approval_engine.api import App + from approval_engine.auth import JWTAuthenticator + from approval_engine.store import Engine + + client = keycape_client.client_from_case(yaml.safe_load(CASE.read_text())) + receipt = { + "schema": "netkingdom.bounded-executor.approval-exercise/v1", + "task": "NK-WP-0056-T19", + "target": "disposable local processes; synthetic keys and credentials", + "started_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"), + "keycape_image": KEYCAPE_IMAGE, + "keycape_contract_commit": commit(args.keycape_source), + "approval_engine_commit": commit(args.approval_engine_source), + "net_kingdom_commit": commit(REPO), + "executor_source_sha256": {n: hashlib.sha256((HERE / n).read_bytes()).hexdigest() for n in UNDER_TEST}, + "case_sha256": hashlib.sha256(CASE.read_bytes()).hexdigest(), + "keycape_client": {k: client[k] for k in ("clientId", "audience", "allowedScopes", "grantTypes", + "serviceSubject", "tenant", "tokenLifetime")}, + "limitations": [ + "approval-engine runs from source in-process, not the deployed image", + "the founder's aal2 entry is seeded through approval-engine's store; no human PKCE login", + "the client secret is handed to the client directly; the OpenBao lane is covered by the T18 tests", + "no informed-decision presentation (T21 tests cover the memo against its real validator)", + "plain HTTP on loopback; production uses KeyCape's public HTTPS issuer", + ], + "checks": {}, + } + checks = receipt["checks"] + values: dict[str, str] = {} + + with tempfile.TemporaryDirectory(prefix="nk56-t19-") as temporary: + root = pathlib.Path(temporary) + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_write(root / "key.pem", key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption()).decode()) + port = free_port() + issuer = f"http://127.0.0.1:{port}" + config = yaml.safe_load((args.keycape_source / "config/dev-config.yaml").read_text()) + examples = yaml.safe_load((args.keycape_source / "config/service-clients.example.yaml").read_text())["clients"] + operator = [c for c in examples if c["clientId"] == OPERATOR] + assert len(operator) == 1 + config.update(issuer=issuer, clients=[client, operator[0]]) + config["authelia"]["issuer"] = "https://synthetic-upstream.invalid" + private_write(root / "config.yaml", yaml.safe_dump(config, sort_keys=False)) + values = {c["clientId"]: secrets.token_urlsafe(48) for c in (client, operator[0])} + private_write(root / "env", "KEYCAPE_CONFIG=/etc/keycape/config.yaml\n" + "".join( + c["secretRef"].removeprefix("env:") + "=" + values[c["clientId"]] + "\n" for c in (client, operator[0]))) + os.chmod(root, 0o755) # the container user reads the bind mount + container = "nk56-t19-keycape-" + secrets.token_hex(5) + created = engine = api_server = None + try: + run(["docker", "run", "-d", "--rm", "--pull=never", "--name", container, "--user", str(os.getuid()), + "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges", + "--publish", f"127.0.0.1:{port}:8080", "--env-file", str(root / "env"), + "--mount", f"type=bind,source={temporary},target=/etc/keycape,readonly", KEYCAPE_IMAGE]) + created = True + for _ in range(80): + try: + with urllib.request.urlopen(issuer + "/jwks", timeout=1) as response: + if response.status == 200: + break + except Exception: + time.sleep(0.25) + else: + raise RuntimeError("disposable KeyCape did not start") + + engine = Engine(root / "approval.sqlite") + app = App(engine, JWTAuthenticator(issuer=issuer, audience="approval-engine", jwks_url=issuer + "/jwks")) + + class Quiet(WSGIRequestHandler): + def log_message(self, *_args): + pass + + api_server = make_server("127.0.0.1", 0, app, handler_class=Quiet) + threading.Thread(target=api_server.serve_forever, daemon=True).start() + api = f"http://127.0.0.1:{api_server.server_port}" + + # ---- the executor's own client, exactly as cli.approval_carrier builds it after the OpenBao read + secret = values[client["clientId"]] + tokens = C.KeyCapeClientCredentials(issuer + "/token", client["clientId"], lambda: secret) + carrier = C.ApprovalEngine(api, token_provider=tokens.token) + operator_tokens = C.KeyCapeClientCredentials(issuer + "/token", OPERATOR, lambda: values[OPERATOR]) + operator_carrier = C.ApprovalEngine(api, token_provider=operator_tokens.token) + + cfg = {"executor_identity": EXECUTOR, "approver_subjects": {"admin": APPROVER_SUB}, + "openbao": {"addr": "http://127.0.0.1:1"}, "journey": {}, "approval_engine": {"url": api}} + run_cfg = X.Config(executor_identity=EXECUTOR, runtime_principal="system:serviceaccount:sso:nk-bind-executor", + tenant="tenant:platform", environment="TEST", approver_subjects={"admin": APPROVER_SUB}) + + def deps_for(doc): + plan = doc["spec"]["plan"] + return X.Deps(carrier=carrier, + operations={s["operation"]: None for s in plan["steps"] + plan.get("revert_steps", [])}, + checks={c["id"]: None for c in doc["spec"]["checks"]}, trust=lambda _s: True, + audit=None, alert=lambda *_a: None) + + def request(doc) -> str: + """The real request phase; only the OpenBao lane is replaced by the client built above.""" + path = root / (doc["spec"]["id"] + ".yaml") + path.write_text(yaml.safe_dump(doc, sort_keys=False)) + with patch.object(cli, "approval_carrier", lambda _cfg, _logins: carrier), \ + patch.object(OPS, "wait_reachable", lambda *_a, **_k: None): + result, code = cli.run_request(str(path), cfg) + assert code == 0 and result["outcome"] == "requested", result.get("reason") + return result["approval_id"] + + def found(approval_id): + return engine.get(approval_id).as_dict() + + # ---- 1. create, approve, consume ------------------------------------------------------------------ + doc = admission("BEA-2026-9101") + digest = X.bea.admission_digest(doc) + aid = request(doc) + obj = found(aid) + assert obj["status"] == "requested" and obj["required_count"] == 1 + assert obj["binding"]["actor"] == EXECUTOR and obj["binding"]["human_control"] is True + assert obj["binding"]["target"] == {"admission_id": "BEA-2026-9101", "admission_digest": digest} + assert obj["binding"]["principal"] == APPROVER_SUB and obj["binding"]["pdp_path"] is False + checks["request_phase_creates_as_executor_via_verified_keycape_jwt"] = True + + binding, validity, required = cli.request_binding(doc, cfg) + spoof = {**binding, "actor": "service:someone-else"} + assert str(expect(lambda: carrier.create(spoof, validity, required), C.ClientError)) == "POST 403" + checks["create_with_another_actor_refused"] = True + + assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status" + checks["preflight_refuses_unapproved"] = True + + assert str(expect(lambda: operator_carrier.http.call("POST", f"/v1/approvals/{aid}/entries", {}, + scope="approval:approve"), C.ClientError)) == "POST 403" + assert found(aid)["entries"] == [] + checks["service_approve_of_human_control_refused"] = True + + engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal2"), principal_type="human", + evidence_ref="synthetic-store-fixture", tenant_source="directory") + assert found(aid)["status"] == "approved" + checks["human_aal2_entry_approves"] = True + + first = X.Run(doc, aid, run_cfg, deps_for(doc)) + first.preflight() + assert first.approver_subject == APPROVER_SUB + checks["executor_preflight_accepts_keycape_aal2_entry"] = True + + first.consume() + obj = found(aid) + assert obj["status"] == "consumed" + assert obj["consumption"]["request_digest"] == digest and obj["consumption"]["decision_id"] == first.run_id + checks["executor_consumes_once_and_confirms_by_read_back"] = True + + again = carrier.consume(aid, digest, first.run_id) + assert again["idempotent"] is True and again["status"] == "consumed" + checks["same_digest_retry_idempotent"] = True + + assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status" + checks["second_run_on_consumed_approval_refused_before_consume"] = True + + assert str(expect(lambda: carrier.consume(aid, "sha256:" + "f" * 64, "other-run"), C.ClientError)) == "POST 409" + assert found(aid)["consumption"]["request_digest"] == digest + checks["different_digest_refused"] = True + + changed = admission("BEA-2026-9101") + changed["spec"]["budget"]["attempts"] = changed["spec"]["budget"]["attempts"] + 1 + assert X.bea.admission_digest(changed) != digest + assert str(expect(lambda: X.Run(changed, aid, run_cfg, deps_for(changed)).preflight(), X.Refused)) \ + == "approval_binding" + checks["changed_admission_refused_by_binding_digest"] = True + + # ---- 2. revoke before consume --------------------------------------------------------------------- + doc = admission("BEA-2026-9102") + aid = request(doc) + engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal2"), principal_type="human", + evidence_ref="synthetic-store-fixture", tenant_source="directory") + X.Run(doc, aid, run_cfg, deps_for(doc)).preflight() + assert str(expect(lambda: carrier.http.call("POST", f"/v1/approvals/{aid}/revoke", {}, + scope="approval:read"), C.ClientError)) == "POST 403" + status, revoked = operator_carrier.http.call("POST", f"/v1/approvals/{aid}/revoke", {}, scope="approval:revoke") + assert status == 200 and revoked["status"] == "revoked" + assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status" + assert str(expect(lambda: carrier.consume(aid, X.bea.admission_digest(doc), "run"), C.ClientError)) == "POST 409" + assert found(aid)["status"] == "revoked" and "consumption" not in found(aid) + checks["revoke_before_consume_refused_by_executor_and_engine"] = True + + # ---- 3. an aal1 human entry: approval-engine accepts it (no floor declared); the executor refuses + doc = admission("BEA-2026-9103") + aid = request(doc) + engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal1"), principal_type="human", + evidence_ref="synthetic-store-fixture", tenant_source="directory") + assert found(aid)["status"] == "approved" + assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_entry" + assert found(aid)["status"] == "approved" + checks["aal1_human_entry_refused_by_executor"] = True + + # ---- 4. wrong scope and wrong secret --------------------------------------------------------------- + form = urllib.parse.urlencode({"grant_type": "client_credentials", "scope": "approval:approve"}).encode() + basic = base64.b64encode(f"{client['clientId']}:{secret}".encode()).decode() + req = urllib.request.Request(issuer + "/token", data=form, headers={"Authorization": "Basic " + basic}) + error = expect(lambda: urllib.request.urlopen(req, timeout=5), urllib.error.HTTPError) + with error: + assert error.code in (400, 401) and json.load(error)["error"] == "invalid_profile_usage" + assert str(expect(lambda: tokens.token("approval:approve"), C.ClientError)).startswith("token 4") + checks["executor_client_refused_approve_scope_by_issuer"] = True + + read_only = C.ApprovalEngine(api, token_provider=lambda _scope: tokens.token("approval:read")) + binding, validity, required = cli.request_binding(admission("BEA-2026-9104"), cfg) + assert str(expect(lambda: read_only.create(binding, validity, required), C.ClientError)) == "POST 403" + assert str(expect(lambda: read_only.consume(aid, "sha256:" + "a" * 64, "run"), C.ClientError)) == "POST 403" + checks["token_for_another_scope_refused_by_engine"] = True + + wrong = C.KeyCapeClientCredentials(issuer + "/token", client["clientId"], lambda: "synthetic-wrong-secret") + assert str(expect(lambda: wrong.token("approval:read"), C.ClientError)).startswith("token 4") + checks["wrong_client_secret_refused"] = True + finally: + if api_server: + api_server.shutdown() + api_server.server_close() + if engine: + engine.close() + if created: + run(["docker", "stop", "--time=5", container]) + + receipt.update(status="passed", cleanup_complete=True, + finished_at=dt.datetime.now(dt.UTC).isoformat(timespec="seconds")) + text = json.dumps(receipt, indent=2, sort_keys=False) + if any(v in text for v in values.values()): + raise RuntimeError("receipt would carry a generated value") + return receipt + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + home = pathlib.Path.home() + parser.add_argument("--keycape-source", type=pathlib.Path, default=home / "key-cape") + parser.add_argument("--approval-engine-source", type=pathlib.Path, default=home / "approval-engine") + parser.add_argument("--receipt", required=True, type=pathlib.Path) + parser.add_argument("--trace", action="store_true", help="print the failing assertion's location (no values)") + args = parser.parse_args(argv) + if args.receipt.exists(): + raise SystemExit("receipt path must be new") + try: + receipt = exercise(args) + except Exception as error: + # Never dump a request, client environment or credential-bearing frame. + where = "" + if args.trace: + tb, line = error.__traceback__, None + while tb: + if tb.tb_frame.f_code.co_filename == __file__: + line = tb.tb_lineno + tb = tb.tb_next + reason = str(error) if isinstance(error, (X.Refused, C.ClientError)) else "" + where = f" at line {line} {reason}".rstrip() + print(f"Executor approval exercise failed: {type(error).__name__}{where}", file=sys.stderr) + return 1 + private_write(args.receipt, json.dumps(receipt, indent=2) + "\n") + print(f"Executor approval exercise passed: {len(receipt['checks'])} checks; temporary resources removed.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/bounded-executor/synthetic_pilot.py b/bounded-executor/synthetic_pilot.py index 8515d13..f6c6328 100644 --- a/bounded-executor/synthetic_pilot.py +++ b/bounded-executor/synthetic_pilot.py @@ -153,7 +153,8 @@ class MemoryCarrier: "target": {"admission_id": spec["id"], "admission_digest": bea.admission_digest(admission)}, "human_control": True, "pdp_path": False}, "validity": {"not_before": spec["window"]["not_before"], "expires_at": spec["window"]["not_after"]}, - "entries": [{"subject_id": APPROVER_SUB, "assurance": "aal2", "principal_type": "human", + "entries": [{"subject_id": APPROVER_SUB, "assurance": json.dumps({"level": "aal2", "mfa": True, "source": "key-cape"}), + "principal_type": "human", "approved_at": OPS.now_utc().isoformat()}]} def get(self, approval_id): diff --git a/bounded-executor/tests/test_approval_request.py b/bounded-executor/tests/test_approval_request.py index 060073c..c688791 100644 --- a/bounded-executor/tests/test_approval_request.py +++ b/bounded-executor/tests/test_approval_request.py @@ -6,6 +6,8 @@ from __future__ import annotations import base64 import contextlib import copy +import datetime as real_dt +import types import importlib.util import io import json @@ -16,6 +18,7 @@ import threading import unittest import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest import mock import yaml @@ -38,6 +41,18 @@ CLIENT_KV = "platform/workloads/keycape/nk-bind-executor-approval" APPROVAL_ROLE = "keycape-nk-bind-executor-approval-bounded-executor-workload-kv-read" +class _Frozen(real_dt.datetime): + """Inside the example admission's window (2026-11-03T09:00Z..11:00Z), so the request phase's window check does + not turn these tests into a time bomb.""" + @classmethod + def now(cls, tz=None): + return cls(2026, 11, 3, 9, 30, tzinfo=real_dt.UTC) + + +FROZEN_DT = types.SimpleNamespace(**{k: getattr(real_dt, k) for k in dir(real_dt) if not k.startswith("__")}) +FROZEN_DT.datetime = _Frozen + + class World: def __init__(self): self.requests = [] # (method, path, headers, body) @@ -117,6 +132,9 @@ def serve(w: World): class Base(unittest.TestCase): def setUp(self): + frozen = mock.patch.object(X, "dt", FROZEN_DT) + frozen.start() + self.addCleanup(frozen.stop) self.w = World() self.server = serve(self.w) self.addCleanup(self.server.shutdown) diff --git a/bounded-executor/tests/test_executor.py b/bounded-executor/tests/test_executor.py index 763c8e9..d74c83a 100644 --- a/bounded-executor/tests/test_executor.py +++ b/bounded-executor/tests/test_executor.py @@ -21,6 +21,11 @@ bea = X.bea PILOT = bea.load(X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml") ADMIN_SUB = "kc-sub-0f3a" # opaque KeyCape subject of the human approver `admin` +# approval-engine stores the verified assurance claim as JSON (api.py); KeyCape mints this shape (token.go). +AAL2 = json.dumps({"at": 1793000000, "level": "aal2", "methods": ["pwd", "otp"], "mfa": True, "source": "key-cape"}, + sort_keys=True) +AAL1 = json.dumps({"at": 1793000000, "level": "aal1", "methods": ["pwd"], "mfa": False, "source": "key-cape"}, + sort_keys=True) NOW = dt.datetime(2026, 11, 3, 9, 30, tzinfo=dt.UTC) SECRET = "synthetic-bind-value-never-emitted" @@ -51,7 +56,7 @@ class FakeCarrier: "human_control": True, "pdp_path": False}, "validity": {"not_before": admission["spec"]["window"]["not_before"], "expires_at": admission["spec"]["window"]["not_after"]}, - "entries": [{"subject_id": ADMIN_SUB, "assurance": "aal2", "principal_type": "human", + "entries": [{"subject_id": ADMIN_SUB, "assurance": AAL2, "principal_type": "human", "approved_at": "2026-11-03T08:55:00Z"}], } for k, v in over.items(): @@ -223,7 +228,7 @@ class RefusalTests(Base): "trust_not_satisfied:runtime_secret_trust": lambda: setattr(self.w, "trust_ok", False), "approval_binding": lambda: self.w.carrier.approval["binding"]["target"].update(admission_digest="sha256:" + "0" * 64), "approval_status": lambda: self.w.carrier.approval.update(status="requested"), - "approval_entry": lambda: self.w.carrier.approval["entries"][0].update(assurance="aal1"), + "approval_entry": lambda: self.w.carrier.approval["entries"][0].update(assurance=AAL1), "approval_expires_before_window_end": lambda: self.w.carrier.approval["validity"].update(expires_at="2026-11-03T10:00:00Z"), } for reason, mutate in cases.items(): @@ -233,13 +238,24 @@ class RefusalTests(Base): self.assert_no_effect(self.run_it(), reason) def test_approval_by_someone_not_listed_or_not_human(self): - for entry in ({"subject_id": "kc-sub-other", "assurance": "aal2", "principal_type": "human"}, - {"subject_id": ADMIN_SUB, "assurance": "aal2", "principal_type": "service"}): + for entry in ({"subject_id": "kc-sub-other", "assurance": AAL2, "principal_type": "human"}, + {"subject_id": ADMIN_SUB, "assurance": AAL2, "principal_type": "service"}): with self.subTest(entry=entry): self.setUp() self.w.carrier.approval["entries"] = [entry] self.assert_no_effect(self.run_it(), "approval_entry") + def test_assurance_must_be_keycapes_verified_aal2(self): + """T19 finding: approval-engine stores the token's assurance object as JSON, never a bare level.""" + for assurance in ("aal2", AAL1, json.dumps({"level": "aal2", "mfa": False}), "not json", None, + json.dumps(["aal2"])): + with self.subTest(assurance=assurance): + self.setUp() + self.w.carrier.approval["entries"][0]["assurance"] = assurance + self.assert_no_effect(self.run_it(), "approval_entry") + for assurance in (AAL2, json.loads(AAL2), json.dumps({"level": "aal3", "mfa": True})): + self.assertTrue(X.entry_is_aal2(assurance)) + def test_actor_must_be_this_executor(self): self.w.carrier.approval["binding"]["actor"] = "service:someone-else" self.assert_no_effect(self.run_it(), "approval_binding")