tegwick
abed4360d4
cli.py --journey-only: read-only, no admission; journey-read role only (CCR-2026-0059), no Kubernetes/directory client, no approval or audit call; exit 0 pass / 1 fail closed / 4 refused. Shared operations.journey_from_kv with the admission check. Synthetic proof: healthy pass, broken svc-authelia bind fails closed at authelia_first_factor, restored pass, journey token cannot read the bind; all_met, no values. 45 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
213 lines
9 KiB
Python
213 lines
9 KiB
Python
"""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 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()
|