NK-WP-0056-T21 done: request phase presents the approval through informed-decision
All checks were successful
Build and Publish bounded-executor / build-and-push (push) Successful in 17s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s
Identity provider journey acceptance / provider (push) Successful in 24s

admission_memo composes the present_existing envelope (admission YAML as
packet, three required highlights, intent = binding without actor, key
nk-bind-executor.<approval_id>); InformedDecision.submit posts it with the
projected intake token and checks the receipt. Exit 7 when the memo is
refused (approval stays usable). Envelope validated by informed-decision's
own validate_submission with the present_existing profile. Service-identity
declaration notes key-cape's prepared approval:read. 61 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
This commit is contained in:
tegwick 2026-11-01 00:39:17 +01:00
parent 45feaa13cc
commit e6252bc0e4
6 changed files with 200 additions and 4 deletions

View file

@ -47,6 +47,8 @@ class World:
self.create_status = 201
self.create_override = {}
self.kv = {"CLIENT_SECRET": SECRET}
self.memos = []
self.memo_status = 201
def serve(w: World):
@ -89,6 +91,14 @@ def serve(w: World):
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/memos":
env = json.loads(raw)
w.memos.append((dict(self.headers), env))
if w.memo_status != 201:
return self._send(w.memo_status, {"error": "approval_not_requested"})
m = env["memo"]
return self._send(201, {"memo_id": m["id"], "approval_id": m["approval_id"],
"approval_binding_digest": m["approval_binding_digest"]})
if self.path == "/v1/approvals":
body = json.loads(raw)
created = {"id": "apr-1", "status": "requested",
@ -185,6 +195,62 @@ class ScopesPerCall(Base):
C.ApprovalEngine(self.addr, token_file="/x", token_provider=lambda s: "t")
INFD = pathlib.Path("/home/worsch/informed-decision")
class PresentToFounder(Base):
"""T21: after creating the approval the executor presents it through informed-decision (present_existing)."""
def run_cli(self, cfg):
(self.root / "cfg.json").write_text(json.dumps(cfg))
(self.root / "intake-token").write_text("synthetic-intake-token\n")
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 cfg(self, **kw):
return {**super().cfg(), "informed_decision": {"url": self.addr,
"token_file": str(self.root / "intake-token")}}
def test_files_then_presents_with_the_admission_as_packet(self):
code, result, text = self.run_cli(self.cfg())
self.assertEqual((code, result["outcome"]), (0, "presented"))
headers, env = self.w.memos[0]
self.assertEqual(headers["Authorization"], "Bearer synthetic-intake-token")
self.assertEqual(env["memo"]["approval_id"], "apr-1")
self.assertEqual(env["memo"]["approval_binding_digest"], "sha256:" + "ab" * 32)
self.assertEqual(env["packet"][0]["content"], ADMISSION.read_text())
body = json.loads(self.calls("/v1/approvals")[0][3])
self.assertEqual(env["approval_intent"], {k: body["binding"][k] for k in ("action", "principal", "purpose", "target")})
self.assertNotIn("actor", env["approval_intent"])
self.assertEqual(result["memo_id"], env["memo"]["id"])
self.assertNotIn(SECRET, text)
def test_refused_memo_keeps_the_approval_and_exits_7(self):
self.w.memo_status = 409
code, result, _ = self.run_cli(self.cfg())
self.assertEqual((code, result["outcome"], result["approval_id"]), (7, "requested_not_presented", "apr-1"))
def test_envelope_passes_informed_decisions_own_validator(self):
if not (INFD / "informed_decision" / "intake.py").exists():
self.skipTest("informed-decision checkout absent")
sys.path.insert(0, str(INFD))
try:
from informed_decision import intake
finally:
sys.path.remove(str(INFD))
doc = X.bea.load(ADMISSION)
binding, _, _ = CLI.request_binding(doc, super().cfg())
approval = {"id": "0f1e2d3c-4b5a-4968-8776-655443322110", "binding": {**binding, "digest": "sha256:" + "b" * 64}}
env = CLI.admission_memo(doc, ADMISSION.read_text(), binding, approval)
grant = intake.SubmitterGrant("nk-bind-executor", frozenset({"bounded-execution"}),
frozenset({"net-kingdom-admins"}))
profile = intake.PresentExisting("service:nk-bind-executor", frozenset({"bounded-execution"}))
validated = intake.validate_submission(env, submitter_id="nk-bind-executor", grant=grant,
present_existing=profile)
self.assertEqual(validated.memo.id, env["memo"]["id"])
class RequestPhase(Base):
def run_cli(self, cfg, admission=ADMISSION):
(self.root / "cfg.json").write_text(json.dumps(cfg))