diff --git a/README.md b/README.md index 49f5f67..dfd9d45 100644 --- a/README.md +++ b/README.md @@ -14,8 +14,11 @@ python3 -m rtut list # id, engine, exercise status, title python3 -m rtut show # steps with owners and done-when python3 -m rtut render -p name=value python3 -m rtut launch # legacy-command packs: start the existing console +python3 -m rtut ui # local web UI on 127.0.0.1:8877: walk a pack, record a run +python3 -m rtut run new|start|verify|done|finish ... # the same run record from the terminal make test ``` -Status: the specification, validator and legacy launch work; run state, evidence -and the UI are planned (`workplans/RBT-WP-0003` and `RBT-WP-0004`). +Status: specification, validator, legacy launch, run records with verification and +exercise-status diff, and a first local web UI work. Terminal walk-through mode and +the console migration are open (`workplans/`). diff --git a/rtut/__main__.py b/rtut/__main__.py index 36ad4e1..6ed3b39 100644 --- a/rtut/__main__.py +++ b/rtut/__main__.py @@ -72,6 +72,9 @@ def main(argv: list[str] | None = None) -> int: sp.add_argument("--yes", action="store_true", help="confirm running a verify command") sp.add_argument("--pass", dest="passed", action="store_true", help="manual check: operator confirms it passed") sp.add_argument("--apply", action="store_true", help="finish: write the exercise-status change") + sp = sub.add_parser("ui", help="serve the local UI on 127.0.0.1") + sp.add_argument("roots", nargs="*", default=["."]) + sp.add_argument("--port", type=int, default=8877) sp = sub.add_parser("launch"); sp.add_argument("manifest") sp.add_argument("--timeout", type=float, default=30.0) sp = sub.add_parser("show"); sp.add_argument("manifest") @@ -79,6 +82,15 @@ def main(argv: list[str] | None = None) -> int: sp.add_argument("--param", "-p", action="append", default=[], help="name=value") a = ap.parse_args(argv) + if a.cmd == "ui": + from . import ui + srv = ui.serve([Path(r) for r in a.roots], a.port) + print(f"http://127.0.0.1:{a.port} (Ctrl-C stops)") + try: + srv.serve_forever() + except KeyboardInterrupt: + pass + return 0 if a.cmd in ("list", "validate"): packs = _packs(a.root) if not packs: diff --git a/rtut/ui.py b/rtut/ui.py new file mode 100644 index 0000000..9c6f772 --- /dev/null +++ b/rtut/ui.py @@ -0,0 +1,195 @@ +"""Local web UI: pick a pack, fill parameters, walk the steps, record evidence. + +Localhost only. Server-rendered forms; the only script is a copy button. +""" +from __future__ import annotations + +import html +import secrets +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +from . import pack as P, run as R + +CSS = """body{font:15px/1.5 system-ui,sans-serif;max-width:52rem;margin:2rem auto;padding:0 1rem;color:#111;background:#fff} +@media(prefers-color-scheme:dark){body{color:#eee;background:#111}pre,input,select{background:#1b1b1b;color:#eee}} +h1,h2{font-weight:600}a{color:inherit}hr{border:0;border-top:1px solid #8886} +.eyebrow{font:12px ui-monospace,monospace;text-transform:uppercase;letter-spacing:.06em} +.step{border:1px solid #8886;padding:.8rem 1rem;margin:.8rem 0}.step.running{border-color:currentColor} +.pill{font:12px ui-monospace,monospace;border:1px solid #8886;padding:0 .4rem} +pre{border:1px solid #8886;padding:.5rem;overflow:auto;margin:.4rem 0} +button,input,select{font:inherit;padding:.25rem .5rem}.note{opacity:.75}.bad{font-weight:600}""" + + +def page(title: str, body: str) -> bytes: + return (f"" + f"{html.escape(title)}" + f"

runbook-tutorials

