"""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 from . import ui_projects as UP from . import ui_v2 as UI2 from . import activity as ACT, user as USER from . import ui_attrs as UA from .attrs import Registry LIGHT = "--fg:#141414;--bg:#fbfbf9;--mut:#6b6b66;--line:#d9d9d2;--card:#fff;--hi:#f2e35a;--ok:#1a7f37;--warn:#9a6700;--bad:#c62828;--side:#f3f3ef" DARK = "--fg:#ececea;--bg:#121212;--mut:#9a9a94;--line:#2e2e2b;--card:#191918;--hi:#f2e35a;--ok:#56d364;--warn:#e3b341;--bad:#ff7b72;--side:#0e0e0d" CSS = """ :root{%s;--radius:6px;color-scheme:light} @media(prefers-color-scheme:dark){:root:not([data-theme=light]){%s;color-scheme:dark}} :root[data-theme=dark]{%s;color-scheme:dark} :root[data-theme=light]{color-scheme:light} """ % (LIGHT, DARK, DARK) + """ *{box-sizing:border-box} body{font:15px/1.55 system-ui,-apple-system,"Segoe UI",sans-serif;margin:0;color:var(--fg);background:var(--bg)} h1{font-size:1.6rem;font-weight:650;letter-spacing:-.01em;margin:.4rem 0 .6rem} h2{font-size:1.1rem;font-weight:650;margin:1.8rem 0 .5rem;padding-bottom:.25rem;border-bottom:1px solid var(--line)} h3{font-size:.95rem;margin:1.2rem 0 .3rem} a{color:inherit;text-underline-offset:.15em}a:hover{background:var(--hi);color:#141414} hr{border:0;border-top:1px solid var(--line)} code,pre{font:13px/1.5 ui-monospace,SFMono-Regular,Menlo,monospace} pre{background:var(--card);border:1px solid var(--line);border-radius:var(--radius);padding:.6rem .8rem;overflow:auto;margin:.5rem 0} .eyebrow{font:11px ui-monospace,monospace;text-transform:uppercase;letter-spacing:.08em;color:var(--mut);margin:.2rem 0} .note{color:var(--mut)}.bad{color:var(--bad);font-weight:600} progress{height:.45rem;border:0;border-radius:99px;overflow:hidden;background:var(--line);accent-color:var(--fg)} progress::-webkit-progress-bar{background:var(--line)}progress::-webkit-progress-value{background:var(--fg)} button,input,select,textarea{font:inherit;color:inherit;background:var(--card);border:1px solid var(--line);border-radius:var(--radius);padding:.3rem .6rem} button{cursor:pointer;background:var(--fg);color:var(--bg);border-color:var(--fg)} button:hover{background:var(--hi);color:#141414;border-color:var(--hi)} button[name=ok][value="0"],button[data-fail],button[type=button]{background:var(--card);color:var(--fg);border-color:var(--line)} button:focus-visible,input:focus-visible,select:focus-visible,textarea:focus-visible,summary:focus-visible{outline:2px solid var(--hi);outline-offset:2px} label{display:inline-block}fieldset{border:1px solid var(--line);border-radius:var(--radius);margin:.8rem 0;padding:.4rem .9rem} table{border-collapse:collapse;width:100%;margin:.5rem 0;background:var(--card);border:1px solid var(--line);border-radius:var(--radius)} td,th{border-bottom:1px solid var(--line);padding:.4rem .6rem;text-align:left;vertical-align:top}th{font:11px ui-monospace,monospace;text-transform:uppercase;letter-spacing:.06em;color:var(--mut)} tr:last-child td{border-bottom:0} table.data td:first-child{width:42%} table.edit input,table.edit select{width:100%} td label{display:block} td input:not([type=checkbox]):not([type=hidden]):not([type=radio]){width:100%} .row{display:flex;gap:.5rem;align-items:center;flex-wrap:wrap}.grow{flex:1;min-width:12rem}[hidden]{display:none} summary{cursor:pointer}.head{display:inline-flex;width:calc(100% - 1.6rem);justify-content:space-between;align-items:baseline;gap:1rem;vertical-align:top} .step,.stage,.panel,.banner,.questions,details.framing,details.override{border:1px solid var(--line);border-radius:var(--radius);background:var(--card);padding:.7rem 1rem;margin:.7rem 0} .stage{padding:.8rem 1.1rem;background:transparent}.stage>summary{font-size:1.02rem} .stage .step,.stage .questions,.stage .stage{background:var(--card)} .step.running{border-color:var(--fg);box-shadow:0 0 0 1px var(--fg)}.step.failed{border-color:var(--bad)}.step.done,.step.pushed{opacity:.9} .step.gate{border-style:dashed}.step.gate.passed{border-style:solid} details.framing{background:transparent;border-left:3px solid var(--warn)}details.override{background:transparent;border-style:dashed;margin:.5rem 0} .panel{background:transparent}.panel[data-s=auto]{border-color:var(--fg)} .banner{background:var(--hi);color:#141414;border-color:var(--hi)}.banner a{text-decoration-thickness:2px}.banner .note{color:#3d3a10} .banner .chip{background:rgba(255,255,255,.55);color:#141414;border-color:rgba(0,0,0,.2)} .questions{background:transparent;border-left:3px solid var(--fg)} .tabs{display:flex;gap:.2rem;margin:1rem 0;border-bottom:1px solid var(--line)} .tab{padding:.4rem .9rem;text-decoration:none;border:1px solid transparent;border-bottom:0;border-radius:var(--radius) var(--radius) 0 0;color:var(--mut)} .tab.on{color:var(--fg);border-color:var(--line);background:var(--card);font-weight:650;margin-bottom:-1px} .chip{display:inline-block;font:12px ui-monospace,monospace;border:1px solid var(--line);border-radius:99px;padding:0 .55rem;margin:.1rem .1rem;text-decoration:none;color:var(--mut)} a.chip:hover{background:var(--hi);color:#141414;border-color:var(--hi)}.chip.on{background:var(--fg);color:var(--bg);border-color:var(--fg)} .chip[data-s=irreversible]{color:var(--bad);border-color:var(--bad)} .pill{font:11.5px ui-monospace,monospace;border:1px solid var(--line);border-radius:99px;padding:.05rem .55rem;white-space:nowrap;color:var(--mut);background:var(--card)} .pill[data-s=done],.pill[data-s=passed],.pill[data-s=ok],.pill[data-s=complete],.pill[data-s=succeeded],.pill[data-s=applied]{color:var(--ok);border-color:var(--ok)} .pill[data-s=running],.pill[data-s="in progress"],.pill[data-s=applying],.pill[data-s=auto],.pill[data-s=auto-confirm]{background:var(--hi);color:#141414;border-color:var(--hi)} .pill[data-s=failed],.pill[data-s=err],.pill[data-s=invalid]{color:var(--bad);border-color:var(--bad)} .pill[data-s=changed-after-use],.pill[data-s=pushed],.pill[data-s=waived],.pill[data-s=stale],.pill[data-s=override],.pill[data-s=push-through],.pill[data-s=waiver],.pill[data-s=modified],.pill[data-s=abandoned]{color:var(--warn);border-color:var(--warn)} .pill[data-s=nil],.pill[data-s=todo],.pill[data-s=open],.pill[data-s=planned],.pill[data-s="not checked"],.pill[data-s=not-applied]{color:var(--mut);border-style:dashed} .feed{list-style:none;padding:0}.feed li{padding:.3rem 0;border-bottom:1px dashed var(--line)} .checks{list-style:none;padding:0;margin:.4rem 0}.checks li{padding:.35rem 0;border-bottom:1px dashed var(--line)}.checks li:last-child{border:0} #current{scroll-margin-top:1rem} .app{display:grid;grid-template-columns:15.5rem minmax(0,1fr);min-height:100vh} main{max-width:58rem;width:100%;padding:1.8rem 1.6rem 4rem} .side{position:sticky;top:0;align-self:start;height:100vh;display:flex;flex-direction:column;gap:.2rem;padding:1.1rem .8rem;background:var(--side);border-right:1px solid var(--line)} .brand{font:12px ui-monospace,monospace;text-transform:uppercase;letter-spacing:.08em;text-decoration:none;padding:.3rem .6rem 1rem;color:var(--fg)} .side nav{display:flex;flex-direction:column;gap:.1rem} .side a[data-nav]{display:block;padding:.4rem .7rem;border-radius:var(--radius);text-decoration:none;color:var(--mut)} .side a[data-nav]:hover{background:var(--hi);color:#141414} .side a[data-nav][aria-current=page]{background:var(--card);color:var(--fg);font-weight:650;box-shadow:inset 3px 0 0 var(--fg)} .side-foot{margin-top:auto;display:flex;flex-direction:column;gap:.5rem;border-top:1px solid var(--line);padding-top:.8rem} .side-foot a.me{display:flex;gap:.5rem;align-items:center} .avatar{display:inline-grid;place-items:center;width:1.7rem;height:1.7rem;border-radius:99px;background:var(--fg);color:var(--bg);font:12px ui-monospace,monospace} .side-foot button{background:var(--card);color:var(--fg);border-color:var(--line);font-size:.85rem} @media(max-width:52rem){.app{grid-template-columns:1fr}.side{position:static;height:auto;flex-direction:row;flex-wrap:wrap;align-items:center;border-right:0;border-bottom:1px solid var(--line);padding:.6rem .8rem} .brand{padding:.3rem .6rem}.side nav{flex-direction:row;flex-wrap:wrap}.side-foot{margin:0 0 0 auto;border:0;padding:0;flex-direction:row;align-items:center}main{padding:1.2rem .9rem 3rem}}details>summary{list-style-position:outside} @media(max-width:40rem){body{margin:1rem auto}.head{display:block;width:auto}.tabs{overflow:auto}} @media print{.side,button,.tabs,form{display:none}.app{display:block}.step,.stage{break-inside:avoid}} """ NAV = (("Home", "/"), ("Projects", "/projects"), ("Runbooks", "/runbooks"), ("Attribute sets", "/attribute-sets"), ("Activity", "/activity")) # applied before the page paints, so a saved theme never flashes; ?theme=dark|light|system also sets it THEME_INIT = ("(function(){try{var q=new URLSearchParams(location.search).get('theme');" "if(q==='dark'||q==='light')localStorage.setItem('rtut-theme',q);else if(q==='system')localStorage.removeItem('rtut-theme');" "var t=localStorage.getItem('rtut-theme');" "if(t==='dark'||t==='light')document.documentElement.setAttribute('data-theme',t);" "else document.documentElement.removeAttribute('data-theme')}catch(x){}})()") def page(title: str, body: str) -> bytes: u = USER.current() label = u["display_name"] or u["operator"] nav = "".join(f"{html.escape(t)}" for t, h in NAV) side = (f"") return (f"" f"{html.escape(title)}" f"
{side}
{body}
" "").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: line = (f"tried {c['tried']} · succeeded {c['succeeded']} · failed {c['failed']} · " f"abandoned {c['abandoned']} · stalled {c['stalled']}") if c.get("avg_seconds") is not None: line += (f" · average {R.fmt_duration(c['avg_seconds'])} per successful run" f" (fastest {R.fmt_duration(c['fastest_seconds'])}, {c['timed_runs']} timed)") return line class App: def __init__(self, roots: list[Path]): self.roots = [Path(r) for r in roots] self.token = secrets.token_urlsafe(16) self.packs = [P.load(p) for r in roots for p in P.discover(r)] def registry(self) -> Registry: """Fresh view of projects, attribute sets, templates and packs (files may have changed). The active user is available to every project as the app-level attribute set `user-attributes`.""" return Registry(self.roots, extra_sets={USER.USER_SET_ID: USER.attribute_set()}) def operator(self) -> str: return USER.current()["operator"] def who(self, name: str = "operator") -> str: """The active user, as a hidden field and a line, instead of asking for a name again.""" u = USER.current() label = f"{e(u['display_name'])} ({e(u['operator'])})" if u["display_name"] else e(u["operator"]) return (f"" f"

Running as {label} · change

") def activities_html(self, limit: int = 12, level: str = "h2") -> str: events = ACT.latest(self.packs, limit) if not events: return f"<{level}>Latest activities

Nothing has happened yet. Start a run or a project.

" rows = "".join( f"
  • {e(x['at'][:16].replace('T', ' '))} · {e(x['who'])} {e(x['text'])} " f"· {e(x['pack'])}{' · project ' + e(x['project']) if x.get('project') else ''}" f"{' · open' if x['link'] else ''}
  • " for x in events) return (f"<{level}>Latest activities (times in UTC)" f"") def user_page(self, msg: str = "") -> bytes: u = USER.current() env = ("

    The operator is set by the RTUT_USER environment variable, which takes precedence over " "anything saved here.

    " if u["source"] == "environment" else "") src = {"environment": "the environment", "saved": "your saved settings", "login": "your operating-system login"}[u["source"]] return page("You", f"

    You

    {'

    ' + e(msg) + '

    ' if msg else ''}" f"

    Runs, reviews and activity are recorded under {e(u['operator'])}, taken from {src}.

    {env}" "

    These are the attributes of the app-level set user-attributes. Every project can reuse it " "(attribute_sets: [user-attributes]). If the UI later sits behind a login, the login fills the same attribute.

    " f"
    {self.hidden()}" f"

    " f"

    " "
    ") def pack_index(self, pack_id: str) -> int | None: return next((i for i, p in enumerate(self.packs) if p.id == pack_id), None) def hidden(self) -> str: return f"" def index(self) -> bytes: return page("Home", UP.index_sections(self) + self.activities_html() + self.runbooks_html()) def runbooks_page(self) -> bytes: return page("Runbooks", self.runbooks_html("h1")) def projects_page(self) -> bytes: return page("Projects", UP.index_sections(self)) def activity_page(self) -> bytes: return page("Activity", self.activities_html(100, "h1")) def runbooks_html(self, level: str = "h2") -> str: 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'])} {UP.maturity_pill(p)} " f"{e(score_line(R.scorecard(p)))}
  • ") return f"<{level}>Runbooks" def pack_page(self, i: int) -> bytes: p = self.packs[i]; d = p.data if p.spec == "runbook-pack/v0.2": framing = "".join(f"
  • {e(x)}
  • " for x in d.get("framing", [])) return page(d["title"], f"

    {e(d['title'])}

    {UP.pill('v0.2')} {UP.maturity_pill(p)}

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

    " f"

    {e(score_line(R.scorecard(p)))}

    " f"
    Keep in mind
    " f"{UP.maturity_block(self, i, p)}{self.runs_list(i, p)}

    Start a run

    {UI2.start_form(self, i, p)}") 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', ''))}

    {e(score_line(R.scorecard(p)))}

    " f"{UP.maturity_block(self, i, p)}{self.runs_list(i, p)}

    Prerequisites

    " f"
    {self.hidden()}{fields}" f"{self.who()}
    ") def runs_list(self, i: int, p) -> str: rows = [] for rid in reversed(R.list_runs(p)): r = R.load(p, rid) t = R.run_times(r) took = (f"took {R.fmt_duration(t['elapsed'])}" if t["finished_run"] else f"running {R.fmt_duration(t['elapsed'])}") rows.append(f"
  • {e(rid)} · {e(r.data['operator'])} · " f"{e(R.effective_outcome(r))} " f"{e(UI2.stamp(t['started']))} · {e(took)}
  • ") return f"

    Runs

    " 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"
    {e(head + c['output'][:4000])}

    captured {e(c['at'][:19])} ({e(c['source'])})

    " if finished or state == "todo": return f"
    Captured output{saved}
    " if saved else "" paste = "" if state == "running" and s["verify"].get("expect") == "output-contains": paste = f"" return (f"
    Capture command and output (optional)" f"{' - ' + str(len(st.get('captures', []))) + ' saved' if st.get('captures') else ''}{saved}" f"
    {self.hidden()}" "

    " "

    " f"

    {paste}

    " "

    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.

    ") 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"

    Run finished: {e(outcome)}{note}

    " if outcome == "succeeded": if r.data.get("exercise_recorded"): box += f"

    Pack recorded as {e(r.data['exercise_recorded'])}. Commit the change in the owning repository.

    " else: old, new = R.exercise_update(r) line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip() box += (f"

    Current pack status: {e(p.data['exercise_status'])}. " f"Recording this run would set it to {e(line)} in {e(p.path)} " "(a reviewable edit; commit it in the owning repository).

    " f"
    {self.hidden()}" "
    ") return f"

    Close this run

    {box}
    " states = [v["state"] for v in r.data["steps"].values()] btns = [] if r.complete and all(x == "done" for x in states): btns.append("") if "failed" in states: btns.append("") btns.append("") return (f"

    Close this run

    " "

    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.

    " f"
    {self.hidden()}" "

    " f"

    {' '.join(btns)}

    ") def feedback_form(self, i: int, rid: str, r) -> str: fb = R.latest_feedback(r) btns = " ".join(f"" for k, face, t in SMILEYS) form = (f"
    {self.hidden()}" "

    Practicable? Anything that felt strange or unclear, or an idea to make it better. " "Never include secrets.

    " f"

    {btns}

    ") if fb: # collected: close the section, suggest the next step face = next(f for k, f, _ in SMILEYS if k == fb["rating"]) return ("

    Thank you. Recorded: " f"{face} {e(fb['comment'] or '(no comment)')}

    " "

    ← Back to the runbook list

    " f"
    Change my feedback{form}
    ") return f"

    How was this runbook?

    {form}
    " def run_page(self, i: int, rid: str, msg: str = "", query: dict | None = None) -> bytes: p = self.packs[i]; r = R.load(p, rid); d = p.data if p.spec == "runbook-pack/v0.2": return UI2.run_page(self, i, rid, msg, query) ctx = r.data.get("context") or {} where = "" if ctx.get("project"): where = (f" · project {e(ctx['project'])}" f"{' @' + e(ctx['target']) if ctx.get('target') else ''} · {e(ctx.get('environment', ''))}") out = [f"

    {e(d['title'])}

    {e(d['exercise_status'])}

    " f"

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

    " f"

    {UI2.times_line(r)}

    "] step_secs = R.run_times(r)["step_seconds"] if msg: out.append(f"

    {e(msg)}

    ") 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"

    {done_n} of {total} steps done

    " f"") 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"
    {e(cmd)}
    " elif cmd and state == "done": block = f"
    {e(cmd)}
    " # 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"
    {self.hidden()}
    " elif state == "todo": act = f"

    Waiting for: {e(waiting_for)}

    " elif state == "running": ver = s["verify"] if ver.get("expect", "manual") == "manual": act = (f"
    {self.hidden()}" f"

    Is this true? {e(ver['done_when'])}

    " "
    ") else: act = (f"
    {self.hidden()}" f"

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

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

    1. Run the command, then check  →  2. Mark done to lock it in

    " + act elif v["passed"]: act += "

    ✔ Check passed. Press Mark done to lock it in; the next step opens after that.

    " else: act += "

    ✘ Check not passed. Fix the cause, then check again, or mark the step failed.

    " ev = "" if v and v["passed"]: ev = "".join(f" " for x in s.get("evidence", [])) ev = f"

    {ev}

    " if ev else "" done_btn = f" " if v and v["passed"] else "" act += (f"
    {self.hidden()}{ev}

    {done_btn}" f" " "" "

    ") elif state == "failed": act = f"

    Failed: {e(st.get('note') or '(no comment)')}

    {self.hidden()}
    " elif state == "done" and st.get("note"): act = f"

    Comment: {e(st['note'])}

    " 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"

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

    " 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"" f"{n}. {e(s['title'])} " f"[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}" f"{' · took ' + e(R.fmt_duration(step_secs[s['id']])) if s['id'] in step_secs and state == 'done' else ''}" f"{state}{block}" f"{self.capture_html(url, s, st, state, finished)}" 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

    ") if r.complete: out.append(self.feedback_form(i, rid, r)) out.append(self.close_run_form(i, rid, r, p)) out.append("

    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 _send_file(self, code: int, body: bytes, ctype: str, download_name: str | None): """Artefacts: never rendered as a page (plain text or download), never sniffed, sandboxed.""" self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Content-Security-Policy", "default-src 'none'; sandbox") if download_name: safe = "".join(ch for ch in download_name if ch.isalnum() or ch in "._-") or "artefact" self.send_header("Content-Disposition", f"attachment; filename=\"{safe}\"") 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 == ["user"]: return self._send(app.user_page()) if parts == ["projects"]: return self._send(app.projects_page()) if parts == ["runbooks"]: return self._send(app.runbooks_page()) if parts == ["activity"]: return self._send(app.activity_page()) if parts == ["attribute-sets"]: return self._send(UA.sets_page(app)) if len(parts) == 2 and parts[0] == "attribute-set": body = UA.set_page(app, parts[1]) if body: return self._send(body) if parts[0] == "run" and len(parts) == 5 and parts[3] == "artefact": status, body, ctype, dl = UI2.serve_artefact(app, int(parts[1]), parts[2], parts[4]) want_dl = bool(parse_qs(urlparse(self.path).query).get("download")) return self._send_file(status, body, ctype, dl or ("artefact-" + parts[4] if want_dl and status == 200 else None)) if parts[0] == "run": return self._send(app.run_page(int(parts[1]), parts[2], query=parse_qs(urlparse(self.path).query))) if parts[0] == "project" and len(parts) == 2: body = UP.project_page(app, parts[1]) if body: return self._send(body) if parts[0] == "template" and len(parts) == 2: body = UP.template_page(app, parts[1]) if body: return self._send(body) 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] if not form.get("operator", "").strip(): form["operator"] = app.operator() # the active user, unless a form names someone if not form.get("by", "").strip(): form["by"] = app.operator() if parts == ["attribute-sets", "new"] or (len(parts) == 2 and parts[0] == "attribute-set"): try: return self._redirect(UA.handle_post(app, parts, form)) except (ValueError, KeyError) as ex: if parts[0] == "attribute-set" and len(parts) == 2: return self._send(UA.set_page(app, parts[1], str(ex)) or page("Not found", "

    Not found.

    "), 400) return self._send(UA.sets_page(app, str(ex)), 400) if parts == ["user"]: try: USER.save(form.get("operator", ""), form.get("display_name", "")) except ValueError as ex: return self._send(app.user_page(str(ex)), 400) return self._redirect("/user") msg = "" try: if parts[0] == "run" and len(parts) >= 4 and parts[1].isdigit() and app.packs[int(parts[1])].spec == "runbook-pack/v0.2": return self._redirect(UI2.handle_post(app, int(parts[1]), parts[2], parts[3:], form)) if len(parts) == 3 and parts[0] == "pack" and parts[2] == "review": from . import maturity as M i = int(parts[1]) try: M.add_review(app.packs[i], form.get("by", ""), form.get("note", "")) except M.ReviewError as ex: raise ValueError(str(ex)) return self._redirect(f"/pack/{i}#maturity") if len(parts) == 3 and parts[0] == "project" and parts[2] == "edit": from . import project as J doc = app.registry().projects[parts[1]] J.update_project(doc, form.get("title", ""), form.get("description", ""), form.get("goal", "")) return self._redirect(f"/project/{parts[1]}") if len(parts) == 3 and parts[0] == "project" and parts[2] in ("attributes", "run"): from . import project as J reg = app.registry(); doc = reg.projects[parts[1]] if parts[2] == "attributes": J.set_attributes(reg, doc, {k[2:]: v for k, v in form.items() if k.startswith("a_")}) return self._redirect(f"/project/{parts[1]}") r = J.start_run(reg, doc, form.get("pack", ""), form.get("operator", ""), form.get("target") or None, form.get("mode", "manual")) i = app.pack_index(form.get("pack", "")) return self._redirect(f"/run/{i}/{r.data['run_id']}#current") if len(parts) == 3 and parts[0] == "template" and parts[2] == "new": pid = UP.create_from_template(app, parts[1], form) return self._redirect(f"/project/{pid}") if parts[:2] == ["run", "new"]: i = int(parts[2]); p = app.packs[i] if p.spec == "runbook-pack/v0.2": r = UI2.create_run(app, i, form) return self._redirect(f"/run/{i}/{r.data['run_id']}#current") 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[0] == "project": return self._send(UP.project_page(app, parts[1], msg) or page("Error", f"

    {e(msg)}

    "), 400) if parts[0] == "template": return self._send(UP.template_page(app, parts[1], msg) or page("Error", f"

    {e(msg)}

    "), 400) if parts[0] == "pack": return self._send(app.pack_page(int(parts[1])).replace(b"

    ", f"

    {e(msg)}

    ".encode(), 1), 400) 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