diff --git a/pagerite/data.py b/pagerite/data.py index cd6c017..fc06ad5 100644 --- a/pagerite/data.py +++ b/pagerite/data.py @@ -50,6 +50,41 @@ class LangEdits(msgspec.Struct, omit_defaults=True): adds: dict[str, str] = {} +class ContactMessage(msgspec.Struct, omit_defaults=True): + """One message in a visitor contact thread (docs/messages.md). + + ``md`` is the Markdown-subset source serialized by the composer; + ``html`` its rendering by the strict parser in pagerite.messages + (html=False — no user markup ever reaches a page, only markdown-it + output from the constrained grammar). Both are stored: the source for + future edits/moderation, the rendering so panels inject it verbatim. + """ + + t: datetime + #: Markdown-subset source. + md: str = "" + #: Strict server-side rendering of ``md``. + html: str = "" + #: Author: "" = the visitor, otherwise the admin's SSO remote-user. + admin: str = "" + #: Page path the message was sent from ("" for admin replies). + page: str = "" + + +class ContactThread(msgspec.Struct, omit_defaults=True): + """One visitor's message thread, keyed in Data.contact by token hash. + + Messages are keyed (never a list) by a sortable timestamp-random id so + a save's database diff touches only the appended message. + """ + + created: datetime + messages: dict[str, ContactMessage] = {} + #: When an admin last looked at the thread; visitor messages newer + #: than this are unread in the admin inbox. + admin_read: datetime | None = None + + class Node(msgspec.Struct, omit_defaults=True): """One item of the site hierarchy. @@ -167,6 +202,9 @@ class Data(msgspec.Struct): #: path -> lang -> LangEdits (paths without leading slash). Replaces #: the old "patches" key (ignored on decode, discarding that data). overrides: dict[str, dict[str, LangEdits]] = {} + #: Visitor contact threads (docs/messages.md), keyed by the blake3 + #: hash of the visitor's session token (never the token itself). + contact: dict[bytes, ContactThread] = {} def node_markdown(data: Data, node: Node) -> str | None: diff --git a/pagerite/messages.py b/pagerite/messages.py new file mode 100644 index 0000000..2860243 --- /dev/null +++ b/pagerite/messages.py @@ -0,0 +1,294 @@ +"""Visitor contact messages: storage, strict rendering, socket protocol. + +Anonymous visitors open a panel from a ``{contact: Label}`` button in an +article (views._contact_tag). The panel rides the public /_ws activity +socket: tracking.activity_ws routes frames carrying a ``token`` field to +``visitor_frame``. The token (a localStorage random, generated on first +use) identifies the visitor's thread; only its blake3 hash is stored. +Admins watch and reply over the analytics socket +(tracking.analytics_websocket), admin-gated like the rest of /_api; their +frames go to ``admin_frame``. Both sides get pushed full thread payloads +(threads are small); there is no REST polling anywhere. + +Message bodies are a Markdown subset (emphasis, links, lists, code plus +/_f/ image and file references) serialized client-side from a constrained +contenteditable composer. They are rendered here with a strict MarkdownIt +instance: html=False, so not a single byte of user markup can reach a +page — the stored html is always markdown-it output from the constrained +grammar, never sanitized user HTML (sanitization is unwinnable; grammar +restriction is not). Links get rel="nofollow ugc noopener"; images are +limited to /_f/ store names (the upload endpoint is the only way in). +""" + +import logging +import re +import secrets +from datetime import UTC, datetime + +import blake3 +from fastapi import WebSocket +from markdown_it import MarkdownIt +from markdown_it.common.utils import escapeHtml + +from pagerite.data import ContactMessage, ContactThread +from pagerite.state import data, kanta + +logger = logging.getLogger(__name__) + +#: Session-token shape: crypto-random from the browser (uuid or hex). +TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$") + +#: Message source length cap (the rendered html is bounded by it too). +MAX_MESSAGE = 8000 + +#: Live sockets: visitor token hash -> panels, and every connected admin. +_visitor_ws: dict[bytes, set[WebSocket]] = {} +_admin_ws: set[WebSocket] = set() + + +# --- Strict rendering ----------------------------------------------------- +# A separate instance from markdown.py's page renderer: no raw HTML (the +# trusted-author passthrough would be an XSS hole here), no plugins, no +# typographer. markdown-it escapes all markup in the source and validates +# link protocols (javascript: etc. never become hrefs). + +_md = MarkdownIt("default", {"html": False, "typographer": False, "breaks": True}) + +#: Images may only reference the content-addressed store (uploads land +#: there via /_contact/files); anything else collapses to its alt text. +_IMG_SRC = re.compile(r"/_f/[0-9a-f]{12}(?:\.[a-z0-9]+)?") + + +def _link_rule(self, tokens, idx, options, env): + """Open external links in a new tab without window.opener; store-file + links (uploaded attachments) render as download chips.""" + token = tokens[idx] + href = token.attrs.get("href", "") + if href.startswith("/_f/"): + token.attrSet("class", "attachment") + token.attrSet("download", "") + else: + token.attrSet("rel", "nofollow ugc noopener") + token.attrSet("target", "_blank") + return self.renderToken(tokens, idx, options, env) + + +def _image_rule(self, tokens, idx, options, env): + """Drop images that do not point at a /_f/ store file (escaped alt).""" + token = tokens[idx] + if not _IMG_SRC.fullmatch(token.attrs.get("src", "")): + return escapeHtml(self.renderInlineAsText(token.children, options, env)) + token.attrSet("loading", "lazy") + return self.renderToken(tokens, idx, options, env) + + +_md.add_render_rule("link_open", _link_rule) +_md.add_render_rule("image", _image_rule) + + +def render_message(text: str) -> str: + """Render message-source Markdown to thread HTML (the only HTML a + visitor's message ever becomes).""" + return _md.render(text) + + +# --- Thread store ---------------------------------------------------------- + + +def thread_key(token: str) -> bytes: + """The Data.contact key for a session token (hash, never the token).""" + return blake3.blake3(token.encode()).digest(12) + + +def _new_id(now: datetime) -> str: + """Sortable message id: timestamp plus a random tiebreak.""" + return f"{now:%Y%m%d%H%M%S%f}-{secrets.token_hex(3)}" + + +def _unread(thread: ContactThread) -> bool: + """Admin perspective: the last message is the visitor's and newer than + the admin's last read.""" + for mid in reversed(thread.messages): + m = thread.messages[mid] + if m.admin: + return False + return thread.admin_read is None or m.t > thread.admin_read + return False + + +def _payload(key: bytes, thread: ContactThread) -> dict: + """Full thread state for the wire (same shape for visitors and admins; + clients replace/upsert by id). Messages are capped to the last 200.""" + return { + "type": "thread", + "id": key.hex(), + "created": thread.created.isoformat(), + "unread": _unread(thread), + "messages": [ + { + "id": mid, + "t": m.t.isoformat(), + "admin": m.admin, + "html": m.html, + "page": m.page, + } + for mid, m in list(thread.messages.items())[-200:] + ], + } + + +def _empty_payload(key: bytes) -> dict: + return {"type": "thread", "id": key.hex(), "created": "", "unread": False, + "messages": []} + + +async def _push(sockets, payload: dict) -> None: + """Send a payload to a socket set, dropping the ones that fail.""" + closed = set() + for ws in sockets: + try: + await ws.send_json(payload) + except Exception: + logger.exception("Message push failed; dropping client") + closed.add(ws) + for ws in closed: + sockets.discard(ws) + + +async def _broadcast(key: bytes, thread: ContactThread) -> None: + """Push a thread update to its visitor's panels and every admin.""" + payload = _payload(key, thread) + if key in _visitor_ws: + await _push(_visitor_ws[key], payload) + await _push(_admin_ws, payload) + + +def _append(key: bytes, msg: ContactMessage, action: str, user: str | None) -> None: + """Append a message to the thread (created on first use), in one + transaction so the database diff is just the new entry.""" + with kanta.transaction(action, user=user, extra=key.hex()): + thread = data.contact.get(key) + if thread is None: + thread = data.contact[key] = ContactThread(created=msg.t) + thread.messages[_new_id(msg.t)] = msg + + +# --- Visitor protocol (/_ws frames carrying "token") ------------------------ + + +async def visitor_frame(ws: WebSocket, raw: dict) -> None: + """Handle one visitor frame: ``{token}`` registers/resyncs (the thread + state is pushed back), ``{token, msg, fr?}`` appends a message and + broadcasts the updated thread to both sides.""" + token = str(raw.get("token") or "") + if not TOKEN_RE.fullmatch(token): + await ws.send_json({"type": "contact-error", "detail": "bad session token"}) + return + key = thread_key(token) + _visitor_ws.setdefault(key, set()).add(ws) + text = str(raw.get("msg") or "").strip() + if not text: + thread = data.contact.get(key) + await ws.send_json(_payload(key, thread) if thread else _empty_payload(key)) + return + if len(text) > MAX_MESSAGE: + await ws.send_json({"type": "contact-error", "detail": "message too long"}) + return + now = datetime.now(UTC) + _append( + key, + ContactMessage( + t=now, + md=text, + html=render_message(text), + page=str(raw.get("fr") or "")[:200], + ), + "contact", + None, + ) + await _broadcast(key, data.contact[key]) + + +def visitor_disconnect(ws: WebSocket) -> None: + """Drop a closed /_ws socket from every thread registry.""" + empty = [key for key, sockets in _visitor_ws.items() if not sockets - {ws}] + for sockets in _visitor_ws.values(): + sockets.discard(ws) + for key in empty: + del _visitor_ws[key] + + +# --- Admin protocol (/_api/ws/analytics frames) ----------------------------- + + +def _key_from_hex(value) -> bytes | None: + try: + key = bytes.fromhex(str(value or "")) + except ValueError: + return None + return key if len(key) == 12 else None + + +def _last_t(thread: ContactThread) -> datetime: + """The thread's last-activity time (its newest message, else creation).""" + if thread.messages: + return list(thread.messages.values())[-1].t + return thread.created + + +async def admin_connect(ws: WebSocket) -> None: + """Register an admin socket and send the inbox: all threads, most + recently active first (capped at 100).""" + _admin_ws.add(ws) + threads = sorted(data.contact.items(), key=lambda kv: _last_t(kv[1]), reverse=True)[ + :100 + ] + await ws.send_json( + {"type": "threads", "threads": [_payload(k, t) for k, t in threads]} + ) + + +def admin_disconnect(ws: WebSocket) -> None: + _admin_ws.discard(ws) + + +async def admin_frame(ws: WebSocket, raw: dict) -> None: + """Handle one admin frame: ``{type: reply, id, text}`` appends an admin + message, ``{type: read, id}`` marks a thread read.""" + match raw.get("type"): + case "reply": + key = _key_from_hex(raw.get("id")) + text = str(raw.get("text") or "").strip() + thread = data.contact.get(key) if key else None + if thread is None: + await ws.send_json({"type": "contact-error", "detail": "no such thread"}) + return + if not text or len(text) > MAX_MESSAGE: + await ws.send_json({"type": "contact-error", "detail": "bad message"}) + return + now = datetime.now(UTC) + user = ws.headers.get("remote-user") + _append( + key, + ContactMessage( + t=now, + md=text, + html=render_message(text), + admin=user or "admin", + ), + "contact:reply", + user, + ) + with kanta.transaction("contact:read", user=user, extra=key.hex()): + thread.admin_read = now + await _broadcast(key, thread) + case "read": + key = _key_from_hex(raw.get("id")) + thread = data.contact.get(key) if key else None + if thread is None: + return + with kanta.transaction( + "contact:read", user=ws.headers.get("remote-user"), extra=key.hex() + ): + thread.admin_read = datetime.now(UTC) + await _push(_admin_ws, _payload(key, thread))