tegwick
e6252bc0e4
All checks were successful
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
322 lines
16 KiB
Python
322 lines
16 KiB
Python
"""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 = "keycape-nk-bind-executor-approval-bounded-executor-workload-kv-read"
|
|
|
|
|
|
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}
|
|
self.memos = []
|
|
self.memo_status = 201
|
|
|
|
|
|
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/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",
|
|
"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")
|
|
|
|
|
|
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))
|
|
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()
|