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
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = ["PyYAML>=6,<7", "jsonschema>=4,<5"]
|
|
# ///
|
|
"""Journey-only run on the T08 synthetic stack: passes when healthy, fails closed on a broken bind (NK-WP-0056-T10).
|
|
|
|
The same code path the Job runs with `cli.py --journey-only` (`operations.journey_from_kv`). The difference is the
|
|
login: here a dev-OpenBao token that holds only the journey-read policy, instead of Kubernetes auth. Scenarios:
|
|
|
|
healthy the journey passes
|
|
broken bind svc-authelia's password is changed in LLDAP while Authelia keeps the old one (a half-done
|
|
rotation). The journey must fail closed at authelia_first_factor.
|
|
restored the password is set back and the journey passes again
|
|
lane boundary the journey token cannot read the svc-authelia bind
|
|
|
|
Run: uv run bounded-executor/synthetic_journey_only.py [--report FILE]. Exit 0 when every expectation is met.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import pathlib
|
|
import secrets
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE))
|
|
import synthetic_pilot as P # noqa: E402 (loads executor, operations and the T08 stack)
|
|
|
|
OPS, stack_mod = P.OPS, P.stack_mod
|
|
|
|
|
|
def summary(receipt: dict) -> dict:
|
|
return {k: receipt.get(k) for k in ("result", "failed_step", "reason")}
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
|
|
parser.add_argument("--report")
|
|
args = parser.parse_args(argv)
|
|
report = {"schema": "netkingdom.bounded-executor.synthetic-journey-only/v1",
|
|
"started_at": OPS.now_utc().isoformat(), "images": {"openbao": P.OPENBAO_IMAGE},
|
|
"code_path": "operations.journey_from_kv (cli.py --journey-only)", "results": []}
|
|
ok = False
|
|
with tempfile.TemporaryDirectory(prefix="nk56-journey-only-") as tmp:
|
|
stack = stack_mod.Stack(pathlib.Path(tmp), pi_cache_timeout=120)
|
|
bao = None
|
|
secrets_in_report = []
|
|
try:
|
|
stack.up()
|
|
bao = P.OpenBaoDev(stack.network, stack.tag)
|
|
write_token, journey_token = bao.setup(stack.pw["svc-authelia"], stack.pw[stack_mod.USER], stack.seed)
|
|
secrets_in_report = [stack.pw["svc-authelia"], stack.pw[stack_mod.USER], write_token, journey_token, bao.root]
|
|
journey_bao = OPS.OpenBao(bao.addr, token=journey_token)
|
|
state: dict = {}
|
|
|
|
def run():
|
|
return OPS.journey_from_kv(journey_bao, P.JOURNEY_KV, stack.journey_config(), state)
|
|
|
|
healthy = run()
|
|
report["results"].append({"scenario": "healthy", "expected": "pass", "journey": summary(healthy),
|
|
"met": healthy["result"] == "pass"})
|
|
|
|
stack.set_password(stack.lldap_token(), "svc-authelia", secrets.token_urlsafe(24))
|
|
broken = run()
|
|
report["results"].append({"scenario": "broken bind (svc-authelia)",
|
|
"expected": "fail at authelia_first_factor", "journey": summary(broken),
|
|
"met": broken["result"] == "fail"
|
|
and broken["failed_step"] == "authelia_first_factor"})
|
|
|
|
stack.set_password(stack.lldap_token(), "svc-authelia", stack.pw["svc-authelia"])
|
|
restored = run()
|
|
report["results"].append({"scenario": "restored", "expected": "pass", "journey": summary(restored),
|
|
"met": restored["result"] == "pass"})
|
|
|
|
try:
|
|
journey_bao.read(P.BIND_KV)
|
|
denied = False
|
|
except OPS.OperationError:
|
|
denied = True
|
|
report["results"].append({"scenario": "lane boundary", "expected": "journey token cannot read the bind",
|
|
"met": denied})
|
|
journey_bao.revoke_self()
|
|
ok = all(r["met"] for r in report["results"])
|
|
except Exception as error: # report the type only: messages could carry context
|
|
report["error"] = type(error).__name__
|
|
finally:
|
|
if bao:
|
|
bao.remove()
|
|
stack.down()
|
|
text = json.dumps(report, sort_keys=True)
|
|
report["no_values_in_report"] = not any(s and s in text for s in secrets_in_report)
|
|
report["all_met"] = ok and report["no_values_in_report"]
|
|
report["finished_at"] = OPS.now_utc().isoformat()
|
|
out = json.dumps(report, indent=2, sort_keys=True)
|
|
if args.report:
|
|
pathlib.Path(args.report).write_text(out + "\n")
|
|
print(out)
|
|
return 0 if report["all_met"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|