challenge_status_tracker/tracker/hedgedoc.py

"""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 /<note>          -> 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/<noteId>, 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