coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
net-kingdom/bounded-executor/exercise_approval_identity.py
tegwick 625ad652b9 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 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 523928@bnt-lap001
Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350
2026-11-03 23:14:29 +01:00

396 lines
21 KiB
Python
Raw Blame History

#!/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())
Reference in a new issue View git blame Copy permalink