"""One OpenBao login per lane, renewal before the lease ends, revocation on every path (NK-WP-0056-T10). Fakes only.""" from __future__ import annotations import contextlib import importlib.util import io import json import pathlib import sys import tempfile import threading import time import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer 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") OPS = sys.modules["operations"] class Bao: def __init__(self, lease=900): self.lease = lease self.logins, self.revoked, self.reads = [], [], [] self.n = 0 def serve(bao: Bao): class H(BaseHTTPRequestHandler): def _send(self, body): raw = json.dumps(body).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def do_POST(self): body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}") if self.path == "/v1/auth/kubernetes/login": bao.n += 1 bao.logins.append((body["role"], body["jwt"])) return self._send({"auth": {"client_token": f"tok-{body['role']}-{bao.n}", "lease_duration": bao.lease}}) if self.path == "/v1/auth/token/revoke-self": bao.revoked.append(self.headers.get("X-Vault-Token")) return self._send({}) self._send({}) def do_GET(self): bao.reads.append((self.path, self.headers.get("X-Vault-Token"))) self._send({"data": {"data": {"PASSWORD": "p", "TOTP_SEED": "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"}, "metadata": {"version": 3}}}) 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.bao = Bao() self.server = serve(self.bao) 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}" class CliLanes(Base): def run_cli(self, cfg): (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")]) return code, json.loads(out.getvalue()) def cfg(self, journey_role="journey-read"): # lldap is missing on purpose: setup fails after both logins return {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")}, "journey": {"kv": "x", "k8s_role": journey_role, "config": {}}} 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([role for role, _ in self.bao.logins], ["bind-write", "journey-read"]) 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): code, result = self.run_cli(self.cfg(journey_role="bind-write")) self.assertEqual((code, result["reason"]), (4, "setup:ValueError")) self.assertEqual(self.bao.logins, []) class JourneyOnly(Base): SEED = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ" def setUp(self): super().setUp() self.calls = [] original = (OPS.journey.run_journey, OPS._load_config) self.addCleanup(lambda: (setattr(OPS.journey, "run_journey", original[0]), setattr(OPS, "_load_config", original[1]))) OPS._load_config = lambda cfg: cfg self.result = "pass" def run_journey(cfg, password, seed, min_time_step=None): self.calls.append(cfg) return {"result": self.result, "failed_step": None if self.result == "pass" else "authelia_first_factor", "reason": None, "otp_time_step": 1} OPS.journey.run_journey = run_journey def cfg(self, **journey): # write role, approval-engine, audit-core, lldap and kubernetes are absent on purpose: not needed return {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")}, "journey": {"kv": "platform/workloads/x", "k8s_role": "journey-read", "config": {"c": 1}, **journey}} def run_cli(self, cfg, *extra): (self.root / "cfg.json").write_text(json.dumps(cfg)) with contextlib.redirect_stdout(io.StringIO()) as out: code = CLI.main(["--journey-only", "--config", str(self.root / "cfg.json"), *extra]) return code, json.loads(out.getvalue()) def test_pass_logs_in_with_the_journey_role_only_and_revokes(self): code, result = self.run_cli(self.cfg()) self.assertEqual((code, result["outcome"], result["effects"]), (0, "pass", False)) self.assertEqual([role for role, _ in self.bao.logins], ["journey-read"]) self.assertEqual(self.bao.revoked, ["tok-journey-read-1"]) self.assertEqual(self.calls, [{"c": 1}]) self.assertNotIn(self.SEED, json.dumps(result)) def test_failed_journey_fails_closed_with_exit_1(self): self.result = "fail" code, result = self.run_cli(self.cfg()) self.assertEqual((code, result["outcome"], result["journey"]["failed_step"]), (1, "fail", "authelia_first_factor")) def test_incomplete_config_is_refused_before_any_login(self): cfg = self.cfg() del cfg["journey"]["k8s_role"] code, result = self.run_cli(cfg) self.assertEqual((code, result["reason"]), (4, "setup:KeyError")) self.assertEqual(self.bao.logins, []) def test_admission_arguments_conflict_with_journey_only(self): with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()): self.run_cli(self.cfg(), "--approval-id", "apr-1") class WaitReachable(unittest.TestCase): """k3s applies NetworkPolicy to a new pod a moment late: early connections are refused (live, 2026-10-31).""" def fake(self, outcomes): self.now, self.slept, calls = 0.0, [], list(outcomes) def open_(request, timeout): outcome = calls.pop(0) if isinstance(outcome, BaseException): raise outcome return contextlib.nullcontext() def sleep(seconds): self.slept.append(seconds) self.now += seconds return dict(opener=open_, clock=lambda: self.now, sleep=sleep) def test_refused_then_reachable(self): refused = OPS.urllib.error.URLError(ConnectionRefusedError()) waited = OPS.wait_reachable("http://bao:8200", **self.fake([refused, refused, "ok"])) self.assertEqual((waited, self.slept), (2.0, [1, 1])) def test_any_http_answer_is_reachable(self): sealed = OPS.urllib.error.HTTPError("http://bao/v1/sys/health", 503, "sealed", {}, io.BytesIO(b"")) self.assertEqual(OPS.wait_reachable("http://bao:8200", **self.fake([sealed])), 0.0) def test_gives_up_after_the_timeout(self): refused = OPS.urllib.error.URLError(ConnectionRefusedError()) with self.assertRaises(OPS.Unreachable): OPS.wait_reachable("http://bao:8200", timeout=3, **self.fake([refused] * 10)) self.assertEqual(self.slept, [1, 1, 1]) class Renewal(Base): def test_logs_in_again_before_the_lease_ends_and_revokes_the_old_token(self): self.bao.lease = 1 # renew_at = max(1 - margin, 0.5 s) client = OPS.OpenBao(self.addr, k8s_role="journey-read", jwt_file=str(self.root / "jwt")) client.read("platform/workloads/x") time.sleep(0.6) client.read("platform/workloads/x") self.assertEqual(len(self.bao.logins), 2) self.assertEqual(self.bao.revoked, ["tok-journey-read-1"]) self.assertEqual([tok for _, tok in self.bao.reads], ["tok-journey-read-1", "tok-journey-read-2"]) def test_a_long_lease_is_not_renewed(self): client = OPS.OpenBao(self.addr, k8s_role="r", jwt_file=str(self.root / "jwt")) client.read("platform/workloads/x") client.read("platform/workloads/x") self.assertEqual(len(self.bao.logins), 1) def test_a_given_token_never_logs_in(self): client = OPS.OpenBao(self.addr, token="static") client.read("platform/workloads/x") client.revoke_self() self.assertEqual((self.bao.logins, self.bao.revoked), ([], ["static"])) class JourneyUsesItsOwnLane(unittest.TestCase): def test_journey_check_reads_only_through_the_journey_client(self): class Recorder: def __init__(self, name): self.name, self.reads = name, [] def read(self, path): self.reads.append(path) return {"PASSWORD": "p", "TOTP_SEED": "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"}, 1 write, read = Recorder("write"), Recorder("read") _, checks = OPS.pilot_registry(bao=write, journey_bao=read, directory=None, delivery=None, restart=None, bind_kv="b", bind_user="svc-authelia", journey_kv="j", journey_config={}) original = (OPS.journey.run_journey, OPS._load_config) OPS.journey.run_journey = lambda cfg, pw, seed, min_time_step=None: {"result": "pass"} OPS._load_config = lambda cfg: cfg try: self.assertTrue(checks["synthetic-journey"]({}, type("Ctx", (), {"data": {}})())) finally: OPS.journey.run_journey, OPS._load_config = original self.assertEqual((write.reads, read.reads), ([], ["j"])) if __name__ == "__main__": unittest.main()