"""Writing the note into HedgeDoc 1.x. HedgeDoc 1.x has no REST endpoint that edits a note's body, so this pretends to be the web editor. Nothing in here knows about challenges. """ from __future__ import annotations import json import random import sys import time from urllib.parse import urlsplit import requests # engine.io v3 packet types EIO_OPEN, EIO_CLOSE, EIO_PING, EIO_PONG, EIO_MSG = "0", "1", "2", "3", "4" # socket.io packet types, as the second character of an EIO_MSG packet SIO_CONNECT, SIO_DISCONNECT, SIO_EVENT, SIO_ERROR = "0", "1", "2", "4" def u16len(s: str) -> int: """Length in UTF-16 code units — what JavaScript's `String.length` counts, and therefore what engine.io payload framing and OT operations count. Matters as soon as the markdown holds an emoji (🔴 is one Python char but two units).""" return len(s.encode("utf-16-le")) // 2 def encode_payload(packets: tuple) -> bytes: return "".join(f"{u16len(p)}:{p}" for p in packets).encode("utf-8") def decode_payload(text: str) -> list[str]: buf = text.encode("utf-16-le") out, i = [], 0 while i < len(buf): j = i while buf[j:j + 2] != b":\x00": j += 2 if j >= len(buf): raise RuntimeError("malformed engine.io payload") n = int(buf[i:j].decode("utf-16-le")) j += 2 out.append(buf[j:j + 2 * n].decode("utf-16-le")) i = j + 2 * n return out class HedgeDocNote: """Overwrite an existing HedgeDoc 1.x note in place. Speaks the same engine.io/socket.io protocol the web editor uses, over XHR long-polling (no extra dependencies beyond `requests`): GET / -> session cookie; the socket rejects us without one GET /socket.io/ -> engine.io handshake, yields a session id poll for 'doc' -> the note's current text and OT revision emit 'operation' -> [-len(old), new]: delete everything, insert ours poll for 'ack' -> the server applied it and told the other editors The note's permission must allow us to write ("freely" for an anonymous run). """ def __init__(self, url: str, timeout: int = 20): parts = urlsplit(url) path = parts.path.rstrip("/") if not parts.scheme or not parts.netloc or "/" not in path: raise ValueError(f"expected a note URL like https://host/, got {url!r}") self.note = path.rsplit("/", 1)[-1] prefix = path[: -len(self.note)].rstrip("/") self.base = f"{parts.scheme}://{parts.netloc}{prefix}" self.timeout = timeout self.sid: str | None = None self.s = requests.Session() self.s.headers["Origin"] = f"{parts.scheme}://{parts.netloc}" # -- transport --------------------------------------------------------- def _params(self) -> dict: p = {"EIO": "3", "transport": "polling", "noteId": self.note, "t": format(random.getrandbits(40), "x")} if self.sid: p["sid"] = self.sid return p def _open(self) -> None: self.sid = None self.s.get(f"{self.base}/{self.note}", timeout=self.timeout).raise_for_status() r = self.s.get(f"{self.base}/socket.io/", params=self._params(), timeout=self.timeout) r.raise_for_status() pkts = decode_payload(r.text) if not pkts or pkts[0][:1] != EIO_OPEN: raise RuntimeError(f"unexpected engine.io handshake: {pkts[:1]}") self.sid = json.loads(pkts[0][1:])["sid"] def _send(self, *packets: str) -> None: r = self.s.post(f"{self.base}/socket.io/", params=self._params(), data=encode_payload(packets), headers={"Content-Type": "text/plain;charset=UTF-8"}, timeout=self.timeout) r.raise_for_status() def _events(self): """Poll once, answer keepalives, yield decoded socket.io events.""" r = self.s.get(f"{self.base}/socket.io/", params=self._params(), timeout=self.timeout) r.raise_for_status() for p in decode_payload(r.text): if p == EIO_PING: self._send(EIO_PONG) elif p == EIO_CLOSE: raise RuntimeError("hedgedoc closed the connection") elif p[:1] == EIO_MSG: kind, data = p[1:2], p[2:] if kind == SIO_ERROR: raise RuntimeError(f"hedgedoc refused the socket: {data}") if kind == SIO_EVENT: yield json.loads(data) def _wait(self, name: str, polls: int = 8): for _ in range(polls): for ev in self._events(): if ev and ev[0] == name: return ev[1] if len(ev) > 1 else None raise RuntimeError(f"timed out waiting for {name!r} from hedgedoc") def _close(self) -> None: try: self._send(EIO_MSG + SIO_DISCONNECT, EIO_CLOSE) except requests.RequestException: pass # we already got our ack; a tidy goodbye is a nicety # -- api --------------------------------------------------------------- def download(self) -> str: r = self.s.get(f"{self.base}/{self.note}/download", timeout=self.timeout) r.raise_for_status() return r.content.decode("utf-8", "replace") def replace(self, markdown: str) -> bool: """Replace the note body. False if it already said exactly this.""" self._open() doc = self._wait("doc") old, revision = doc["str"], doc["revision"] if old == markdown: self._close() return False # an OT operation is a list of: n>0 keep n chars, n<0 drop n, str insert ops = ([-u16len(old)] if old else []) + ([markdown] if markdown else []) self._send(EIO_MSG + SIO_EVENT + json.dumps(["operation", revision, ops, None], ensure_ascii=False)) self._wait("ack") time.sleep(2) # give the realtime server a moment to flush the note to its DB self._close() return True def update_hedgedoc(url: str, md: str, retries: int = 3, quiet: bool = False) -> bool: """Write `md` into the existing note at `url`, then read it back to be sure.""" for attempt in range(1, retries + 1): try: note = HedgeDocNote(url) changed = note.replace(md) if note.download() != md: raise RuntimeError("note does not match what we wrote") return changed except (requests.RequestException, RuntimeError, ValueError, KeyError) as e: if attempt == retries: raise if not quiet: print(f"hedgedoc update failed ({e}) — retry {attempt + 1}/{retries}", file=sys.stderr) time.sleep(3 * attempt) return False def push_hedgedoc(base: str, md: str) -> str: """POST `md` as a *new* note; returns its URL.""" r = requests.post(base.rstrip("/") + "/new", data=md.encode(), headers={"Content-Type": "text/markdown"}, allow_redirects=False) r.raise_for_status() loc = r.headers.get("Location", "") return loc if loc.startswith("http") else base.rstrip("/") + loc