{body}" + "").encode() + + +def e(s) -> str: + return html.escape(str(s)) + + +class App: + def __init__(self, roots: list[Path]): + self.token = secrets.token_urlsafe(16) + self.packs = [P.load(p) for r in roots for p in P.discover(r)] + + def hidden(self) -> str: + return f"" + + def index(self) -> bytes: + rows = [] + for i, p in enumerate(self.packs): + d = p.data + if not p.ok: + rows.append(f"
  • {e(p.id)}: invalid ({e(p.errors[0])})
    • ") + elif d["engine"] != "native": + rows.append(f"
  • {e(d['title'])} legacy console " + f"start with: python -m rtut launch {e(p.path)}
    • ") + else: + rows.append(f"
  • {e(d['title'])} " + f"{e(d['exercise_status'])}
    • ") + return page("Runbooks", f"

    Runbooks

      {''.join(rows) or '
    • No packs found.
      • '}
    ") + + def pack_page(self, i: int) -> bytes: + p = self.packs[i]; d = p.data + prereq = "".join(f"
  • [{e(x['owner'])}] {e(x['text'])}
    • " for x in d.get("prerequisites", [])) + fields = "".join( + f"


    {e(x.get('help', ''))} Never enter secrets.

    " + for x in d.get("parameters", [])) + return page(d["title"], f"

    {e(d['title'])}

    {e(d['exercise_status'])}

    " + f"

    {e(d.get('outcome', ''))}

    Prerequisites

      {prereq}
    " + f"
    {self.hidden()}{fields}" + "

    " + "
    ") + + def run_page(self, i: int, rid: str, msg: str = "") -> bytes: + p = self.packs[i]; r = R.load(p, rid); d = p.data + out = [f"

    {e(d['title'])}

    {e(d['exercise_status'])}

    " + f"

    run {e(rid)} · operator {e(r.data['operator'])}

    "] + if msg: + out.append(f"

    {e(msg)}

    ") + blocked = False + for s in d["steps"]: + st = r.state(s["id"]); state = st["state"] + act = "" + cmd = R.command_for(r, s["id"]) + block = f"
    {e(cmd)}
    " if cmd else "" + url = f"/run/{i}/{e(rid)}/{e(s['id'])}" + v = st.get("verification") + if state == "todo" and not blocked: + act = f"
    {self.hidden()}
    " + elif state == "running": + ver = s["verify"] + if ver.get("expect", "manual") == "manual": + act = (f"
    {self.hidden()}
    ") + else: + act = (f"
    {self.hidden()}" + f"

    Runs: {e(R.render(ver['command'], r.data['params'], p))}

    " + f"
    ") + if v: + act += f"

    {'✔ passed' if v['passed'] else '✘ failed'}

    " + if v and v["passed"]: + ev = "".join(f" " for n in s.get("evidence", [])) + act += f"
    {self.hidden()}{ev}
    " + act += (f"
    {self.hidden()} " + "
    ") + elif state == "failed": + act = f"

    Failed: {e(st.get('note', ''))}

    {self.hidden()}
    " + if state not in ("done", "skipped"): + blocked = True + rb = f"

    Rollback: {e(s['rollback'])}

    " if s.get("rollback") else "" + out.append(f"
    {state} {e(s['title'])} " + f"[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}{block}" + f"

    Done when: {e(s['verify']['done_when'])}

    {rb}{act}
    ") + if r.complete: + out.append("

    All steps done. To record this run as exercising the pack, review and apply: " + f"python -m rtut run finish {e(p.path)} {e(rid)} --apply

    ") + out.append("

    Threat checks

      " + "".join(f"
    • {e(t)}
      • " for t in d["threat_checks"]) + "
    ") + return page(d["title"], "".join(out)) + + +def make_handler(app: App, port_holder: list): + class H(BaseHTTPRequestHandler): + def log_message(self, *a): # quiet + pass + + def _send(self, body: bytes, code=200): + self.send_response(code) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; form-action 'self'") + self.end_headers(); self.wfile.write(body) + + def _host_ok(self) -> bool: + return (self.headers.get("Host", "").split(":")[0] in ("127.0.0.1", "localhost")) + + def do_GET(self): + if not self._host_ok(): + return self._send(b"forbidden", 403) + parts = [x for x in urlparse(self.path).path.split("/") if x] + try: + if not parts: + return self._send(app.index()) + if parts[0] == "pack": + return self._send(app.pack_page(int(parts[1]))) + if parts[0] == "run": + return self._send(app.run_page(int(parts[1]), parts[2])) + except (IndexError, ValueError, R.RunError): + pass + self._send(page("Not found", "

    Not found.

    "), 404) + + def do_POST(self): + if not self._host_ok(): + return self._send(b"forbidden", 403) + n = int(self.headers.get("Content-Length", 0)) + form = {k: v[0] for k, v in parse_qs(self.rfile.read(n).decode()).items()} + if form.get("_t") != app.token: + return self._send(b"bad token", 403) + parts = [x for x in urlparse(self.path).path.split("/") if x] + msg = "" + try: + if parts[:2] == ["run", "new"]: + i = int(parts[2]); p = app.packs[i] + params = {k[2:]: v for k, v in form.items() if k.startswith("p_")} + r = R.create(p, form.get("operator", ""), params) + return self._redirect(f"/run/{i}/{r.data['run_id']}") + i, rid, sid, action = int(parts[1]), parts[2], parts[3], parts[4] + p = app.packs[i]; r = R.load(p, rid) + if action == "start": R.start(r, sid) + elif action == "verify": + R.verify(r, sid, confirm=bool(form.get("yes")), manual_ok=bool(form.get("ok")) if not form.get("yes") else None) + elif action == "done": + R.complete(r, sid, {k[2:]: v for k, v in form.items() if k.startswith("e_")}) + elif action == "fail": R.fail(r, sid, form.get("note", "")) + elif action == "retry": R.retry(r, sid) + return self._redirect(f"/run/{i}/{rid}") + except (R.RunError, KeyError, ValueError, IndexError) as ex: + msg = str(ex) + try: + if parts[:2] == ["run", "new"]: + return self._send(app.pack_page(int(parts[2])).replace(b"

    ", f"

    {e(msg)}

    ".encode(), 1), 400) + return self._send(app.run_page(int(parts[1]), parts[2], msg), 400) + except Exception: + return self._send(page("Error", f"

    {e(msg)}

    "), 400) + + def _redirect(self, to: str): + self.send_response(303); self.send_header("Location", to); self.end_headers() + + return H + + +def serve(roots: list[Path], port: int = 8877) -> ThreadingHTTPServer: + app = App(roots) + srv = ThreadingHTTPServer(("127.0.0.1", port), make_handler(app, [])) + srv.app = app # type: ignore[attr-defined] + return srv diff --git a/tests/test_ui.py b/tests/test_ui.py new file mode 100644 index 0000000..80b554a --- /dev/null +++ b/tests/test_ui.py @@ -0,0 +1,68 @@ +import http.client +import re +import shutil +import threading +from pathlib import Path +from urllib.parse import urlencode + +import pytest + +from rtut import ui + +ROOT = Path(__file__).resolve().parent.parent + + +@pytest.fixture +def srv(tmp_path): + d = tmp_path / "runbooks" / "hello"; d.mkdir(parents=True) + shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", d / "pack.yaml") + s = ui.serve([tmp_path], 0) + threading.Thread(target=s.serve_forever, daemon=True).start() + yield s + s.shutdown() + + +def req(s, method, path, form=None, host="127.0.0.1"): + c = http.client.HTTPConnection("127.0.0.1", s.server_address[1]) + body = urlencode(form) if form is not None else None + c.request(method, path, body, {"Host": host, "Content-Type": "application/x-www-form-urlencoded"}) + r = c.getresponse() + return r.status, r.getheader("Location"), r.read().decode() + + +def test_index_lists_pack(srv): + st, _, body = req(srv, "GET", "/") + assert st == 200 and "Example - check a host answers" in body + + +def test_foreign_host_and_missing_token_refused(srv): + assert req(srv, "GET", "/", host="evil.example")[0] == 403 + assert req(srv, "POST", "/run/new/0", {"operator": "bernd"})[0] == 403 + + +def test_walk_a_run(srv): + t = srv.app.token + st, loc, _ = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "localhost", "p_count": "1"}) + assert st == 303 and loc.startswith("/run/0/") + assert req(srv, "POST", loc + "/ping/start", {"_t": t})[0] == 303 + st, _, body = req(srv, "GET", loc) + assert "ping -c 1 localhost" in body and "Run check" in body + req(srv, "POST", loc + "/ping/verify", {"_t": t, "yes": "1"}) + st, _, body = req(srv, "GET", loc) + if "passed" in body and "failed" not in body.split("Threat")[0].replace("Mark failed", ""): + assert req(srv, "POST", loc + "/ping/done", {"_t": t})[0] == 303 + assert "All steps done" in req(srv, "GET", loc)[2] + + +def test_unsafe_parameter_rejected_and_escaped(srv): + t = srv.app.token + st, _, body = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "