runbook-tutorials/rtut/ui.py

"""Local web UI: pick a pack, fill parameters, walk the steps, record evidence.

Localhost only. Server-rendered forms; the only script is small: copy button, fail-comment warning, scroll to current step.
"""
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}.head{display:inline-flex;width:calc(100% - 1.6rem);justify-content:space-between;align-items:baseline;gap:1rem;vertical-align:top}summary{cursor:pointer}.row{display:flex;gap:.5rem;align-items:center;flex-wrap:wrap}.grow{flex:1;min-width:12rem}[hidden]{display:none}#current{scroll-margin-top:1rem}.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));"
            # first click on 'Mark failed' without a comment only warns; a second click fails the step without one
            "document.querySelectorAll('[data-fail]').forEach(b=>b.addEventListener('click',ev=>{"
            "const f=b.closest('form'),n=f.querySelector('[name=note]'),w=f.querySelector('[data-warn]');"
            "if(n&&!n.value.trim()&&!b.dataset.warned){ev.preventDefault();b.dataset.warned='1';w.hidden=false;n.focus();}}));"
            # keep the operator on the current step after every action
            "const c=document.getElementById(location.hash==='#feedback'?'feedback':location.hash==='#close'?'close':'current');if(c)c.scrollIntoView({block:'center'});"
            "</script>").encode()


def e(s) -> str:
    return html.escape(str(s))


SMILEYS = (("good", "🙂", "Went well"), ("neutral", "😐", "So-so"), ("poor", "🙁", "Poor"))


def score_line(c: dict) -> str:
    return (f"tried {c['tried']} · succeeded {c['succeeded']} · failed {c['failed']} · "
            f"abandoned {c['abandoned']} · stalled {c['stalled']}")


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.get("spec") != "runbook-pack/v0.1" and d["engine"] == "native":
                rows.append(f"<li>{e(d['title'])} <span class=pill>{e(d.get('spec'))}</span> "
                            "<span class=note>cannot be run yet (RBT-WP-0007)</span></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> "
                            f"<span class=note>{e(score_line(R.scorecard(p)))}</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><p class=note>{e(score_line(R.scorecard(p)))}</p>"
                    f"{self.runs_list(i, 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 runs_list(self, i: int, p) -> str:
        rows = []
        for rid in reversed(R.list_runs(p)):
            r = R.load(p, rid)
            rows.append(f"<li><a href='/run/{i}/{e(rid)}'>{e(rid)}</a> · {e(r.data['operator'])} · "
                        f"<span class=pill>{e(R.effective_outcome(r))}</span></li>")
        return f"<h2>Runs</h2><ul>{''.join(rows)}</ul>" if rows else ""

    def capture_html(self, url: str, s: dict, st: dict, state: str, finished: bool) -> str:
        saved = ""
        for c in st.get("captures", []):
            head = f"$ {c['command']}\n" if c.get("command") else ""
            saved += f"<pre>{e(head + c['output'][:4000])}</pre><p class=note>captured {e(c['at'][:19])} ({e(c['source'])})</p>"
        if finished or state == "todo":
            return f"<details><summary class=note>Captured output</summary>{saved}</details>" if saved else ""
        paste = ""
        if state == "running" and s["verify"].get("expect") == "output-contains":
            paste = f"<button formaction='{url}/verify-pasted' formnovalidate>Check against this output</button>"
        return (f"<details><summary class=note>Capture command and output (optional)"
                f"{' - ' + str(len(st.get('captures', []))) + ' saved' if st.get('captures') else ''}</summary>{saved}"
                f"<form method=post action='{url}/capture'>{self.hidden()}"
                "<p><input name=command class=grow style='width:100%' placeholder='command you actually ran (optional)'></p>"
                "<p><textarea name=output rows=5 style='width:100%' placeholder='paste the output'></textarea></p>"
                f"<p class=row><button>Save capture</button>{paste}</p>"
                "<p class=note>Kept in the run record as documentation. Never paste secrets: obvious markers are "
                "refused, but not everything can be detected. Use the check button when the command ran on a system "
                "this runbook cannot reach.</p></form></details>")

    def close_run_form(self, i: int, rid: str, r, p) -> str:
        outcome = r.data.get("outcome")
        if outcome:
            note = f" - {e(r.data['outcome_note'])}" if r.data.get("outcome_note") else ""
            box = f"<p><b>Run finished: {e(outcome)}</b>{note}</p>"
            if outcome == "succeeded":
                if r.data.get("exercise_recorded"):
                    box += f"<p>Pack recorded as <b>{e(r.data['exercise_recorded'])}</b>. Commit the change in the owning repository.</p>"
                else:
                    old, new = R.exercise_update(r)
                    line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip()
                    box += (f"<p>Current pack status: <span class=pill>{e(p.data['exercise_status'])}</span>. "
                            f"Recording this run would set it to <b>{e(line)}</b> in <code>{e(p.path)}</code> "
                            "(a reviewable edit; commit it in the owning repository).</p>"
                            f"<form method=post action='/run/{i}/{e(rid)}/exercise'>{self.hidden()}"
                            "<button>Record pack as exercised</button></form>")
            return f"<div id=close class=step><h2>Close this run</h2>{box}</div>"
        states = [v["state"] for v in r.data["steps"].values()]
        btns = []
        if r.complete and all(x == "done" for x in states):
            btns.append("<button name=outcome value=succeeded>Finish: succeeded</button>")
        if "failed" in states:
            btns.append("<button name=outcome value=failed onclick=\"return confirm('Finish this run as failed?')\">Finish: failed</button>")
        btns.append("<button name=outcome value=abandoned onclick=\"return confirm('Abandon this run?')\">Abandon run</button>")
        return (f"<div id=close class=step><h2>Close this run</h2>"
                "<p class=note>Finishing records the outcome in the pack's score. A finished run is read-only. "
                "Unfinished runs count as stalled, and as abandoned after a week.</p>"
                f"<form method=post action='/run/{i}/{e(rid)}/finalize'>{self.hidden()}"
                "<p><input name=note class=grow style='width:100%' placeholder='comment (optional)'></p>"
                f"<p class=row>{' '.join(btns)}</p></form></div>")

    def feedback_form(self, i: int, rid: str, r) -> str:
        fb = R.latest_feedback(r)
        btns = " ".join(f"<button name=rating value={k} title='{e(t)}' aria-label='{e(t)}'"
                        f" style='font-size:1.6rem;padding:.1rem .6rem'>{face}</button>"
                        for k, face, t in SMILEYS)
        form = (f"<form method=post action='/run/{i}/{e(rid)}/feedback'>{self.hidden()}"
                "<p class=note>Practicable? Anything that felt strange or unclear, or an idea to make it better. "
                "Never include secrets.</p>"
                f"<p><textarea name=comment rows=3 style='width:100%' placeholder='optional comment'>"
                f"{e(fb['comment']) if fb else ''}</textarea></p><p class=row>{btns}</p></form>")
        if fb:                                              # collected: close the section, suggest the next step
            face = next(f for k, f, _ in SMILEYS if k == fb["rating"])
            return ("<div id=feedback class=step><p><b>Thank you.</b> Recorded: "
                    f"{face} {e(fb['comment'] or '(no comment)')}</p>"
                    "<p><a href='/'>← Back to the runbook list</a></p>"
                    f"<details><summary class=note>Change my feedback</summary>{form}</details></div>")
        return f"<div id=feedback class=step><h2>How was this runbook?</h2>{form}</div>"

    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
        waiting_for = ""
        finished = bool(r.data.get("outcome"))
        total = len(d["steps"])
        done_n = sum(1 for x in d["steps"] if r.state(x["id"])["state"] in ("done", "skipped"))
        out.append(f"<p class=eyebrow>{done_n} of {total} steps done</p>"
                   f"<progress max={total} value={done_n} style='width:100%'></progress>")
        current = next((x["id"] for x in d["steps"] if r.state(x["id"])["state"] == "running"), None)
        if current is None:
            current = next((x["id"] for x in d["steps"] if r.state(x["id"])["state"] in ("todo", "failed")), None)
        for n, s in enumerate(d["steps"], 1):
            st = r.state(s["id"]); state = st["state"]
            act = ""
            cmd = R.command_for(r, s["id"])
            block = ""
            if cmd and state == "running":       # the command and Copy appear only once the step is started
                block = f"<pre>{e(cmd)}</pre><button type=button data-copy>Copy</button>"
            elif cmd and state == "done":
                block = f"<pre>{e(cmd)}</pre>"                 # kept as a record; no Copy once done
            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 this step</button></form>"
            elif state == "todo":
                act = f"<p class=note>Waiting for: {e(waiting_for)}</p>"
            elif state == "running":
                ver = s["verify"]
                if ver.get("expect", "manual") == "manual":
                    act = (f"<form method=post action='{url}/verify'>{self.hidden()}"
                           f"<p>Is this true? <b>{e(ver['done_when'])}</b></p>"
                           "<button name=ok value=1>Yes, it is</button> <button name=ok value=0>Not yet</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 not v:
                    act = "<p class=note>1. Run the command, then check &nbsp;→&nbsp; 2. Mark done to lock it in</p>" + act
                elif v["passed"]:
                    act += "<p><b>✔ Check passed.</b> Press <b>Mark done</b> to lock it in; the next step opens after that.</p>"
                else:
                    act += "<p class=bad>✘ Check not passed. Fix the cause, then check again, or mark the step failed.</p>"
                ev = ""
                if v and v["passed"]:
                    ev = "".join(f"<label>{e(x)} <input name='e_{e(x)}' required></label> " for x in s.get("evidence", []))
                    ev = f"<p>{ev}</p>" if ev else ""
                done_btn = f"<button formaction='{url}/done'>Mark done</button> " if v and v["passed"] else ""
                act += (f"<form method=post action='{url}/fail'>{self.hidden()}{ev}<p class=row>{done_btn}"
                        f"<button formaction='{url}/fail' formnovalidate data-fail>Mark failed</button> "
                        "<input name=note class=grow placeholder='comment (what happened, what went wrong; no secrets)'>"
                        "</p><p class='note bad' data-warn hidden>Add a comment about what went wrong, or click "
                        "<b>Mark failed</b> again to fail without one.</p></form>")
            elif state == "failed":
                act = f"<p class=bad>Failed: {e(st.get('note') or '(no comment)')}</p><form method=post action='{url}/retry'>{self.hidden()}<button>Retry</button></form>"
            elif state == "done" and st.get("note"):
                act = f"<p class=note>Comment: {e(st['note'])}</p>"
            if finished and state != "done":
                act = ""                                  # a finished run is read-only
            if state not in ("done", "skipped") and not blocked:
                blocked = True
                waiting_for = f"step {n}, {s['title']}"
            rb = f"<p class=note>Rollback: {e(s['rollback'])}</p>" if s.get("rollback") and state in ("running", "failed") else ""
            anchor = " id=current" if s["id"] == current else ""
            fold = "" if state in ("done", "skipped") else " open"     # finished steps start folded; the triangle toggles
            out.append(f"<details{anchor} class='step {state}'{fold}><summary><span class=head>"
                       f"<span><b>{n}. {e(s['title'])}</b> "
                       f"<span class=note>[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}</span></span>"
                       f"<span class=pill>{state}</span></span></summary>{block}"
                       f"{self.capture_html(url, s, st, state, finished)}"
                       f"<p class=note>Done when: {e(s['verify']['done_when'])}</p>{rb}{act}</details>")
        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>")
        if r.complete:
            out.append(self.feedback_form(i, rid, r))
        out.append(self.close_run_form(i, rid, r, 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']}#current")
                if len(parts) == 4 and parts[0] == "run" and parts[3] == "feedback":
                    i, rid = int(parts[1]), parts[2]
                    R.add_feedback(R.load(app.packs[i], rid), form.get("rating", ""), form.get("comment", ""))
                    return self._redirect(f"/run/{i}/{rid}#feedback")
                if len(parts) == 4 and parts[0] == "run" and parts[3] in ("finalize", "exercise"):
                    i, rid = int(parts[1]), parts[2]
                    r = R.load(app.packs[i], rid)
                    if parts[3] == "finalize":
                        R.finalize(r, form.get("outcome", ""), form.get("note", ""))
                    else:
                        R.apply_exercise(r)
                        app.packs[i] = P.load(app.packs[i].path)      # pick up the new exercise status
                    return self._redirect(f"/run/{i}/{rid}#close")
                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=(form.get("ok") == "1") 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_")}, form.get("note", ""))
                elif action == "fail": R.fail(r, sid, form.get("note", ""))
                elif action == "capture": R.add_capture(r, sid, form.get("output", ""), form.get("command", ""))
                elif action == "verify-pasted": R.verify_pasted(r, sid, form.get("output", ""), form.get("command", ""))
                elif action == "retry": R.retry(r, sid)
                return self._redirect(f"/run/{i}/{rid}#current")
            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