vh
751eecb771
Three operator reports, one of them a real defect I had claimed was working. THE REVEAL BUTTON DID NOTHING, for a day. Its handler sat after the content block's closing tag, and a child template's out-of-block content is silently DISCARDED by Jinja. The button rendered. The handler never reached the browser. Two commits and a README paragraph said click-to-reveal worked, and the suite passed the entire time because nothing asserted against the SERVED page -- the template really did contain the code. Two guards, both confirmed to FAIL when the defect is reintroduced rather than merely added and assumed protective: * test_reveal_handler_actually_reaches_the_served_page greps the response * test_no_orphaned_markup_after_the_content_block guards the structure While moving it, caught a second instance of the same class: the explanatory comment I wrote for the fix contained a literal Jinja endblock tag, which Jinja would have parsed as a real tag and used to close the block early. KEPT-LANE ×. Wiping a kept booth required release-then-find-it-in-the-other- lane. That protected nothing and cost a hunt -- the board you just released is loose in a feed that turns over, and you have to go find it to finish a job you had already decided on. Protection now lives in the confirmation, which names the booth and says KEPT, instead of in the number of lanes you must traverse. Release stays as the reversible option. IN-BOOTH KEEP. `☆ keep` / `★ kept — release` beside "Wipe now", so promoting does not mean navigating back to the index. The booth page did not previously know its own kept state; it does now. Both post a `next` field to stay put -- and `next` is a form field, so it is attacker-controlled: only same-site absolute paths are honoured, with `//host`, schemes and backslashes refused, tested. 173 tests pass.
1062 lines
46 KiB
Python
1062 lines
46 KiB
Python
"""The Booth — a standing web server that renders drop-folders as ephemeral media booths.
|
||
|
||
Model (deliberately dead-simple, no database):
|
||
* The data dir holds one subfolder per "booth". A booth is created by a CC
|
||
session simply making a folder and dropping files in — there is no upload API.
|
||
* GET / -> index: a card per booth (scan of the data dir).
|
||
* GET /b/<name>/ -> if <name>/index.html exists, serve it verbatim; otherwise
|
||
auto-render a gallery of the images / webm-videos / audio in it.
|
||
* GET /b/<name>/<file> -> serve a file out of the booth (also feeds a custom index.html's assets).
|
||
* 24h TTL: a background sweeper wipes any booth untouched for TTL hours. A booth's
|
||
age is measured from the *newest* mtime in its tree, so it lives while it's being
|
||
worked on and self-destructs TTL hours after the last activity.
|
||
* KEPT BOOTHS: a booth containing the KEEP_MARKER dotfile (`.forever`) is exempt
|
||
from the sweep and renders in its own lane above the ephemeral grid. That is the
|
||
home for durable operator-facing boards — chiefly the standing link board agent
|
||
sessions post to, whose whole purpose is to survive longer than the scrollback
|
||
it replaces. Opt-in per booth, so the ephemeral default is unchanged and nobody
|
||
inherits a cleanup chore; `rm` the sentinel and the booth rejoins the sweep.
|
||
|
||
State is the filesystem — `ls ~/booth-data` tells you everything. That is the whole point.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import fcntl
|
||
import hashlib
|
||
import io
|
||
import os
|
||
import re
|
||
import secrets
|
||
import shutil
|
||
import time
|
||
import zipfile
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
from urllib.parse import quote, unquote
|
||
|
||
from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
|
||
from fastapi.responses import (
|
||
FileResponse,
|
||
HTMLResponse,
|
||
JSONResponse,
|
||
RedirectResponse,
|
||
Response,
|
||
)
|
||
from fastapi.templating import Jinja2Templates
|
||
|
||
try:
|
||
import markdown as _markdown
|
||
except ImportError: # optional dep — .md then degrades to a plain-text view
|
||
_markdown = None
|
||
|
||
TEMPLATES_DIR = Path(__file__).parent / "templates"
|
||
|
||
# Browser-playable media buckets. Anything else renders as a download link.
|
||
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".svg", ".bmp"}
|
||
VIDEO_EXTS = {".webm", ".mp4", ".ogv", ".m4v", ".mov"}
|
||
AUDIO_EXTS = {".mp3", ".wav", ".ogg", ".oga", ".flac", ".m4a", ".opus", ".aac"}
|
||
|
||
CAPTION_MAX = 800 # chars of a sidecar .txt caption we render
|
||
|
||
# Loose text docs that render as a readable in-booth page (not a download).
|
||
MARKDOWN_EXTS = {".md", ".markdown", ".mdown"}
|
||
TEXT_EXTS = {".txt", ".text", ".log"}
|
||
DOC_MAX_BYTES = 2 * 1024 * 1024 # above this, a doc is handed back raw, not rendered
|
||
|
||
# Sentinel dotfile that exempts a booth from the TTL sweep — see the "kept
|
||
# booths" note in the module docstring. A dotfile because the existing listing
|
||
# code already skips dotfiles, so it costs nothing in item counts or galleries,
|
||
# and because `touch`/`rm` is the entire user interface: no flag to remember, no
|
||
# state anywhere but the filesystem.
|
||
KEEP_MARKER = ".forever"
|
||
|
||
# Per-item blur state: one relative item path per line, like .pins is one id per
|
||
# line. Filesystem IS the state here, same as everything else in this service.
|
||
#
|
||
# ⚠⚠ BLUR IS COSMETIC, NOT ACCESS CONTROL. The file is still served at its own
|
||
# URL, still in the zip, still on disk. This hides an item from a glance — a
|
||
# shoulder, a screen-share, a scroll past something you did not want to see
|
||
# full-size — and nothing more. The Booth has no auth by design; if a thing
|
||
# must not be seen by whoever can reach port 8090, it must not be in a booth.
|
||
# Anyone who reads this marker as protection has misread it.
|
||
BLUR_FILE = ".blurred"
|
||
|
||
|
||
def read_blurred(booth: Path) -> set[str]:
|
||
"""Blurred item paths for a booth. Missing file -> empty set."""
|
||
try:
|
||
text = (booth / BLUR_FILE).read_text()
|
||
except (OSError, UnicodeDecodeError):
|
||
return set()
|
||
return {ln.strip() for ln in text.splitlines() if ln.strip()}
|
||
|
||
|
||
def set_blurred(booth: Path, rel: str, on: bool) -> set[str]:
|
||
"""Add or remove one item from the blur set. Atomic replace, so a crash
|
||
mid-write cannot leave a half-file that read_blurred would parse as a
|
||
shorter — and therefore more revealing — set. Returns the new set."""
|
||
current = read_blurred(booth)
|
||
if on:
|
||
current.add(rel)
|
||
else:
|
||
current.discard(rel)
|
||
path = booth / BLUR_FILE
|
||
if not current:
|
||
path.unlink(missing_ok=True)
|
||
return current
|
||
tmp = path.with_suffix(".tmp")
|
||
tmp.write_text("".join(f"{r}\n" for r in sorted(current)))
|
||
tmp.replace(path)
|
||
return current
|
||
|
||
# The link-board logic lives in booth/links.py (stdlib only) so the `booth` CLI
|
||
# can use it without pulling FastAPI in. Re-exported here because call sites and
|
||
# tests already reference these names through app.
|
||
from booth.asks import ( # noqa: E402
|
||
ANSWER_SUFFIX,
|
||
ASK_SUFFIX,
|
||
AskError,
|
||
is_answer_file,
|
||
is_ask_file,
|
||
list_asks,
|
||
load_ask,
|
||
valid_stem,
|
||
write_answer,
|
||
)
|
||
from booth.inline import ( # noqa: E402
|
||
form_id as ask_form_id,
|
||
has_placeholders,
|
||
place as place_asks,
|
||
)
|
||
from booth.links import ( # noqa: E402
|
||
LINK_LOCK,
|
||
LINKS_FILE,
|
||
PINS_FILE,
|
||
link_entry_id,
|
||
order_for_display,
|
||
parse_link_entries,
|
||
read_pins,
|
||
remove_link_entry,
|
||
toggle_pin,
|
||
)
|
||
|
||
|
||
def doc_kind(name: str) -> str | None:
|
||
"""'markdown' | 'text' | None — a booth file viewable as a readable page."""
|
||
ext = Path(name).suffix.lower()
|
||
if ext in MARKDOWN_EXTS:
|
||
return "markdown"
|
||
if ext in TEXT_EXTS:
|
||
return "text"
|
||
return None
|
||
|
||
|
||
def render_doc(text: str, kind: str) -> tuple[str, bool]:
|
||
"""(rendered, is_html). Markdown → HTML (fenced code, tables, sane lists);
|
||
plain text — or markdown when the lib is unavailable — → raw text for <pre>."""
|
||
if kind == "markdown" and _markdown is not None:
|
||
html = _markdown.markdown(text, extensions=["fenced_code", "tables", "sane_lists"])
|
||
return html, True
|
||
return text, False
|
||
|
||
|
||
def booth_image_names(child: Path) -> list[str]:
|
||
"""Image files in a booth, in gallery (sorted-rel) order — for viewer prev/next."""
|
||
return sorted(
|
||
p.relative_to(child).as_posix()
|
||
for p in child.rglob("*")
|
||
if p.is_file() and not p.name.startswith(".") and classify(p.name) == "image"
|
||
)
|
||
|
||
|
||
def classify(name: str) -> str:
|
||
"""image | video | audio | other, by extension."""
|
||
ext = Path(name).suffix.lower()
|
||
if ext in IMAGE_EXTS:
|
||
return "image"
|
||
if ext in VIDEO_EXTS:
|
||
return "video"
|
||
if ext in AUDIO_EXTS:
|
||
return "audio"
|
||
return "other"
|
||
|
||
|
||
def human_dur(seconds: float) -> str:
|
||
s = int(seconds)
|
||
if s <= 0:
|
||
return "expired"
|
||
h, rem = divmod(s, 3600)
|
||
m, _ = divmod(rem, 60)
|
||
if h and m:
|
||
return f"{h}h {m}m"
|
||
if h:
|
||
return f"{h}h"
|
||
if m:
|
||
return f"{m}m"
|
||
return "<1m"
|
||
|
||
|
||
def _newest_mtime(path: Path) -> float:
|
||
"""Newest mtime among a folder and everything under it."""
|
||
try:
|
||
newest = path.stat().st_mtime
|
||
except OSError:
|
||
return 0.0
|
||
for p in path.rglob("*"):
|
||
try:
|
||
m = p.stat().st_mtime
|
||
except OSError:
|
||
continue
|
||
if m > newest:
|
||
newest = m
|
||
return newest
|
||
|
||
|
||
def booth_age_seconds(path: Path, now: float | None = None) -> float:
|
||
now = time.time() if now is None else now
|
||
return now - _newest_mtime(path)
|
||
|
||
|
||
def is_expired(path: Path, ttl_seconds: float, now: float | None = None) -> bool:
|
||
"""Pure age question. Deliberately does NOT consider the keep sentinel.
|
||
|
||
Expiry arithmetic (what `expires_in` renders) and reaper policy (what
|
||
actually gets deleted) are kept apart so they cannot drift into each other.
|
||
Only `sweep_once` honours the pin.
|
||
"""
|
||
return booth_age_seconds(path, now) > ttl_seconds
|
||
|
||
|
||
def is_kept(path: Path) -> bool:
|
||
"""True if this booth carries the keep sentinel and must never be swept."""
|
||
return (path / KEEP_MARKER).exists()
|
||
|
||
|
||
def sweep_once(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[str]:
|
||
"""Wipe every direct-child booth older than the TTL. Returns names wiped.
|
||
|
||
Only ever removes direct children of data_dir (never data_dir itself), and
|
||
skips dotfolders so a stray control dir can opt out.
|
||
|
||
A booth carrying KEEP_MARKER is exempt no matter how stale it is. That is
|
||
the one escape hatch from the 24h contract, and it is opt-in per booth: the
|
||
default stays ephemeral, so nobody inherits a cleanup chore they did not ask
|
||
for. Removing the sentinel hands the booth straight back to the sweeper.
|
||
"""
|
||
wiped: list[str] = []
|
||
if not data_dir.is_dir():
|
||
return wiped
|
||
for child in data_dir.iterdir():
|
||
if not child.is_dir() or child.name.startswith("."):
|
||
continue
|
||
try:
|
||
if is_kept(child):
|
||
continue
|
||
if is_expired(child, ttl_seconds, now):
|
||
shutil.rmtree(child)
|
||
wiped.append(child.name)
|
||
except OSError:
|
||
pass
|
||
return wiped
|
||
|
||
|
||
def list_booths(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[dict]:
|
||
now = time.time() if now is None else now
|
||
booths: list[dict] = []
|
||
if not data_dir.is_dir():
|
||
return booths
|
||
for child in data_dir.iterdir():
|
||
if not child.is_dir() or child.name.startswith("."):
|
||
continue
|
||
files = [
|
||
p for p in child.rglob("*")
|
||
if p.is_file() and not p.name.startswith(".")
|
||
and not is_ask_file(p.name) and not is_answer_file(p.name)
|
||
]
|
||
# Asks are questions, not items: counted separately so the index can
|
||
# flag a booth that is waiting on the operator.
|
||
asks = list_asks(child)
|
||
kinds = {"image": 0, "video": 0, "audio": 0, "other": 0}
|
||
thumb_url = None
|
||
for f in files:
|
||
k = classify(f.name)
|
||
kinds[k] += 1
|
||
if k == "image" and thumb_url is None:
|
||
thumb_url = quote(f.relative_to(child).as_posix(), safe="/")
|
||
mtime = _newest_mtime(child)
|
||
booths.append(
|
||
{
|
||
"name": child.name,
|
||
"name_url": quote(child.name, safe=""),
|
||
"count": len(files),
|
||
"kinds": kinds,
|
||
"thumb_url": thumb_url,
|
||
# If the cover image is blurred inside the booth, blur it on the
|
||
# index too — otherwise the front page cheerfully displays the
|
||
# exact thing someone asked to hide.
|
||
"thumb_blurred": thumb_url is not None
|
||
and unquote(thumb_url) in read_blurred(child),
|
||
"has_index": (child / "index.html").is_file(),
|
||
"uploaded": (child / UPLOAD_MARKER).exists(),
|
||
"kept": is_kept(child),
|
||
"asks_total": len(asks),
|
||
"asks_open": sum(1 for a in asks if a["answer"] is None and not a["error"]),
|
||
"expires_in": max(0.0, ttl_seconds - (now - mtime)),
|
||
"mtime": mtime,
|
||
}
|
||
)
|
||
booths.sort(key=lambda b: b["mtime"], reverse=True)
|
||
return booths
|
||
|
||
|
||
def build_gallery(child: Path) -> list[dict]:
|
||
"""Files in a booth as render items, with caption sidecars folded in.
|
||
|
||
A `<file>.txt` (e.g. `a.png.txt`) or a same-stem `<stem>.txt` (e.g. `a.txt`
|
||
next to `a.png`) is consumed as that item's caption rather than shown itself —
|
||
the natural way to label an A/B pair.
|
||
"""
|
||
all_files = [
|
||
p for p in child.rglob("*")
|
||
if p.is_file() and not p.name.startswith(".")
|
||
# `*.ask.json` / `*.answer.json` render as the asks panel, not as tiles
|
||
and not is_ask_file(p.name) and not is_answer_file(p.name)
|
||
]
|
||
by_rel = {p.relative_to(child).as_posix(): p for p in all_files}
|
||
caption: dict[str, str] = {}
|
||
sidecars: set[str] = set()
|
||
|
||
for rel, p in by_rel.items():
|
||
if not rel.lower().endswith(".txt"):
|
||
continue
|
||
target = None
|
||
base_full = rel[:-4] # strip ".txt" -> "a.png.txt" => "a.png"
|
||
if base_full in by_rel:
|
||
target = base_full
|
||
else: # "a.txt" beside "a.png"
|
||
parent = str(Path(rel).parent)
|
||
stem = Path(rel).stem
|
||
for q_rel, q in by_rel.items():
|
||
if q_rel == rel:
|
||
continue
|
||
if (
|
||
str(Path(q_rel).parent) == parent
|
||
and Path(q_rel).stem == stem
|
||
and classify(q.name) != "other"
|
||
):
|
||
target = q_rel
|
||
break
|
||
if target is not None:
|
||
try:
|
||
caption[target] = p.read_text(errors="replace").strip()[:CAPTION_MAX]
|
||
except OSError:
|
||
pass
|
||
sidecars.add(rel)
|
||
|
||
blurred = read_blurred(child)
|
||
items = []
|
||
for rel in sorted(by_rel):
|
||
if rel in sidecars:
|
||
continue
|
||
p = by_rel[rel]
|
||
dkind = doc_kind(p.name)
|
||
rendered = None
|
||
rendered_html = False
|
||
# Pre-render docs so the gallery can show them INLINE (collapsible)
|
||
# instead of linking out to a separate page. Bounded by DOC_MAX_BYTES:
|
||
# a giant log stays a download link rather than being inlined into every
|
||
# index render. Markdown → HTML (marked safe in the template); plain text
|
||
# is returned RAW and the template escapes it inside <pre> — pre-escaping
|
||
# here would double-encode under Jinja autoescape.
|
||
if dkind is not None:
|
||
try:
|
||
if p.stat().st_size <= DOC_MAX_BYTES:
|
||
text = p.read_text(errors="replace")
|
||
rendered, rendered_html = render_doc(text, dkind)
|
||
except OSError:
|
||
rendered = None
|
||
items.append(
|
||
{
|
||
"name": rel,
|
||
"kind": classify(p.name),
|
||
"doc": dkind,
|
||
"url": quote(rel, safe="/"),
|
||
"caption": caption.get(rel),
|
||
"rendered": rendered,
|
||
"rendered_html": rendered_html,
|
||
"blurred": rel in blurred,
|
||
}
|
||
)
|
||
return items
|
||
|
||
|
||
def zip_booth(booth: Path) -> bytes:
|
||
"""Zip a booth's whole tree (dotfiles excluded) into an in-memory archive.
|
||
|
||
Lets a booth be downloaded as one artifact regardless of shape — the case a
|
||
verbatim `index.html` booth (e.g. a rendered brief + its assets) has no
|
||
per-file download affordance for, since the page is served raw.
|
||
"""
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for p in sorted(booth.rglob("*")):
|
||
if p.is_file() and not p.name.startswith("."):
|
||
zf.write(p, p.relative_to(booth).as_posix())
|
||
return buf.getvalue()
|
||
|
||
|
||
def _zip_filename(name: str) -> str:
|
||
"""A Content-Disposition-safe `<booth>.zip` (strip quotes/control chars)."""
|
||
safe = "".join(c for c in name if c.isprintable() and c != '"')
|
||
return f"{safe or 'booth'}.zip"
|
||
|
||
|
||
# ---- verbatim-index.html wrapper -------------------------------------------
|
||
|
||
# Mirror of base.html's favicon (the app templates set it there; this is the copy
|
||
# injected into a booth's *verbatim* index.html so a raw page inherits the same
|
||
# icon). Keep the two in sync if the Booth's icon ever changes.
|
||
FAVICON_HREF = (
|
||
"data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 32 32'"
|
||
"%3E%3Crect width='32' height='32' rx='7' fill='%23171a23'/%3E%3Ccircle cx='16' "
|
||
"cy='16' r='6' fill='none' stroke='%2342dcd1' stroke-width='2.5'/%3E%3Ccircle "
|
||
"cx='16' cy='16' r='2.2' fill='%2342dcd1'/%3E%3C/svg%3E"
|
||
)
|
||
FAVICON_LINK = f'<link rel="icon" href="{FAVICON_HREF}">'
|
||
|
||
# A self-contained floating "back to all booths" chip injected into verbatim
|
||
# booths. Scoped class + fixed positioning + max z-index so it overlays the raw
|
||
# page without touching its layout; hidden in print so downloaded reports stay clean.
|
||
_BACK_CHIP = (
|
||
'<a href="/" class="booth-nav-home" aria-label="back to all booths">‹ all booths</a>'
|
||
# top-right: empty on left-aligned report layouts (a top-left chip clips the
|
||
# page title), and consistent with the zoom view's top-right back affordance.
|
||
"<style>.booth-nav-home{position:fixed;top:0;right:0;z-index:2147483647;"
|
||
"display:inline-block;margin:.6rem;padding:.34rem .72rem;"
|
||
"font:600 13px/1.25 ui-sans-serif,system-ui,-apple-system,'Segoe UI',Roboto,sans-serif;"
|
||
"color:#dfe7ef;text-decoration:none;letter-spacing:.01em;"
|
||
"background:rgba(20,23,32,.82);border:1px solid rgba(66,220,209,.35);border-radius:8px;"
|
||
"-webkit-backdrop-filter:blur(6px);backdrop-filter:blur(6px);"
|
||
"box-shadow:0 2px 10px rgba(0,0,0,.35);transition:background .18s,border-color .18s}"
|
||
".booth-nav-home:hover{background:rgba(28,33,46,.95);border-color:rgba(66,220,209,.75)}"
|
||
"@media print{.booth-nav-home{display:none}}</style>"
|
||
)
|
||
|
||
# A booth's own index.html is served VERBATIM, so the asks panel — which lives in
|
||
# the auto-gallery template — can never appear on it. Without this chip an ask
|
||
# posted into a custom-report booth is INVISIBLE to the operator with nothing to
|
||
# say so (found 2026-09-09 on `emmie-anchor`: valid ask, CLI listed it, page
|
||
# showed nothing). Same injection mechanism as the back chip; it links to the
|
||
# standalone /asks page, which renders the real forms.
|
||
def asks_chip(name: str, open_count: int, href: str | None = None) -> str:
|
||
if open_count < 1:
|
||
return ""
|
||
label = f"? {open_count} open ask" + ("" if open_count == 1 else "s")
|
||
href = href or f"/b/{quote(name, safe='')}/asks"
|
||
return (
|
||
f'<a href="{href}" class="booth-nav-asks">{label}</a>'
|
||
"<style>.booth-nav-asks{position:fixed;top:0;right:7.2rem;z-index:2147483647;"
|
||
"display:inline-block;margin:.6rem;padding:.34rem .72rem;"
|
||
"font:700 13px/1.25 ui-sans-serif,system-ui,-apple-system,'Segoe UI',Roboto,sans-serif;"
|
||
"color:#171a23;text-decoration:none;letter-spacing:.01em;"
|
||
"background:#ffe14e;border:1px solid #ffe14e;border-radius:8px;"
|
||
"box-shadow:0 2px 10px rgba(0,0,0,.35);transition:filter .18s}"
|
||
".booth-nav-asks:hover{filter:brightness(1.08)}"
|
||
"@media print{.booth-nav-asks{display:none}}</style>"
|
||
)
|
||
|
||
|
||
WRAP_MAX_BYTES = 8 * 1024 * 1024 # above this, serve the verbatim page raw (unwrapped)
|
||
|
||
_ICON_RE = re.compile(r"<link\b[^>]*\brel\s*=\s*[\"']?[^\"'>]*icon", re.IGNORECASE)
|
||
_HEAD_CLOSE_RE = re.compile(r"</head\s*>", re.IGNORECASE)
|
||
_HTML_OPEN_RE = re.compile(r"<html\b[^>]*>", re.IGNORECASE)
|
||
_DOCTYPE_RE = re.compile(r"<!doctype[^>]*>", re.IGNORECASE)
|
||
_BODY_CLOSE_RE = re.compile(r"</body\s*>", re.IGNORECASE)
|
||
_HTML_CLOSE_RE = re.compile(r"</html\s*>", re.IGNORECASE)
|
||
|
||
|
||
def _insert_before(html: str, pattern: re.Pattern, snippet: str) -> tuple[str, bool]:
|
||
m = pattern.search(html)
|
||
if m:
|
||
return html[: m.start()] + snippet + html[m.start() :], True
|
||
return html, False
|
||
|
||
|
||
def _insert_after(html: str, pattern: re.Pattern, snippet: str) -> tuple[str, bool]:
|
||
m = pattern.search(html)
|
||
if m:
|
||
return html[: m.end()] + snippet + html[m.end() :], True
|
||
return html, False
|
||
|
||
|
||
def wrap_verbatim_html(html: str, favicon_link: str = FAVICON_LINK, extra: str = "") -> str:
|
||
"""Inject a floating 'all booths' back-chip — and the Booth favicon, if the page
|
||
declares none — into a booth's verbatim index.html, without altering the page's
|
||
rendered content.
|
||
|
||
Robust to the compact HTML real booths use (`<!doctype html><meta charset><title>
|
||
<style>…content`, no explicit head/body). The two hard constraints:
|
||
* NEVER put anything ahead of a leading <!doctype> — that forces quirks mode.
|
||
* Keep the charset <meta> within the first 1024 bytes so it's still honoured.
|
||
So the favicon lands at the first head-ish seam (before </head>, else after
|
||
<html>, else right after the doctype — a ~250B link keeps charset in range), and
|
||
the fixed-position chip is appended at the END of the document (before </body> /
|
||
</html> or appended), which renders top-left regardless and disturbs nothing.
|
||
"""
|
||
if favicon_link and not _ICON_RE.search(html):
|
||
for inserter, pat in (
|
||
(_insert_before, _HEAD_CLOSE_RE), # inside an explicit <head>
|
||
(_insert_after, _HTML_OPEN_RE), # top of an explicit <html>
|
||
(_insert_after, _DOCTYPE_RE), # right after the doctype (compact HTML)
|
||
):
|
||
html, done = inserter(html, pat, favicon_link)
|
||
if done:
|
||
break
|
||
else:
|
||
html = favicon_link + html # bare fragment, no doctype: safe to prepend
|
||
|
||
chips = _BACK_CHIP + (extra or "")
|
||
for pat in (_BODY_CLOSE_RE, _HTML_CLOSE_RE):
|
||
html, done = _insert_before(html, pat, chips)
|
||
if done:
|
||
break
|
||
else:
|
||
html = html + chips # no </body>/</html>: append to the end
|
||
return html
|
||
|
||
|
||
# ---- uploads (browser drop-off for pickup) ---------------------------------
|
||
|
||
UPLOAD_MARKER = ".uploaded" # dotfile stamped into upload booths (excluded from listings)
|
||
|
||
# Friendly, unambiguous words for human-readable pickup ids (4-wombat / star-84).
|
||
PICKUP_WORDS = (
|
||
"wombat otter panda koala tiger walrus gecko heron badger beaver falcon marmot "
|
||
"lemur narwhal ocelot puffin quokka raccoon tapir urchin vulture weasel yak zebra "
|
||
"alpaca bison cobra dingo egret ferret gibbon hare ibis jaguar llama moose newt "
|
||
"osprey possum quail robin seal toad viper wren lynx mole swan crane finch sloth "
|
||
"shrew stoat skunk heronry orca walnut sparrow "
|
||
"star comet moon cloud river maple cedar birch fern moss reef dune mesa cove glade "
|
||
"brook pine cedarwood kelp coral amber opal jade onyx slate flint ember spark frost "
|
||
"storm tide wave ridge peak vale marsh delta atoll canyon fjord geyser lagoon prairie "
|
||
"anchor beacon lantern kettle copper brass velvet cobalt indigo crimson violet olive "
|
||
"hazel cocoa mango guava papaya plum kiwi lime pear quince radish turnip acorn clover "
|
||
"thistle poppy aster dahlia iris lily sage thyme basil clove nutmeg ginger honey"
|
||
).split()
|
||
|
||
|
||
def safe_upload_name(name: str, fallback: str) -> str:
|
||
"""Reduce a client-supplied filename to a safe basename (no path, no hidden)."""
|
||
base = (name or "").replace("\\", "/").split("/")[-1].strip()
|
||
base = base.lstrip(".") # a leading dot would hide the file from every listing
|
||
return base[:200] or fallback
|
||
|
||
|
||
def _dedupe_name(name: str, used: set) -> str:
|
||
if name not in used:
|
||
return name
|
||
stem, dot, ext = name.partition(".")
|
||
i = 1
|
||
while f"{stem}-{i}{dot}{ext}" in used:
|
||
i += 1
|
||
return f"{stem}-{i}{dot}{ext}"
|
||
|
||
|
||
def generate_pickup_id(exists) -> str:
|
||
"""A human-readable id like '4-wombat' or 'star-84'. `exists(name)->bool` gates collisions."""
|
||
for _ in range(400):
|
||
word = secrets.choice(PICKUP_WORDS)
|
||
num = secrets.randbelow(99) + 1
|
||
name = f"{num}-{word}" if secrets.randbelow(2) else f"{word}-{num}"
|
||
if not exists(name):
|
||
return name
|
||
# astronomically unlikely fallback: two words keep it human-readable
|
||
while True:
|
||
name = f"{secrets.choice(PICKUP_WORDS)}-{secrets.choice(PICKUP_WORDS)}-{secrets.randbelow(999) + 1}"
|
||
if not exists(name):
|
||
return name
|
||
|
||
|
||
def create_app(
|
||
data_dir,
|
||
ttl_hours: float = 24.0,
|
||
host_label: str = "",
|
||
start_sweeper: bool = True,
|
||
sweep_interval_s: int = 900,
|
||
max_upload_mb: float = 1024.0,
|
||
max_files: int = 50,
|
||
) -> FastAPI:
|
||
data_dir = Path(data_dir).expanduser().resolve()
|
||
data_dir.mkdir(parents=True, exist_ok=True)
|
||
ttl_seconds = ttl_hours * 3600.0
|
||
max_upload_bytes = int(max_upload_mb * 1024 * 1024)
|
||
|
||
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
|
||
templates.env.filters["dur"] = human_dur
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
task = None
|
||
if start_sweeper:
|
||
|
||
async def loop():
|
||
while True:
|
||
try:
|
||
wiped = sweep_once(data_dir, ttl_seconds)
|
||
if wiped:
|
||
print(f"[booth] swept {len(wiped)} expired: {', '.join(wiped)}", flush=True)
|
||
except Exception as exc: # never let the sweeper die
|
||
print(f"[booth] sweep error: {exc}", flush=True)
|
||
await asyncio.sleep(sweep_interval_s)
|
||
|
||
task = asyncio.create_task(loop())
|
||
try:
|
||
yield
|
||
finally:
|
||
if task is not None:
|
||
task.cancel()
|
||
|
||
app = FastAPI(title="The Booth", lifespan=lifespan)
|
||
|
||
ttl_display = int(ttl_hours) if float(ttl_hours).is_integer() else ttl_hours
|
||
base_ctx = {
|
||
"ttl_hours": ttl_display,
|
||
"host": host_label,
|
||
"data_dir": str(data_dir),
|
||
"keep_marker": KEEP_MARKER, # shown in the kept lane so the mechanism is discoverable
|
||
}
|
||
|
||
def resolve_booth(name: str) -> Path:
|
||
if not name or name.startswith(".") or "/" in name or "\\" in name or ".." in name:
|
||
raise HTTPException(status_code=404, detail="no such booth")
|
||
candidate = data_dir / name
|
||
try:
|
||
resolved = candidate.resolve()
|
||
except OSError:
|
||
raise HTTPException(status_code=404, detail="no such booth")
|
||
# resolved.parent must be the data dir itself — blocks symlink escape + nesting.
|
||
if resolved.parent != data_dir or not resolved.is_dir():
|
||
raise HTTPException(status_code=404, detail="no such booth")
|
||
return resolved
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index(request: Request):
|
||
# Two lanes, split here rather than in the template: kept boards are a
|
||
# different KIND of thing from the ephemeral churn — durable, deliberate,
|
||
# operator-facing — and burying them in a feed that turns over daily is
|
||
# exactly how they would get lost, which is the problem they exist to
|
||
# solve. Kept renders first.
|
||
everything = list_booths(data_dir, ttl_seconds)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"index.html",
|
||
{
|
||
**base_ctx,
|
||
"kept": [b for b in everything if b["kept"]],
|
||
"booths": [b for b in everything if not b["kept"]],
|
||
},
|
||
)
|
||
|
||
@app.get("/healthz")
|
||
def healthz():
|
||
return {"ok": True, "ttl_hours": ttl_hours, "booths": len(list_booths(data_dir, ttl_seconds))}
|
||
|
||
@app.get("/b/{name}", include_in_schema=False)
|
||
def booth_redirect(name: str):
|
||
resolve_booth(name)
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=307)
|
||
|
||
@app.get("/b/{name}/", response_class=HTMLResponse)
|
||
def booth_view(request: Request, name: str, download: int = 0):
|
||
booth = resolve_booth(name)
|
||
if download:
|
||
# whole-booth zip — the download path for a verbatim index.html booth
|
||
# (which has no gallery/per-file chrome), and a "download all" for any.
|
||
return Response(
|
||
content=zip_booth(booth),
|
||
media_type="application/zip",
|
||
headers={"Content-Disposition": f'attachment; filename="{_zip_filename(name)}"'},
|
||
)
|
||
own_index = booth / "index.html"
|
||
if own_index.is_file():
|
||
# Serve the operator's verbatim report, but inject a floating
|
||
# back-to-booths chip + the Booth favicon (if it declares none) so a
|
||
# raw page still has a way home. Small HTML -> read + wrap in memory;
|
||
# a pathological large file falls back to serving raw, unwrapped.
|
||
try:
|
||
if own_index.stat().st_size <= WRAP_MAX_BYTES:
|
||
raw = own_index.read_text(encoding="utf-8", errors="replace")
|
||
# Asks render INLINE, where the report author put them (or
|
||
# appended, if they marked nothing) — a question about an
|
||
# artifact belongs beside that artifact, not on another page.
|
||
body, tail = inject_asks(name, booth, raw)
|
||
return HTMLResponse(wrap_verbatim_html(body, extra=tail))
|
||
except OSError:
|
||
pass
|
||
return FileResponse(str(own_index), media_type="text/html")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"booth.html",
|
||
{
|
||
**base_ctx,
|
||
"name": name,
|
||
"name_url": quote(name, safe=""),
|
||
# The page could not previously tell keep from release, so it
|
||
# offered neither and you had to go back to the index.
|
||
"kept": is_kept(booth),
|
||
# links.md is rendered AS the board below, so it must not also
|
||
# appear as a markdown doc tile — that would show the same
|
||
# content twice, once interactive and once not.
|
||
"items": [
|
||
it for it in build_gallery(booth)
|
||
if not ((booth / LINKS_FILE).is_file() and it["name"] == LINKS_FILE)
|
||
],
|
||
# A booth carrying links.md is the standing link board: render
|
||
# its rows as real UI (link, provenance, pin, per-row + bulk
|
||
# remove) instead of a markdown blob you can only edit by hand.
|
||
# Ordered pinned-first then newest-first, each row stamped with a
|
||
# `pinned` flag. Empty list for every other booth, so the template
|
||
# branch simply does not fire.
|
||
"board": (
|
||
order_for_display(
|
||
parse_link_entries((booth / LINKS_FILE).read_text()),
|
||
read_pins(booth),
|
||
)
|
||
if (booth / LINKS_FILE).is_file() else []
|
||
),
|
||
# Asks: multiple-choice questions a session left for the
|
||
# operator, rendered as forms above the gallery (open ones)
|
||
# or as their recorded answer. See booth/asks.py.
|
||
"asks": list_asks(booth),
|
||
"uploaded": (booth / UPLOAD_MARKER).exists(),
|
||
"expires_in": max(0.0, ttl_seconds - booth_age_seconds(booth)),
|
||
},
|
||
)
|
||
|
||
@app.post("/b/{name}/answer")
|
||
async def booth_answer(request: Request, name: str):
|
||
"""Record the operator's answer to one ask: validates every choice
|
||
against the ask and writes `<stem>.answer.json` atomically.
|
||
Re-submitting overwrites — the sidecar is the current answer.
|
||
|
||
Form fields: `ask` (stem); single-question → `choice` + `notes`;
|
||
multi-question → `choice.<key>` per question, optional `notes.<key>`,
|
||
plus the form-level `notes`. 404 for an unknown/invalid stem, 400 for
|
||
a missing choice or one the ask does not offer.
|
||
"""
|
||
booth = resolve_booth(name)
|
||
form = await request.form()
|
||
ask = form.get("ask")
|
||
if not isinstance(ask, str) or not valid_stem(ask) or not (booth / f"{ask}{ASK_SUFFIX}").is_file():
|
||
raise HTTPException(status_code=404, detail="no such ask")
|
||
who = request.client.host if request.client else ""
|
||
try:
|
||
spec = load_ask(booth, ask)
|
||
if spec["multi"]:
|
||
choice = {q["key"]: form.get(f"choice.{q['key']}") for q in spec["questions"]}
|
||
qnotes = {q["key"]: form.get(f"notes.{q['key']}") for q in spec["questions"]}
|
||
write_answer(booth, ask, choice, form.get("notes", ""), who=who, qnotes=qnotes)
|
||
else:
|
||
write_answer(booth, ask, form.get("choice"), form.get("notes", ""), who=who)
|
||
except AskError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
# Land where the form was: the standalone /asks page for a verbatim booth
|
||
# (its own index.html cannot show the recorded answer), else the booth.
|
||
base = f"/b/{quote(name, safe='')}/"
|
||
if form.get("back") == "asks":
|
||
base = f"/b/{quote(name, safe='')}/asks"
|
||
return RedirectResponse(url=f"{base}#ask-{quote(ask, safe='')}", status_code=303)
|
||
|
||
_frag = templates.env.get_template("_ask_inline.html").module
|
||
|
||
def inject_asks(name: str, booth: Path, html: str) -> tuple[str, str]:
|
||
"""(body, tail) for a verbatim booth: placeholders substituted in place,
|
||
and whatever still has to be appended before </body>.
|
||
|
||
Marked-up pages get each fragment exactly where the author put it. An
|
||
unmarked page gets the whole ask appended — an ask is NEVER invisible,
|
||
which is the guarantee; markup only moves it somewhere better. A stem
|
||
whose questions were placed but whose submit block was not gets that
|
||
block appended, so a scattered form is always submittable.
|
||
"""
|
||
asks = list_asks(booth)
|
||
if not asks:
|
||
return html, ""
|
||
url = quote(name, safe="")
|
||
|
||
seen: set[str] = set()
|
||
|
||
def render(kind: str, ask: dict, key: str | None) -> str:
|
||
fid = ask_form_id(ask["stem"])
|
||
if kind == "whole":
|
||
frag = str(_frag.whole(ask, fid, url))
|
||
elif kind == "submit":
|
||
frag = str(_frag.submit(ask, fid, url))
|
||
else:
|
||
q = next(q for q in ask["questions"] if q.get("key") == key)
|
||
frag = str(_frag.question(ask, q, fid, url))
|
||
# An anchor on the FIRST fragment of each stem, wherever it landed,
|
||
# so the floating chip can jump to it on a long report. Computed
|
||
# here rather than in the macros because only the caller knows
|
||
# which fragment came first.
|
||
if ask["stem"] not in seen:
|
||
seen.add(ask["stem"])
|
||
frag = f'<a id="bk-ask-{ask["stem"]}-top"></a>' + frag
|
||
return frag
|
||
|
||
tail = [str(_frag.styles())]
|
||
if has_placeholders(html):
|
||
html, placed, submitted = place_asks(html, asks, render)
|
||
for a in asks:
|
||
keys = placed.get(a["stem"])
|
||
if keys is None:
|
||
tail.append(render("whole", a, None)) # unmarked: never dropped
|
||
continue
|
||
if a["error"]:
|
||
continue
|
||
if None not in keys:
|
||
# Partially marked up: append every question the author did
|
||
# NOT place. A multi-question ask needs all of them or the
|
||
# POST is a 400 — met only after the operator fills it in.
|
||
for q in a["questions"]:
|
||
if q.get("key") not in keys:
|
||
tail.append(render("question", a, q.get("key")))
|
||
if a["stem"] not in submitted:
|
||
tail.append(render("submit", a, None)) # scattered but submittable
|
||
else:
|
||
for a in asks:
|
||
tail.append(render("whole", a, None))
|
||
|
||
# The chip is now a JUMP LINK to the inline block, not a way out to a
|
||
# separate page: on a long report the question can be well below the
|
||
# fold, and "there is a question waiting" still has to be visible at
|
||
# first paint.
|
||
first_open = next((a for a in asks if a["answer"] is None and not a["error"]), None)
|
||
open_n = sum(1 for a in asks if a["answer"] is None and not a["error"])
|
||
if first_open is not None:
|
||
tail.append(asks_chip(name, open_n, href=f'#bk-ask-{first_open["stem"]}-top'))
|
||
return html, "".join(tail)
|
||
|
||
@app.get("/b/{name}/asks", response_class=HTMLResponse)
|
||
def booth_asks_page(request: Request, name: str):
|
||
"""The asks panel on its own page. Reachable from any booth, and the ONLY
|
||
place a verbatim-index.html booth can show its asks — that page is served
|
||
untouched by design, so the inline panel never renders there."""
|
||
booth = resolve_booth(name)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"asks.html",
|
||
{**base_ctx, "name": name, "name_url": quote(name, safe=""),
|
||
"asks": list_asks(booth), "asks_page": True},
|
||
)
|
||
|
||
@app.get("/b/{name}/view", response_class=HTMLResponse)
|
||
def booth_view_file(request: Request, name: str, f: str):
|
||
booth = resolve_booth(name)
|
||
try:
|
||
target = (booth / f).resolve()
|
||
except OSError:
|
||
raise HTTPException(status_code=404, detail="no such file")
|
||
if not str(target).startswith(str(booth) + os.sep) or not target.is_file():
|
||
raise HTTPException(status_code=404, detail="no such file")
|
||
file_url = quote(f, safe="/")
|
||
common = {**base_ctx, "name": name, "name_url": quote(name, safe=""), "file": f, "file_url": file_url}
|
||
if classify(target.name) == "image":
|
||
# prev/next image nav (wraps around; only when >1 image in the booth)
|
||
names = booth_image_names(booth)
|
||
prev_url = next_url = None
|
||
if f in names and len(names) > 1:
|
||
i = names.index(f)
|
||
prev_url = quote(names[(i - 1) % len(names)], safe="/")
|
||
next_url = quote(names[(i + 1) % len(names)], safe="/")
|
||
return templates.TemplateResponse(
|
||
request, "view.html", {**common, "prev_url": prev_url, "next_url": next_url}
|
||
)
|
||
# .md renders, .txt/.log show as text — viewable in-booth, no download
|
||
dk = doc_kind(target.name)
|
||
if dk:
|
||
try:
|
||
if target.stat().st_size <= DOC_MAX_BYTES:
|
||
body, is_html = render_doc(target.read_text(encoding="utf-8", errors="replace"), dk)
|
||
return templates.TemplateResponse(
|
||
request, "doc.html", {**common, "kind": dk, "body": body, "is_html": is_html}
|
||
)
|
||
except OSError:
|
||
raise HTTPException(status_code=404, detail="no such file")
|
||
# nothing to render — hand back the raw file
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/{file_url}", status_code=307)
|
||
|
||
@app.get("/b/{name}/{filepath:path}")
|
||
def booth_file(name: str, filepath: str, dl: int = 0):
|
||
booth = resolve_booth(name)
|
||
try:
|
||
target = (booth / filepath).resolve()
|
||
except OSError:
|
||
raise HTTPException(status_code=404, detail="no such file")
|
||
if not str(target).startswith(str(booth) + os.sep) or not target.is_file():
|
||
raise HTTPException(status_code=404, detail="no such file")
|
||
# ?dl=1 forces a download (Content-Disposition: attachment) instead of the
|
||
# browser rendering inline — the fix for html/md/text that otherwise opens
|
||
# in-page with no easy "save".
|
||
if dl:
|
||
return FileResponse(str(target), filename=target.name)
|
||
return FileResponse(str(target))
|
||
|
||
@app.post("/upload")
|
||
async def upload(files: list[UploadFile] = File(...)):
|
||
"""Browser/curl drop-off: files land in a new booth with a human-readable
|
||
pickup id (e.g. 4-wombat), sweep-expiring in the usual TTL. Redirects (303)
|
||
to the pickup page; curl clients read the Location header for the id."""
|
||
files = [f for f in files if f and f.filename]
|
||
if not files:
|
||
raise HTTPException(status_code=400, detail="no files uploaded")
|
||
if len(files) > max_files:
|
||
raise HTTPException(status_code=413, detail=f"too many files (max {max_files})")
|
||
|
||
booth_id = generate_pickup_id(lambda n: (data_dir / n).exists())
|
||
dest = data_dir / booth_id
|
||
dest.mkdir(parents=True)
|
||
(dest / UPLOAD_MARKER).write_text("") # stamp as an upload (dotfile, not listed)
|
||
|
||
total = 0
|
||
used: set = {UPLOAD_MARKER}
|
||
try:
|
||
for i, f in enumerate(files):
|
||
name = _dedupe_name(safe_upload_name(f.filename, f"file-{i + 1}"), used)
|
||
used.add(name)
|
||
with (dest / name).open("wb") as out:
|
||
while chunk := await f.read(1024 * 1024):
|
||
total += len(chunk)
|
||
if total > max_upload_bytes:
|
||
raise HTTPException(
|
||
status_code=413,
|
||
detail=f"upload too large (max {max_upload_mb:g} MB)",
|
||
)
|
||
out.write(chunk)
|
||
await f.close()
|
||
except Exception:
|
||
shutil.rmtree(dest, ignore_errors=True) # never leave a half-written booth
|
||
raise
|
||
|
||
return RedirectResponse(url=f"/b/{quote(booth_id, safe='')}/", status_code=303)
|
||
|
||
# Releasing a kept board. The kept lane has no wipe control on purpose —
|
||
# destroying a durable board should not be one misclick — but "deliberate"
|
||
# had been built as "impossible from the UI": the only ways out were ssh or
|
||
# a hand-written API call. These two routes make the release step reachable
|
||
# while keeping deletion two deliberate acts (release, then wipe).
|
||
#
|
||
# NOTE ON THE TTL, which is not intuitive: removing the sentinel BUMPS the
|
||
# booth directory's mtime, and booth_age_seconds reads the newest mtime in
|
||
# the tree — so a released board's clock resets to zero and it survives
|
||
# another full TTL. "Unkeep and let the sweeper take it" therefore does NOT
|
||
# delete promptly. Release is the step that makes the × available; the ×
|
||
# is what deletes. Anything relying on release-then-sweep is relying on a
|
||
# 24h delay it probably did not intend.
|
||
|
||
@app.post("/b/{name}/unlink")
|
||
def board_unlink(name: str, entry: str = Form(...)):
|
||
"""Remove ONE row from a link board, by content id.
|
||
|
||
Deliberately not by index: the board is append-only and multi-writer,
|
||
so between rendering the page and clicking × another session may have
|
||
posted. A content id either matches the row the operator saw or matches
|
||
nothing — it can never resolve to a neighbour.
|
||
"""
|
||
removed = remove_link_entry(resolve_booth(name), entry)
|
||
if removed is None:
|
||
# Already gone (double-click, stale tab, someone else pruned it).
|
||
# Not an error worth a 404 page — the desired end state holds.
|
||
pass
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)
|
||
|
||
@app.post("/b/{name}/unlink-many")
|
||
def board_unlink_many(name: str, sel: list[str] = Form(default=[])):
|
||
"""Remove SEVERAL rows in one go — the multi-select delete.
|
||
|
||
Each `sel` is a content id (same identity the per-row × uses), so the same
|
||
race-safety holds: an id either matches the row the operator selected or
|
||
matches nothing, never a neighbour that another session appended in the
|
||
meantime. An empty selection is a no-op, not an error.
|
||
"""
|
||
board = resolve_booth(name)
|
||
for entry_id in sel:
|
||
remove_link_entry(board, entry_id)
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)
|
||
|
||
@app.post("/b/{name}/pin")
|
||
def board_pin(name: str, entry: str = Form(...)):
|
||
"""Toggle a row's pinned (favorite) state, by content id. Pinned rows
|
||
float to the top of the board; toggling again unpins. Reversible, so no
|
||
confirmation — unlike removal."""
|
||
toggle_pin(resolve_booth(name), entry)
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)
|
||
|
||
def _safe_next(nxt: str) -> str:
|
||
"""Where to land after keep/unkeep. Defaults to the index; a booth page
|
||
can ask to stay put. Only same-site absolute paths are honoured — `//`
|
||
and any scheme are refused, because a redirect target taken from a form
|
||
field is an open redirect if you do not check it."""
|
||
if nxt.startswith("/") and not nxt.startswith("//") and "\\" not in nxt:
|
||
return nxt
|
||
return "/"
|
||
|
||
@app.post("/b/{name}/keep")
|
||
def booth_keep(name: str, next: str = Form("/")):
|
||
(resolve_booth(name) / KEEP_MARKER).touch()
|
||
return RedirectResponse(url=_safe_next(next), status_code=303)
|
||
|
||
@app.post("/b/{name}/unkeep")
|
||
def booth_unkeep(name: str, next: str = Form("/")):
|
||
# missing_ok: releasing an already-released board is a no-op, not a 500.
|
||
(resolve_booth(name) / KEEP_MARKER).unlink(missing_ok=True)
|
||
return RedirectResponse(url=_safe_next(next), status_code=303)
|
||
|
||
@app.post("/b/{name}/blur")
|
||
def booth_blur(name: str, f: str = Form(...), on: str = Form("1")):
|
||
"""Toggle one item's blur. Reversible and cosmetic, so no confirmation.
|
||
See BLUR_FILE: this hides an item from a glance, it does not protect it."""
|
||
booth = resolve_booth(name)
|
||
# Guard the path the same way the file route must: a blur entry is only
|
||
# ever a booth-relative path, never an escape.
|
||
rel = f.strip().lstrip("/")
|
||
if ".." in Path(rel).parts:
|
||
raise HTTPException(status_code=400, detail="bad item path")
|
||
set_blurred(booth, rel, on not in ("0", "false", ""))
|
||
return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)
|
||
|
||
@app.post("/b/{name}/delete")
|
||
def booth_delete_form(name: str):
|
||
shutil.rmtree(resolve_booth(name))
|
||
return RedirectResponse(url="/", status_code=303)
|
||
|
||
@app.delete("/b/{name}")
|
||
def booth_delete_api(name: str):
|
||
shutil.rmtree(resolve_booth(name))
|
||
return JSONResponse({"wiped": name})
|
||
|
||
return app
|
||
|
||
|
||
def _from_env() -> FastAPI:
|
||
data = os.environ.get("BOOTH_DATA_DIR", str(Path.home() / "booth-data"))
|
||
ttl = float(os.environ.get("BOOTH_TTL_HOURS", "24"))
|
||
host = os.environ.get("BOOTH_HOST_LABEL", "")
|
||
interval = int(float(os.environ.get("BOOTH_SWEEP_INTERVAL_MIN", "15")) * 60)
|
||
max_mb = float(os.environ.get("BOOTH_MAX_UPLOAD_MB", "1024"))
|
||
max_n = int(os.environ.get("BOOTH_MAX_FILES", "50"))
|
||
return create_app(
|
||
data,
|
||
ttl_hours=ttl,
|
||
host_label=host,
|
||
sweep_interval_s=interval,
|
||
max_upload_mb=max_mb,
|
||
max_files=max_n,
|
||
)
|
||
|
||
|
||
app = _from_env()
|