From c2a0f2f4e049f721f2d3e8d66481d782ffd6e3c9 Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 5 Nov 2026 00:39:10 +0100 Subject: [PATCH] NK-WP-0056-T06: executor waits for every endpoint; errors named by phase, never mislabelled as no-effect Two live runs of BEA-2026-0002 ended setup:URLError while four read-only diagnostic pods (same image, labels, mounts and config) reached every endpoint. Likely cause: a fresh Job pod reaches OpenBao before the sso/approval-engine ingress rules admit it (NetworkPolicy is programmed per destination). Now: - build waits until OpenBao, KeyCape, approval-engine, audit-core and LLDAP all answer HTTP (one shared 30 s budget); an unreachable one is named: setup: not reachable. - a preflight error is a refusal named preflight_error: (preflight has no effect). - an error in the consume read-back is outcome unknown, consume_unconfirmed:, alerted; before, any exception after setup was reported as 'refused, no effect' even after consume. - any other escaping error after setup is outcome unknown (exit 5), never 'no effect'. 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/cli.py | 27 +++++++++-- bounded-executor/executor.py | 7 +++ bounded-executor/operations.py | 20 +++++++-- bounded-executor/tests/test_executor.py | 24 ++++++++++ bounded-executor/tests/test_openbao_lanes.py | 47 ++++++++++++++++++-- 5 files changed, 116 insertions(+), 9 deletions(-) diff --git a/bounded-executor/cli.py b/bounded-executor/cli.py index a81877b..9cd683e 100644 --- a/bounded-executor/cli.py +++ b/bounded-executor/cli.py @@ -54,6 +54,7 @@ import pathlib import signal import sys import threading +import urllib.parse HERE = pathlib.Path(__file__).resolve().parent @@ -106,6 +107,19 @@ def distinct_roles(cfg: dict) -> None: raise ValueError("every lane must use its own OpenBao role") +def run_endpoints(cfg: dict) -> dict: + """Every service the run talks to, with a path that answers HTTP without credentials (any status counts).""" + ends = {"openbao": (cfg["openbao"]["addr"], "/v1/sys/health"), + "lldap": (cfg["lldap"]["url"], "/"), + "audit-core": (cfg["audit_core"]["url"], "/healthz"), + "approval-engine": (cfg["approval_engine"]["url"], "/healthz")} + client = cfg["approval_engine"].get("client") + if client: # the KeyCape token endpoint answers a GET with an HTTP error, which proves the path + token = urllib.parse.urlsplit(client["token_url"]) + ends["keycape"] = (f"{token.scheme}://{token.netloc}", "/healthz") + return ends + + def build(cfg: dict, logins: list): """`logins` collects every OpenBao client as soon as it exists, so the caller revokes them all, even when a later step of setup fails.""" @@ -114,7 +128,7 @@ def build(cfg: dict, logins: list): raise ValueError("the journey read and the bind write must use separate roles") distinct_roles(cfg) - OPS.wait_reachable(o["addr"], ca_file=o.get("ca_file")) # new pod: NetworkPolicy may not be applied yet + OPS.wait_all_reachable(run_endpoints(cfg)) # new pod: NetworkPolicy is programmed per destination def login(role: str): client = OPS.OpenBao(o["addr"], k8s_role=role, jwt_file=o["jwt_file"], ca_file=o.get("ca_file")) @@ -336,9 +350,16 @@ def main(argv=None) -> int: logins: list = [] try: config, deps = build(cfg, logins) - result = X.Run(admission, args.approval_id, config, deps).execute() except Exception as error: # setup failed before preflight: nothing was consumed - result = {"outcome": "refused", "reason": "setup:" + type(error).__name__, "run_id": None, "effects": False} + where = str(error) if isinstance(error, OPS.Unreachable) else type(error).__name__ + result = {"outcome": "refused", "reason": "setup:" + where, "run_id": None, "effects": False} + else: + run = X.Run(admission, args.approval_id, config, deps) + try: + result = run.execute() # labels its own preflight and consume errors + except Exception as error: # anything else after setup: the run's effect is not established + result = {"outcome": "unknown", "reason": "run:" + type(error).__name__, "run_id": run.run_id, + "effects": "unknown"} finally: for client in logins: client.revoke_self() diff --git a/bounded-executor/executor.py b/bounded-executor/executor.py index d320ff2..d194ced 100644 --- a/bounded-executor/executor.py +++ b/bounded-executor/executor.py @@ -354,6 +354,9 @@ class Run: self.preflight() except Refused as refusal: return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False} + except Exception as error: # preflight has no effect: an error there is a refusal, named by its type + return {"outcome": "refused", "reason": "preflight_error:" + type(error).__name__, "run_id": None, + "effects": False} if self.d.should_stop(): # the founder's stop before the one irreversible act: nothing is spent return {"outcome": "refused", "reason": "stopped_before_consume", "run_id": None, "effects": False} self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic() @@ -362,6 +365,10 @@ class Run: except Refused as refusal: # consumed or not, nothing else may happen now self.d.alert("consume_unconfirmed", {"admission": self.spec["id"], "run_id": self.run_id}) return {"outcome": "refused", "reason": str(refusal), "run_id": self.run_id, "effects": "consume-only"} + except Exception as error: # the read-back failed: whether the approval is spent is not established + self.d.alert("consume_unconfirmed", {"admission": self.spec["id"], "run_id": self.run_id}) + return {"outcome": "unknown", "reason": "consume_unconfirmed:" + type(error).__name__, + "run_id": self.run_id, "effects": "consume-unknown"} try: self.d.audit.emit(self._event(STARTED, { "receipt_version": RECEIPT_VERSION, "run_id": self.run_id, "admission": self.spec["id"], diff --git a/bounded-executor/operations.py b/bounded-executor/operations.py index 2148c47..52c019b 100644 --- a/bounded-executor/operations.py +++ b/bounded-executor/operations.py @@ -62,7 +62,8 @@ def _opener(ca_file: str | None): def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_file: str | None = None, - opener=None, clock=time.monotonic, sleep=time.sleep) -> float: + opener=None, clock=time.monotonic, sleep=time.sleep, path: str = "/v1/sys/health", + name: str = "openbao") -> float: """Wait until OpenBao answers on the network, before the first login. k3s applies NetworkPolicy to a new pod a moment after it starts, and connections in that window are refused (seen live 2026-10-31: refused at 0 s, open at 3 s). Polls the unauthenticated sys/health. Any HTTP answer, including 429/473/503, counts as reachable; @@ -70,7 +71,7 @@ def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_fi open_ = opener or _opener(ca_file).open start = clock() while True: - request = urllib.request.Request(addr.rstrip("/") + "/v1/sys/health", method="GET") + request = urllib.request.Request(addr.rstrip("/") + path, method="GET") try: with open_(request, timeout=5): return clock() - start @@ -79,10 +80,23 @@ def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_fi return clock() - start except (urllib.error.URLError, OSError): if clock() - start >= timeout: - raise Unreachable("openbao not reachable") from None + raise Unreachable(name + " not reachable") from None sleep(interval) +def wait_all_reachable(endpoints: dict, *, timeout: float = 30, **kwargs) -> dict: + """Wait until every named endpoint answers HTTP at all (any status counts), within one shared timeout. A new + pod's NetworkPolicy is programmed per destination: OpenBao can already be reachable while the sso and + approval-engine ingress rules do not yet admit the pod (NK-WP-0056-T06, two live runs refused setup:URLError). + `endpoints` maps name -> (base url, path). Returns seconds waited per endpoint; raises Unreachable(name).""" + clock = kwargs.get("clock", time.monotonic) + start, waited = clock(), {} + for name, (base, path) in endpoints.items(): + left = max(timeout - (clock() - start), 0.5) + waited[name] = round(wait_reachable(base, timeout=left, path=path, name=name, **kwargs), 2) + return waited + + class Unreachable(Exception): pass diff --git a/bounded-executor/tests/test_executor.py b/bounded-executor/tests/test_executor.py index 7cb5fe2..b18dcbf 100644 --- a/bounded-executor/tests/test_executor.py +++ b/bounded-executor/tests/test_executor.py @@ -402,6 +402,30 @@ class FailureTests(Base): self.assertIn("restore-kv", self.ops()) self.assertNotIn("deliver", self.ops()) + def test_preflight_error_is_a_named_refusal(self): + def broken(_id): + raise ConnectionRefusedError() + self.w.carrier.get = broken + result = self.run_it() + self.assertEqual((result["outcome"], result["reason"], result["effects"]), + ("refused", "preflight_error:ConnectionRefusedError", False)) + self.assertEqual(self.w.carrier.consumes, 0) + + def test_consume_read_back_error_is_unknown_never_no_effect(self): + real_get, calls = self.w.carrier.get, [] + def get(approval_id): + calls.append(approval_id) + if len(calls) > 1: # preflight reads once; the consume read-back fails + raise ConnectionResetError() + return real_get(approval_id) + self.w.carrier.get = get + self.w.carrier.raise_on_consume = True + result = self.run_it() + self.assertEqual((result["outcome"], result["reason"], result["effects"]), + ("unknown", "consume_unconfirmed:ConnectionResetError", "consume-unknown")) + self.assertIn("consume_unconfirmed", self.w.alerts) + self.assertEqual(self.w.events, []) + def test_stop_before_consume_spends_nothing(self): self.w.stop_after = 0 result = self.run_it() diff --git a/bounded-executor/tests/test_openbao_lanes.py b/bounded-executor/tests/test_openbao_lanes.py index db0a471..a285122 100644 --- a/bounded-executor/tests/test_openbao_lanes.py +++ b/bounded-executor/tests/test_openbao_lanes.py @@ -92,14 +92,19 @@ class CliLanes(Base): return code, json.loads(out.getvalue()) def cfg(self, journey_role="journey-read"): - # lldap is missing on purpose: setup fails after both logins + # every endpoint answers (the fake answers any GET); setup fails after both logins, at the Kubernetes client + # (no in-cluster service-account CA outside a pod) return {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")}, - "journey": {"kv": "x", "k8s_role": journey_role, "config": {}}} + "journey": {"kv": "x", "k8s_role": journey_role, "config": {}}, + "lldap": {"url": self.addr}, "audit_core": {"url": self.addr, "token_file": str(self.root / "jwt")}, + "approval_engine": {"url": self.addr, "token_file": str(self.root / "jwt")}} def test_two_lanes_two_logins_both_revoked_when_setup_fails(self): code, result = self.run_cli(self.cfg()) - self.assertEqual((code, result["reason"]), (4, "setup:KeyError")) + self.assertEqual((code, result["reason"]), (4, "setup:FileNotFoundError")) self.assertEqual([role for role, _ in self.bao.logins], ["bind-write", "journey-read"]) + for path in ("/v1/sys/health", "/", "/healthz"): # every endpoint was waited for before any login + self.assertIn(path, [p for p, _ in self.bao.reads]) self.assertEqual(sorted(self.bao.revoked), ["tok-bind-write-1", "tok-journey-read-2"]) def test_one_role_for_both_lanes_is_refused_before_any_login(self): @@ -244,3 +249,39 @@ class JourneyUsesItsOwnLane(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class EveryEndpointReachable(Base): + def test_unreachable_endpoint_is_named_and_nothing_logs_in(self): + closed = socket_port() + cfg = {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")}, + "journey": {"kv": "x", "k8s_role": "journey-read", "config": {}}, + "lldap": {"url": self.addr}, "audit_core": {"url": self.addr, "token_file": "x"}, + "approval_engine": {"url": self.addr, "client": {"token_url": f"http://127.0.0.1:{closed}/token", + "client_id": "c", "k8s_role": "a", "kv": "k"}}} + original = OPS.wait_reachable + self.addCleanup(setattr, OPS, "wait_reachable", original) + OPS.wait_reachable = lambda *a, **k: original(*a, **{**k, "timeout": 0.2, "interval": 0.05}) + (self.root / "cfg.json").write_text(json.dumps(cfg)) + admission = CLI.X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml" + with contextlib.redirect_stdout(io.StringIO()) as out: + code = CLI.main(["--admission", str(admission), "--approval-id", "apr-1", "--config", str(self.root / "cfg.json")]) + result = json.loads(out.getvalue()) + self.assertEqual((code, result["reason"], result["effects"]), (4, "setup:keycape not reachable", False)) + self.assertEqual(self.bao.logins, []) + + def test_endpoints_cover_every_service_the_run_uses(self): + cfg = {"openbao": {"addr": "http://bao:8200"}, "lldap": {"url": "http://lldap:17170"}, + "audit_core": {"url": "http://audit:8080"}, + "approval_engine": {"url": "http://ae:8080", "client": {"token_url": "http://kc:8080/token"}}} + self.assertEqual(CLI.run_endpoints(cfg), { + "openbao": ("http://bao:8200", "/v1/sys/health"), "lldap": ("http://lldap:17170", "/"), + "audit-core": ("http://audit:8080", "/healthz"), "approval-engine": ("http://ae:8080", "/healthz"), + "keycape": ("http://kc:8080", "/healthz")}) + + +def socket_port(): + import socket + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1]