tegwick
6d70f4d97d
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
664 lines
46 KiB
Python
664 lines
46 KiB
Python
"""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 . import ui_projectrun as UPR
|
|
from . import projectrun as PRUN
|
|
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}
|
|
.hl a{text-decoration:none}.hl a::after{content:" →";opacity:.35;font-weight:400}.hl a:hover::after{opacity:1}
|
|
.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"<a href='{h}' data-nav>{html.escape(t)}</a>" for t, h in NAV)
|
|
side = (f"<aside class=side><a class=brand href='/'>runbook-tutorials</a><nav>{nav}</nav>"
|
|
f"<div class=side-foot><a class=me href='/user' data-nav title='Your settings'>"
|
|
f"<span class=avatar>{html.escape(label[:1].upper())}</span><span>{html.escape(label)}</span></a>"
|
|
"<button type=button data-theme-toggle title='Switch between system, light and dark'>Theme: system</button></div></aside>")
|
|
return (f"<!doctype html><html lang=en><head><meta charset=utf-8><meta name=viewport content='width=device-width,initial-scale=1'>"
|
|
f"<title>{html.escape(title)}</title><style>{CSS}</style><script>{THEME_INIT}</script></head>"
|
|
f"<body><div class=app>{side}<main>{body}</main></div>"
|
|
"<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();}}));"
|
|
# sidebar: mark where we are; theme switch cycles system -> light -> dark
|
|
"const p=location.pathname;const alias={'/projects':['/project/','/template/','/project-run/'],'/runbooks':['/pack/','/run/'],'/attribute-sets':['/attribute-set/']};"
|
|
"document.querySelectorAll('a[data-nav]').forEach(a=>{const h=a.getAttribute('href');"
|
|
"if(h==='/'?p==='/':p===h||p.startsWith(h+'/')||(alias[h]||[]).some(x=>p.startsWith(x)))a.setAttribute('aria-current','page')});"
|
|
"const tb=document.querySelector('[data-theme-toggle]');const cur=()=>{try{return localStorage.getItem('rtut-theme')||'system'}catch(x){return 'system'}};"
|
|
"const show=()=>{if(tb)tb.textContent='Theme: '+cur()};show();"
|
|
"if(tb)tb.onclick=()=>{const n={system:'light',light:'dark',dark:'system'}[cur()];"
|
|
"try{n==='system'?localStorage.removeItem('rtut-theme'):localStorage.setItem('rtut-theme',n)}catch(x){};"
|
|
"if(n==='system')document.documentElement.removeAttribute('data-theme');else document.documentElement.setAttribute('data-theme',n);show()};"
|
|
# 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></body></html>").encode()
|
|
|
|
|
|
def e(s) -> str:
|
|
return html.escape(str(s))
|
|
|
|
|
|
def heading(level: str, text: str, href: str | None = None, extra: str = "") -> str:
|
|
"""A section headline. On the Home page it links to the matching sidebar section, as a shortcut."""
|
|
if href:
|
|
return f"<{level} class=hl><a href='{href}'>{html.escape(text)}</a>{extra}</{level}>"
|
|
return f"<{level}>{html.escape(text)}{extra}</{level}>"
|
|
|
|
|
|
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"<input type=hidden name={name} value='{e(u['operator'])}'>"
|
|
f"<p class=note>Running as <b>{label}</b> · <a href='/user'>change</a></p>")
|
|
|
|
def activities_html(self, limit: int = 12, level: str = "h2", link: bool = False) -> str:
|
|
events = ACT.latest(self.packs, limit, PRUN.list_runs(self.registry()))
|
|
if not events:
|
|
return f"{heading(level, 'Latest activities', '/activity' if link else None)}<p class=note>Nothing has happened yet. Start a run or a project.</p>"
|
|
rows = "".join(
|
|
f"<li><span class=note>{e(x['at'][:16].replace('T', ' '))}</span> · <b>{e(x['who'])}</b> {e(x['text'])} "
|
|
f"<span class=note>· {e(x['pack'])}{' · project ' + e(x['project']) if x.get('project') else ''}</span>"
|
|
f"{' · <a href=' + chr(39) + x['link'] + chr(39) + '>open</a>' if x['link'] else ''}</li>" for x in events)
|
|
return (f"{heading(level, 'Latest activities', '/activity' if link else None, ' <span class=note style=' + chr(39) + 'font-weight:400;font-size:.85rem' + chr(39) + '>(times in UTC)</span>')}"
|
|
f"<ul class=feed>{rows}</ul>")
|
|
|
|
def user_page(self, msg: str = "") -> bytes:
|
|
u = USER.current()
|
|
env = ("<p class=note>The operator is set by the <code>RTUT_USER</code> environment variable, which takes precedence over "
|
|
"anything saved here.</p>" if u["source"] == "environment" else "")
|
|
src = {"environment": "the environment", "saved": "your saved settings", "login": "your operating-system login"}[u["source"]]
|
|
return page("You", f"<h1>You</h1>{'<p class=bad>' + e(msg) + '</p>' if msg else ''}"
|
|
f"<p>Runs, reviews and activity are recorded under <b>{e(u['operator'])}</b>, taken from {src}.</p>{env}"
|
|
"<p class=note>These are the attributes of the app-level set <code>user-attributes</code>. Every project can reuse it "
|
|
"(<code>attribute_sets: [user-attributes]</code>). If the UI later sits behind a login, the login fills the same attribute.</p>"
|
|
f"<form method=post action='/user'>{self.hidden()}"
|
|
f"<p><label>Operator (one word)<br><input name=operator value='{e(u['operator'])}' required pattern='[A-Za-z0-9][A-Za-z0-9._-]*' maxlength=40></label></p>"
|
|
f"<p><label>Display name (optional)<br><input name=display_name value='{e(u['display_name'])}' maxlength=80></label></p>"
|
|
"<button>Save</button></form>")
|
|
|
|
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"<input type=hidden name=_t value='{self.token}'>"
|
|
|
|
def index(self) -> bytes:
|
|
return page("Home", UP.index_sections(self, home=True) + self.activities_html(link=True) + self.runbooks_html(link=True))
|
|
|
|
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", link: bool = False) -> str:
|
|
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> {UP.maturity_pill(p)} "
|
|
f"<span class=note>{e(score_line(R.scorecard(p)))}</span></li>")
|
|
return f"{heading(level, 'Runbooks', '/runbooks' if link else None)}<ul>{''.join(rows) or '<li>No packs found.</li>'}</ul>"
|
|
|
|
def pack_page(self, i: int) -> bytes:
|
|
p = self.packs[i]; d = p.data
|
|
if p.spec == "runbook-pack/v0.2":
|
|
framing = "".join(f"<li>{e(x)}</li>" for x in d.get("framing", []))
|
|
return page(d["title"], f"<h1>{e(d['title'])}</h1><p>{UP.pill('v0.2')} {UP.maturity_pill(p)}</p><p>{e(d.get('outcome', ''))}</p>"
|
|
f"<p class=note>{e(score_line(R.scorecard(p)))}</p>"
|
|
f"<details class=framing open><summary><b>Keep in mind</b></summary><ul>{framing}</ul></details>"
|
|
f"{UP.maturity_block(self, i, p)}{self.runs_list(i, p)}<h2>Start a run</h2>{UI2.start_form(self, i, p)}")
|
|
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>{UP.maturity_pill(p)}</p>"
|
|
f"<p>{e(d.get('outcome', ''))}</p><p class=note>{e(score_line(R.scorecard(p)))}</p>"
|
|
f"{UP.maturity_block(self, i, p)}{self.runs_list(i, p)}<h2>Prerequisites</h2><ul>{prereq}</ul>"
|
|
f"<form method=post action='/run/new/{i}'>{self.hidden()}{fields}"
|
|
f"{self.who()}<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)
|
|
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"<li><a href='/run/{i}/{e(rid)}'>{e(rid)}</a> · {e(r.data['operator'])} · "
|
|
f"<span class=pill data-s='{e(R.effective_outcome(r))}'>{e(R.effective_outcome(r))}</span> "
|
|
f"<span class=note>{e(UI2.stamp(t['started']))} · {e(took)}</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>{self.locked_note(r)}"
|
|
f"<p><a href='/run/{i}/{e(rid)}/record.json'>Download the run record (JSON)</a> "
|
|
"<span class=note>Check it anywhere with <code>rtut seal <file></code>.</span></p>")
|
|
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()}"
|
|
f"<p><input name=note class=grow style='width:100%' placeholder='comment (optional)'></p>{self.close_feedback_fields(r)}"
|
|
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 r.data.get("outcome"): # a closed run is locked: show what was said, offer no change
|
|
if not fb:
|
|
return ""
|
|
face = next(f for k, f, _ in SMILEYS if k == fb["rating"])
|
|
return (f"<div id=feedback class=step><h2>Feedback</h2><p>{face} {e(fb['comment'] or '(no comment)')}</p>"
|
|
"<p class=note>The run is closed and locked, so its feedback can no longer be changed.</p></div>")
|
|
if fb: # collected while the run is open: close the section
|
|
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>"
|
|
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 close_feedback_fields(self, r) -> str:
|
|
"""Optional rating and comment inside the closing form: a closed run cannot take feedback afterwards."""
|
|
if R.latest_feedback(r):
|
|
return ""
|
|
faces = " ".join(f"<label title='{e(t)}'><input type=radio name=rating value={k}> <span style='font-size:1.4rem'>{face}</span></label>"
|
|
for k, face, t in SMILEYS)
|
|
return ("<fieldset><legend>How was this runbook? <span class=note>(optional; it cannot be added after closing)</span></legend>"
|
|
f"<p class=row>{faces}</p><textarea name=fb_comment rows=2 style='width:100%' "
|
|
"placeholder='anything strange or unclear, or an idea to make it better (no secrets)'></textarea></fieldset>")
|
|
|
|
def locked_note(self, r) -> str:
|
|
"""What 'closed' means: the record is sealed and locked. A record changed outside the app is called out."""
|
|
state = R.seal_state(r)
|
|
if state == "broken":
|
|
return ("<p class=bad>⚠ This closed run's record was changed outside the app after it was sealed. "
|
|
"It no longer counts towards maturity.</p>")
|
|
base = ("This run is <b>closed and locked</b>: its record is sealed and cannot be changed. "
|
|
"Start a new run to try again.")
|
|
if state == "unsealed":
|
|
base += " (It was closed before sealing existed, so its record carries no seal.)"
|
|
return f"<p class=note>{base}</p>"
|
|
|
|
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 <a href='/project/{e(ctx['project'])}'>{e(ctx['project'])}</a>"
|
|
f"{' @' + e(ctx['target']) if ctx.get('target') else ''} · {e(ctx.get('environment', ''))}"
|
|
f"{' · part of <a href=' + chr(39) + '/project-run/' + e(ctx['project_run']) + chr(39) + '>project run ' + e(ctx['project_run']) + '</a>' if ctx.get('project_run') else ''}")
|
|
out = [f"<h1>{e(d['title'])}</h1><p>{UP.maturity_pill(p)}</p>"
|
|
f"<p class=note>run {e(rid)} · operator {e(r.data['operator'])}{where}</p>"
|
|
f"<p class=note>{UI2.times_line(r)}</p>"]
|
|
step_secs = R.run_times(r)["step_seconds"]
|
|
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 → 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'))}"
|
|
f"{' · took ' + e(R.fmt_duration(step_secs[s['id']])) if s['id'] in step_secs and state == 'done' else ''}</span></span>"
|
|
f"<span class=pill data-s='{state}'>{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 or r.data.get("outcome") or R.latest_feedback(r):
|
|
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 _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 _send_record(self, path, filename: str):
|
|
"""A closed record, exactly as stored, as a download. Its seal can be checked with `rtut seal <file>`."""
|
|
import json
|
|
data = json.loads(path.read_text())
|
|
if not data.get("outcome"):
|
|
return self._send_file(409, b"the record can be downloaded once the run is closed", "text/plain; charset=utf-8", None)
|
|
state = R.record_seal_state(data)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("X-Rtut-Seal", state)
|
|
self.send_header("Content-Security-Policy", "default-src 'none'; sandbox")
|
|
self.send_header("Content-Disposition", f"attachment; filename=\"{filename}\"")
|
|
self.end_headers(); self.wfile.write(path.read_bytes())
|
|
|
|
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) == 4 and parts[3] == "record.json":
|
|
return self._send_record(R.load(app.packs[int(parts[1])], parts[2]).path, f"run-{parts[2]}.json")
|
|
if parts[0] == "project-run" and len(parts) == 3 and parts[2] == "record.json":
|
|
return self._send_record(PRUN.load(app.registry(), parts[1]).path, f"project-run-{parts[1]}.json")
|
|
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-run" and len(parts) == 2:
|
|
body = UPR.page_for(app, parts[1])
|
|
if body:
|
|
return self._send(body)
|
|
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", "<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]
|
|
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", "<p>Not found.</p>"), 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] == "runall") or (len(parts) == 3 and parts[0] == "project-run"):
|
|
return self._redirect(UPR.handle_post(app, parts, form))
|
|
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] == "finalize":
|
|
i, rid = int(parts[1]), parts[2]
|
|
R.finalize(R.load(app.packs[i], rid), form.get("outcome", ""), form.get("note", ""),
|
|
rating=form.get("rating") or None, comment=form.get("fb_comment", ""))
|
|
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-run":
|
|
return self._send(UPR.page_for(app, parts[1], msg) or page("Error", f"<p class=bad>{e(msg)}</p>"), 400)
|
|
if parts[0] == "project":
|
|
return self._send(UP.project_page(app, parts[1], msg) or page("Error", f"<p class=bad>{e(msg)}</p>"), 400)
|
|
if parts[0] == "template":
|
|
return self._send(UP.template_page(app, parts[1], msg) or page("Error", f"<p class=bad>{e(msg)}</p>"), 400)
|
|
if parts[0] == "pack":
|
|
return self._send(app.pack_page(int(parts[1])).replace(b"<h1>", f"<p class=bad>{e(msg)}</p><h1>".encode(), 1), 400)
|
|
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
|