tegwick
45feaa13cc
All checks were successful
ESO role audience + explicit max TTL and ESO policy self-lifecycle (ops-mason R3/R4, railiance-platform R3); consumer role renamed to differ from its policy (railiance-platform R4); default SA and external-secrets/openbao namespaces refused (ops-mason R6); consumer-cannot-read-delivered-secret verification (ops-warden R7, ops-mason R7); excess scope invalid_profile_usage (key-cape R3); attended rotation, broker admission pending, one rollout per case; catalog fields (ops-warden R6). First case consumer_owner_repo rapp-bounded-executor. CCR-2026-0060/0061 names checked by test. Memo r2 validated against informed-decision intake and State Hub dispatcher. 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
256 lines
12 KiB
Python
256 lines
12 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}
|
|
|
|
|
|
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()
|