contact messages: data model, strict renderer, socket protocol core
Data.contact holds visitor threads keyed by blake3(session token). Message bodies are a Markdown subset rendered with a strict MarkdownIt (html=False): the stored html is always markdown-it output, never sanitized user markup. pagerite/messages.py owns the store ops, wire payloads and both socket protocols (visitor frames on /_ws, admin frames on /_api/ws/analytics); the socket wiring follows.
This commit is contained in:
LeoVasanko committed
1 parent
3c9ef04ffd
commit
1d827da62a
2 files changed
+332
No files matched your search
@@ -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:
|
||||
|
||||
@@ -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))
|
||||
Reference in new issue
Block a user