Add local web UI for native packs (RBT-WP-0004-T02)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 295952@bnt-lap001
Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477

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:

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"<!doctype html><meta charset=utf-8><meta name=viewport content='width=device-width'>"
+            f"<title>{html.escape(title)}</title><style>{CSS}</style>"
+            f"<p class=eyebrow><a href='/'>runbook-tutorials</a></p>{body}"
+            "<script>document.querySelectorAll('[data-copy]').forEach(b=>b.onclick=()=>"
+            "navigator.clipboard.writeText(b.previousElementSibling.textContent))</script>").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"<input type=hidden name=_t value='{self.token}'>"
+
+    def index(self) -> bytes:
+        rows = []
+        for i, p in enumerate(self.packs):
+            d = p.data
+            if not p.ok:
+                rows.append(f"<li class=bad>{e(p.id)}: invalid ({e(p.errors[0])})</li>")
+            elif d["engine"] != "native":
+                rows.append(f"<li>{e(d['title'])} <span class=pill>legacy console</span> "
+                            f"<span class=note>start with: python -m rtut launch {e(p.path)}</span></li>")
+            else:
+                rows.append(f"<li><a href='/pack/{i}'>{e(d['title'])}</a> "
+                            f"<span class=pill>{e(d['exercise_status'])}</span></li>")
+        return page("Runbooks", f"<h1>Runbooks</h1><ul>{''.join(rows) or '<li>No packs found.</li>'}</ul>")
+
+    def pack_page(self, i: int) -> bytes:
+        p = self.packs[i]; d = p.data
+        prereq = "".join(f"<li>[{e(x['owner'])}] {e(x['text'])}</li>" for x in d.get("prerequisites", []))
+        fields = "".join(
+            f"<p><label>{e(x['label'])}<br>"
+            + (f"<select name='p_{x['id']}'>" + "".join(
+                f"<option{' selected' if o == x.get('default') else ''}>{e(o)}</option>" for o in x["values"]) + "</select>"
+               if x.get("values") else f"<input name='p_{x['id']}' value='{e(x.get('default', ''))}'>")
+            + f"</label><br><span class=note>{e(x.get('help', ''))} Never enter secrets.</span></p>"
+            for x in d.get("parameters", []))
+        return page(d["title"], f"<h1>{e(d['title'])}</h1><p class=pill>{e(d['exercise_status'])}</p>"
+                    f"<p>{e(d.get('outcome', ''))}</p><h2>Prerequisites</h2><ul>{prereq}</ul>"
+                    f"<form method=post action='/run/new/{i}'>{self.hidden()}{fields}"
+                    "<p><label>Your name (one word)<br><input name=operator required></label></p>"
+                    "<button>Start run</button></form>")
+
+    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"<h1>{e(d['title'])}</h1><p class=pill>{e(d['exercise_status'])}</p>"
+               f"<p class=note>run {e(rid)} · operator {e(r.data['operator'])}</p>"]
+        if msg:
+            out.append(f"<p class=bad>{e(msg)}</p>")
+        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"<pre>{e(cmd)}</pre><button type=button data-copy>Copy</button>" if cmd else ""
+            url = f"/run/{i}/{e(rid)}/{e(s['id'])}"
+            v = st.get("verification")
+            if state == "todo" and not blocked:
+                act = f"<form method=post action='{url}/start'>{self.hidden()}<button>Start</button></form>"
+            elif state == "running":
+                ver = s["verify"]
+                if ver.get("expect", "manual") == "manual":
+                    act = (f"<form method=post action='{url}/verify'>{self.hidden()}<label><input type=checkbox name=ok value=1> "
+                           f"I confirm: {e(ver['done_when'])}</label> <button>Record check</button></form>")
+                else:
+                    act = (f"<form method=post action='{url}/verify'>{self.hidden()}<input type=hidden name=yes value=1>"
+                           f"<p class=note>Runs: <code>{e(R.render(ver['command'], r.data['params'], p))}</code></p>"
+                           f"<button>Run check</button></form>")
+                if v:
+                    act += f"<p>{'✔ passed' if v['passed'] else '✘ failed'}</p>"
+                if v and v["passed"]:
+                    ev = "".join(f"<label>{e(n)} <input name='e_{e(n)}' required></label> " for n in s.get("evidence", []))
+                    act += f"<form method=post action='{url}/done'>{self.hidden()}{ev}<button>Mark done</button></form>"
+                act += (f"<form method=post action='{url}/fail'>{self.hidden()}<input name=note placeholder='what went wrong (no secrets)'> "
+                        "<button>Mark failed</button></form>")
+            elif state == "failed":
+                act = f"<p class=bad>Failed: {e(st.get('note', ''))}</p><form method=post action='{url}/retry'>{self.hidden()}<button>Retry</button></form>"
+            if state not in ("done", "skipped"):
+                blocked = True
+            rb = f"<p class=note>Rollback: {e(s['rollback'])}</p>" if s.get("rollback") else ""
+            out.append(f"<div class='step {state}'><span class=pill>{state}</span> <b>{e(s['title'])}</b> "
+                       f"<span class=note>[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}</span>{block}"
+                       f"<p class=note>Done when: {e(s['verify']['done_when'])}</p>{rb}{act}</div>")
+        if r.complete:
+            out.append("<p><b>All steps done.</b> To record this run as exercising the pack, review and apply: "
+                       f"<code>python -m rtut run finish {e(p.path)} {e(rid)} --apply</code></p>")
+        out.append("<h2>Threat checks</h2><ul>" + "".join(f"<li>{e(t)}</li>" for t in d["threat_checks"]) + "</ul>")
+        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", "<p>Not found.</p>"), 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"<h1>", f"<p class=bad>{e(msg)}</p><h1>".encode(), 1), 400)
+                return self._send(app.run_page(int(parts[1]), parts[2], msg), 400)
+            except Exception:
+                return self._send(page("Error", f"<p class=bad>{e(msg)}</p>"), 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