commit 7c546ab673f4ba2c6aaff5c9a8f5d3acc7d818fd Author: cato447 Date: Sun Aug 30 14:25:56 2026 +0200 added nextloud tracker diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..276e14a --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +ctf_status.env +*.md.out diff --git a/README.md b/README.md new file mode 100644 index 0000000..5f4cd79 --- /dev/null +++ b/README.md @@ -0,0 +1,110 @@ +# ctf_status.py + +Reads the per-author challenge folders straight off the Nextcloud share (WebDAV, +no sync, no mount) and tells you who to scream at. + +## Setup + +```sh +cp ctf_status.env.example ctf_status.env && chmod 600 ctf_status.env +$EDITOR ctf_status.env # app password from Nextcloud → Settings → Security +``` + +## Use + +```sh +./ctf_status.py # terminal summary +./ctf_status.py -o overview.md # + HedgeDoc markdown to a file +./ctf_status.py --stdout | wl-copy # straight to clipboard, paste into HedgeDoc +./ctf_status.py --update # overwrite the note in HEDGEDOC_NOTE +./ctf_status.py --update # ... or an explicit one +``` + +## Keeping a HedgeDoc note up to date + +`--update` overwrites an **existing** note in place, so the URL you share stays the +same. Every run rewrites the whole body, starting with + +``` +--- +tags: Logistics, h4tum CTF 2026 +--- +``` + +and a `**Last updated:** ` line under the heading, so a stale note +is obvious at a glance. Tags come from `HEDGEDOC_TAGS` (or `--tags`). + +The note's permission has to be **freely** (Menu → Permission), because the run is +anonymous. HedgeDoc 1.x has no REST API for editing a note, so the script talks the +editor's own engine.io/socket.io + OT protocol over long-polling — it reads the +current text, sends one "delete everything, insert this" operation, waits for the +server's `ack`, and reads the note back to confirm. Anyone with the note open sees +the change live. Still `requests`-only, no new dependencies. + +`--hedgedoc ` (POST a *new* note) is unrelated and only works where +anonymous note creation is enabled — on `hedgedoc.cit.tum.de` it is not. + +## Every 30 minutes, on a server + +Copy the repo to the server, create `ctf_status.env` there (it is gitignored, so it +does not travel with the code), then: + +```sh +./install-timer.sh +``` + +It writes systemd **user** units into `~/.config/systemd/user/`, substituting +whatever directory the repo actually sits in, enables the timer, turns on lingering +so it keeps firing when you are not logged in, and does one run to prove it works. +Re-run it any time; it just overwrites the units. + +```sh +systemctl --user list-timers ctf-status.timer # when it next fires +journalctl --user -u ctf-status.service -n 50 # what it did +systemctl --user start ctf-status.service # run it now +systemctl --user disable --now ctf-status.timer # stop it +``` + +The timer fires at `:00` and `:30` with up to a minute of jitter, and +`Persistent=true` catches up a run missed while the machine was down. Plain cron +works too, if you would rather: + +``` +*/30 * * * * /path/to/planner/ctf_status.py --update --quiet +``` + +`--quiet` keeps a successful run silent, so you only hear about it when the note +changed or something broke. Overlapping runs are prevented by a lock file, and the +HedgeDoc write is retried (`--retries`, default 3) before the run gives up with a +one-line error and a non-zero exit. + +## What it flags + +| verdict | meaning | +| --- | --- | +| 🔴 SCREAM | nothing submitted, **or** they filled out `template.md` in place | +| 🟡 NAG | submitted, but placeholders / empty sections remain | +| 🟢 OK | every challenge fully filled out | + +A `template.md` that is still pristine is ignored. One that has real content in it +is counted as a challenge *and* flagged — so nobody's work is lost, they just get +yelled at. + +Also flagged per challenge: unnamed challenge, `[placeholder]` values left in any +of Category/Difficulty/Event/Author/Tester/Status, empty prose sections, and +category/difficulty values outside the allowed set. + +## Layout it expects + +``` +/ + alice/ + README.md # ignored + template.md # ignored while pristine + baby-rsa.md # ← a challenge, any filename + bob/ +``` + +Any `.md` that isn't `README.md` is parsed as a challenge, so the file naming +convention doesn't have to be enforced by the script. If you want it enforced, +add the check in `Challenge.problems`. diff --git a/ctf-status.service b/ctf-status.service new file mode 100644 index 0000000..93d5eb4 --- /dev/null +++ b/ctf-status.service @@ -0,0 +1,12 @@ +[Unit] +Description=Push the CTF challenge status to HedgeDoc +Documentation=file://__DIR__/README.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +WorkingDirectory=__DIR__ +ExecStart=__DIR__/ctf_status.py --update --quiet +# Nextcloud walk + HedgeDoc write; never let one run squat on the next timer slot +TimeoutStartSec=5min diff --git a/ctf-status.timer b/ctf-status.timer new file mode 100644 index 0000000..a05bf50 --- /dev/null +++ b/ctf-status.timer @@ -0,0 +1,13 @@ +[Unit] +Description=Refresh the CTF challenge status every 30 minutes + +[Timer] +OnCalendar=*:0/30 +# don't hammer Nextcloud and HedgeDoc on the exact half hour +RandomizedDelaySec=60 +AccuracySec=30s +# catch up on the last run if the machine was down when it was due +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/ctf_status.env.example b/ctf_status.env.example new file mode 100644 index 0000000..e9b92ba --- /dev/null +++ b/ctf_status.env.example @@ -0,0 +1,14 @@ +# cp to ctf_status.env and chmod 600 +NC_URL=https://cloud.h4tum.de +NC_USER=your-nextcloud-login +# Settings -> Security -> "Create new app password" (NOT your real password) +NC_PASS=xxxxx-xxxxx-xxxxx-xxxxx-xxxxx +# path of the folder holding the per-author folders, relative to your Nextcloud root +NC_ROOT=CTF/challenges + +# --- HedgeDoc: the existing note that gets overwritten every run ------------ +# full URL of the note (its permission must be "freely" so an anonymous +# run may write to it), e.g. https://hedgedoc.cit.tum.de/AbCdEf123 +HEDGEDOC_NOTE= +# YAML front-matter tags written at the top of the note +HEDGEDOC_TAGS=Logistics, h4tum CTF 2026 diff --git a/ctf_status.py b/ctf_status.py new file mode 100755 index 0000000..a001cec --- /dev/null +++ b/ctf_status.py @@ -0,0 +1,618 @@ +#!/usr/bin/env python3 +"""CTF challenge tracker. + +Reads the per-author challenge folders from a Nextcloud share over WebDAV, +parses the filled-out challenge templates and reports: + * a terminal "who do I have to scream at" list + * a HedgeDoc-ready markdown overview + +Config: see ctf_status.env.example -> ctf_status.env (chmod 600) +""" +from __future__ import annotations + +import argparse +import fcntl +import json +import os +import random +import re +import sys +import tempfile +import time +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from urllib.parse import quote, unquote, urljoin, urlsplit +from xml.etree import ElementTree as ET + +import requests + +# ---------------------------------------------------------------- config ---- + +CFG_KEYS = ("NC_URL", "NC_USER", "NC_PASS", "NC_ROOT") +OPT_KEYS = ("HEDGEDOC_NOTE", "HEDGEDOC_TAGS", "TITLE") +DEFAULT_TAGS = "Logistics, h4tum CTF 2026" +DEFAULT_TITLE = "CTF challenge tracking" + + +def load_env() -> dict: + cfg = {} + for candidate in ( + Path(__file__).with_name("ctf_status.env"), + Path.home() / ".config" / "ctf-status" / "env", + ): + if candidate.exists(): + for line in candidate.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + cfg[k.strip()] = v.strip().strip("'\"") + break + for k in CFG_KEYS + OPT_KEYS: + if os.environ.get(k): + cfg[k] = os.environ[k] + missing = [k for k in CFG_KEYS if not cfg.get(k)] + if missing: + sys.exit(f"missing config: {', '.join(missing)} (see ctf_status.env.example)") + return cfg + + +# --------------------------------------------------------------- webdav ----- + +DAV_NS = "{DAV:}" + + +class Dav: + def __init__(self, base_url: str, user: str, password: str): + self.base = base_url.rstrip("/") + f"/remote.php/dav/files/{quote(user)}/" + self.s = requests.Session() + self.s.auth = (user, password) + + def _url(self, path: str) -> str: + return urljoin(self.base, quote(path.strip("/")) + "/") + + def walk(self, root: str) -> list[str]: + """Return all file paths (relative to root) under root.""" + body = ( + '' + "" + ) + out: list[str] = [] + for depth in ("infinity", "1"): + try: + out = self._propfind(root, body, depth) + if out: + return out + except requests.HTTPError as e: + if depth == "infinity" and e.response is not None and e.response.status_code in (403, 405, 507): + continue # server forbids Depth: infinity -> recurse manually + raise + return out + + def _propfind(self, root: str, body: str, depth: str) -> list[str]: + r = self.s.request( + "PROPFIND", self._url(root), + headers={"Depth": depth, "Content-Type": "application/xml"}, data=body, + ) + r.raise_for_status() + tree = ET.fromstring(r.content) + base_path = unquote(self._url(root).split(self.base.split("/remote.php")[0], 1)[-1]) + files, dirs = [], [] + for resp in tree.findall(f"{DAV_NS}response"): + href = unquote(resp.findtext(f"{DAV_NS}href") or "") + if not href or href.rstrip("/") == base_path.rstrip("/"): + continue + rel = href.split(base_path, 1)[-1].strip("/") if base_path in href else href + is_dir = resp.find(f".//{DAV_NS}collection") is not None + (dirs if is_dir else files).append(rel) + if depth == "1": + for d in dirs: + files += [f"{d}/{f}" for f in self._propfind(f"{root}/{d}", body, "1")] + return files + + def get(self, path: str) -> str: + r = self.s.get(self._url(path).rstrip("/")) + r.raise_for_status() + return r.content.decode("utf-8", "replace") + + +# --------------------------------------------------------------- parsing ---- + +FIELDS = ("Category", "Difficulty", "Event", "Author", "Tester", "Status") +HEADING_RE = re.compile(r"^\s{0,3}#{2,4}\s+(.+?)\s*$", re.M) +FIELD_RE = re.compile(r"^\s*[-*]\s*\**(?P[A-Za-z ]+?)\**\s*:\s*(?P.*?)\s*$", re.M) +SECTION_RE = re.compile(r"^\s{0,3}#{3,5}\s*(?P.+?)\s*$\n(?P.*?)(?=^\s{0,3}#{2,5}\s|\Z)", + re.M | re.S) +# a value the author never touched: "[crypto, pwn, ...]" or "[Author Name]" +PLACEHOLDER_RE = re.compile(r"^\\?\[.*\\?\]$") +PROSE_PROMPTS = ( + "what should the player take home", "please answer in 1-2", + "how do you plan on implementing", "needed for infrastructure", + "how will the players exploit", +) +STATUS_ORDER = ["Planning", "PoC Draft", "Testing Ready", "Tested", "Ready"] +CATEGORIES = ["crypto", "pwn", "web", "rev", "misc", "osint", "forensic", "hardware"] +DIFFICULTIES = ["intro", "easy", "medium", "hard"] + + +def clean(v: str) -> str: + return v.replace("\\[", "[").replace("\\]", "]").strip() + + +def is_placeholder(v: str) -> bool: + v = clean(v) + return not v or bool(PLACEHOLDER_RE.match(v)) + + +@dataclass +class Challenge: + author_dir: str + path: str + name: str = "" + fields: dict = field(default_factory=dict) + sections: dict = field(default_factory=dict) + in_template: bool = False + + # -- derived ----------------------------------------------------------- + @property + def missing_fields(self) -> list[str]: + return [f for f in FIELDS if is_placeholder(self.fields.get(f, ""))] + + @property + def missing_sections(self) -> list[str]: + return [k for k, v in self.sections.items() if not v] + + @property + def named(self) -> bool: + return bool(self.name) and not is_placeholder(self.name) + + @property + def touched(self) -> bool: + """Did anyone put *any* real content in here?""" + return self.named or len(self.missing_fields) < len(FIELDS) or bool( + [k for k, v in self.sections.items() if v]) + + @property + def complete(self) -> bool: + return self.named and not self.missing_fields and not self.missing_sections + + def get(self, key: str, dash: str = "—") -> str: + v = clean(self.fields.get(key, "")) + return dash if is_placeholder(v) else v + + @property + def status(self) -> str: + return self.get("Status", "—") + + @property + def problems(self) -> list[str]: + p = [] + if self.in_template and self.touched: + p.append("filled out template.md in place (rename it!)") + if not self.named: + p.append("no challenge name") + if self.missing_fields: + p.append("unfilled: " + ", ".join(self.missing_fields)) + if self.missing_sections: + p.append("empty section(s): " + ", ".join(self.missing_sections)) + cat = self.get("Category", "").lower() + if cat and cat != "—" and cat not in CATEGORIES and "?" not in cat: + p.append(f"odd category {cat!r}") + diff = self.get("Difficulty", "").lower() + if diff and diff != "—" and diff not in DIFFICULTIES: + p.append(f"odd difficulty {diff!r}") + return p + + +def parse_challenge(author: str, path: str, text: str, in_template: bool) -> Challenge: + c = Challenge(author_dir=author, path=path, in_template=in_template) + m = HEADING_RE.search(text) + if m: + c.name = clean(m.group(1)) + for fm in FIELD_RE.finditer(text): + key = fm.group("key").strip().title() + if key in FIELDS and key not in c.fields: + c.fields[key] = fm.group("val") + for sm in SECTION_RE.finditer(text): + name = clean(sm.group("name")) + if name == c.name or name in FIELDS: + continue + body = sm.group("body") + # strip the template's own prompt lines, keep what the author wrote + real = [ln.strip() for ln in body.splitlines() + if ln.strip() and not ln.strip().startswith(("-", "*", "#")) + and not any(p in ln.lower() for p in PROSE_PROMPTS)] + c.sections[name] = " ".join(real) + return c + + +# ---------------------------------------------------------------- report ---- + +BOLD, DIM, RED, YEL, GRN, CYA, RST = ( + "\033[1m", "\033[2m", "\033[31m", "\033[33m", "\033[32m", "\033[36m", "\033[0m") + + +@dataclass +class Author: + name: str + challenges: list = field(default_factory=list) + files_seen: list = field(default_factory=list) + + @property + def real(self) -> list: + return [c for c in self.challenges if c.touched] + + @property + def verdict(self) -> tuple[str, str]: + if not self.real: + return "SCREAM", "nothing submitted at all" + if any(c.in_template for c in self.real): + return "SCREAM", "worked inside template.md instead of copying it" + bad = [c for c in self.real if not c.complete] + if bad: + return "NAG", f"{len(bad)}/{len(self.real)} challenge(s) incomplete" + return "OK", f"{len(self.real)} challenge(s) done" + + +def collect(dav: Dav, root: str, no_color: bool) -> dict: + paths = [p for p in dav.walk(root) if p.lower().endswith(".md")] + authors: dict[str, Author] = {} + for p in sorted(paths): + parts = p.split("/") + if len(parts) < 2: + continue # loose file in the root, not an author's + adir, fname = parts[0], parts[-1] + a = authors.setdefault(adir, Author(adir)) + a.files_seen.append(fname) + if fname.lower() == "readme.md": + continue + text = dav.get(f"{root}/{p}") + a.challenges.append( + parse_challenge(adir, p, text, in_template=fname.lower() == "template.md")) + # authors with a folder but zero md files still deserve screaming at + for p in dav.walk(root): + top = p.split("/")[0] + if "/" in p: + authors.setdefault(top, Author(top)) + return authors + + +def print_terminal(authors: dict, c: bool) -> None: + def col(x): return x if c else "" + order = {"SCREAM": 0, "NAG": 1, "OK": 2} + rows = sorted(authors.values(), key=lambda a: (order[a.verdict[0]], a.name.lower())) + print(f"\n{col(BOLD)}=== authors ({len(rows)}) ==={col(RST)}") + for a in rows: + v, why = a.verdict + tint = {"SCREAM": RED, "NAG": YEL, "OK": GRN}[v] + print(f" {col(tint)}{v:<7}{col(RST)} {a.name:<24} {col(DIM)}{why}{col(RST)}") + for ch in a.real: + for prob in ch.problems: + print(f" {col(DIM)}↳ {Path(ch.path).name}: {prob}{col(RST)}") + + chs = [ch for a in authors.values() for ch in a.real] + done = sum(1 for ch in chs if ch.complete) + print(f"\n{col(BOLD)}=== {len(chs)} challenges, {done} fully filled out ==={col(RST)}") + by_cat = Counter(ch.get("Category", "?").lower() for ch in chs) + by_diff = Counter(ch.get("Difficulty", "?").lower() for ch in chs) + by_stat = Counter(ch.status for ch in chs) + for label, ctr, ref in (("category", by_cat, CATEGORIES), + ("difficulty", by_diff, DIFFICULTIES), + ("status", by_stat, STATUS_ORDER)): + keys = [k for k in ref if ctr.get(k.lower() if label != "status" else k)] + keys += [k for k in ctr if k not in keys and k not in [r.lower() for r in ref]] + parts = [f"{k}:{ctr[k]}" for k in keys if ctr.get(k)] + print(f" {col(CYA)}{label:<11}{col(RST)} " + " ".join(parts)) + missing = [k for k in CATEGORIES if not by_cat.get(k)] + if missing: + print(f" {col(YEL)}no coverage {col(RST)}" + ", ".join(missing)) + print() + + +def render_hedgedoc(authors: dict, title: str, tags: str = "") -> str: + chs = [ch for a in authors.values() for ch in a.real] + stamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") + L = [] + if tags: + # HedgeDoc reads this YAML block off the top of the note + L += ["---", f"tags: {tags}", "---", ""] + L += [f"# {title}", "", + f"**Last updated:** {stamp}", "", + f"*generated by `ctf_status.py` — {len(chs)} challenges from " + f"{len({ch.author_dir for ch in chs})} authors*", "", + "## Challenges", "", + "| Challenge | Category | Difficulty | Event | Author | Tester | Status |", + "| --- | --- | --- | --- | --- | --- | --- |"] + for ch in sorted(chs, key=lambda c: (c.get("Category", "zz").lower(), + DIFFICULTIES.index(c.get("Difficulty", "").lower()) + if c.get("Difficulty", "").lower() in DIFFICULTIES else 9, + c.name.lower())): + name = ch.name if ch.named else f"*(unnamed — {Path(ch.path).name})*" + flag = "" if ch.complete else " ⚠️" + L.append(f"| {name}{flag} | {ch.get('Category')} | {ch.get('Difficulty')} | " + f"{ch.get('Event')} | {ch.get('Author') if not is_placeholder(ch.fields.get('Author','')) else ch.author_dir} | " + f"{ch.get('Tester')} | {ch.status} |") + + cols = DIFFICULTIES + ["?"] + L += ["", "## Coverage (category × difficulty)", "", + "| Category | " + " | ".join(cols) + " | total |", + "| --- |" + " --- |" * (len(cols) + 1)] + grid = defaultdict(Counter) + for ch in chs: + grid[ch.get("Category", "?").lower()][ch.get("Difficulty", "?").lower()] += 1 + for cat in CATEGORIES + [k for k in grid if k not in CATEGORIES]: + row = grid.get(cat, Counter()) + cells = [str(row.get(d, 0) or "·") for d in cols] + L.append(f"| {cat} | " + " | ".join(cells) + f" | **{sum(row.values())}** |") + tot = Counter() + for row in grid.values(): + tot.update(row) + L.append("| **total** | " + " | ".join(f"**{tot.get(d, 0)}**" for d in cols) + + f" | **{sum(tot.values())}** |") + + L += ["", "## Author status", "", "| Author | State | Note |", "| --- | --- | --- |"] + icon = {"SCREAM": "🔴", "NAG": "🟡", "OK": "🟢"} + for a in sorted(authors.values(), key=lambda a: a.name.lower()): + v, why = a.verdict + L.append(f"| {a.name} | {icon[v]} | {why} |") + L += ["", "## Open issues", ""] + any_issue = False + for a in sorted(authors.values(), key=lambda a: a.name.lower()): + items = [(ch, p) for ch in a.real for p in ch.problems] + if not items and a.real: + continue + any_issue = True + L.append(f"- **{a.name}**") + if not a.real: + L.append(" - nothing submitted at all") + for ch, p in items: + L.append(f" - `{Path(ch.path).name}`: {p}") + if not any_issue: + L.append("*none — everyone is done. suspicious.*") + return "\n".join(L) + "\n" + + +# ------------------------------------------------------ hedgedoc (write) ---- + +# 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. + + HedgeDoc 1.x has no REST endpoint that edits a note's body, so this 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: + 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 + + +def single_run_lock(): + """Keep a slow or hung cron run from colliding with the next one.""" + fh = open(Path(tempfile.gettempdir()) / "ctf_status.lock", "w") + try: + fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + sys.exit("another ctf_status.py run is still going; skipping this one") + return fh # held until the process exits + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("-o", "--out", help="write the HedgeDoc markdown to this file") + ap.add_argument("--stdout", action="store_true", help="print the markdown instead of the summary") + ap.add_argument("--hedgedoc", metavar="URL", + help="POST the markdown to this HedgeDoc instance as a *new* note " + "(needs an instance that lets you create notes anonymously)") + ap.add_argument("--update", metavar="NOTE_URL", nargs="?", const="", + help="overwrite an existing HedgeDoc note in place; defaults to " + "HEDGEDOC_NOTE from the config. This is the one to cron.") + ap.add_argument("--retries", type=int, default=3, + help="attempts at the HedgeDoc write before giving up (default 3)") + ap.add_argument("--tags", help="YAML front-matter tags for the note " + f"(default {DEFAULT_TAGS!r}, or HEDGEDOC_TAGS)") + ap.add_argument("--title", help=f"note heading (default {DEFAULT_TITLE!r}, or TITLE)") + ap.add_argument("-q", "--quiet", action="store_true", + help="say nothing unless something changed or broke (for cron)") + ap.add_argument("--no-color", action="store_true") + args = ap.parse_args() + + cfg = load_env() + title = args.title or cfg.get("TITLE") or DEFAULT_TITLE + tags = args.tags if args.tags is not None else cfg.get("HEDGEDOC_TAGS", DEFAULT_TAGS) + + note_url = None + if args.update is not None: + note_url = args.update or cfg.get("HEDGEDOC_NOTE", "") + if not note_url: + sys.exit("--update needs a note URL, or HEDGEDOC_NOTE in the config") + try: + HedgeDocNote(note_url) # fail on a typo now, not after the Nextcloud walk + except ValueError as e: + sys.exit(str(e)) + lock = single_run_lock() # noqa: F841 (held for the lifetime of the process) + + dav = Dav(cfg["NC_URL"], cfg["NC_USER"], cfg["NC_PASS"]) + authors = collect(dav, cfg["NC_ROOT"].strip("/"), args.no_color) + if not authors: + sys.exit(f"no author folders found under {cfg['NC_ROOT']!r}") + + md = render_hedgedoc(authors, title, tags) + if args.out: + Path(args.out).write_text(md) + if args.stdout: + print(md, end="") + elif not args.quiet: + print_terminal(authors, c=not args.no_color and sys.stdout.isatty()) + if args.out: + print(f"markdown → {args.out}") + if args.hedgedoc: + print("hedgedoc → " + push_hedgedoc(args.hedgedoc, md)) + if note_url: + try: + changed = update_hedgedoc(note_url, md, retries=args.retries, quiet=args.quiet) + except (requests.RequestException, RuntimeError, ValueError, KeyError) as e: + sys.exit(f"hedgedoc update of {note_url} failed: {e}") + if changed or not args.quiet: + print(("updated → " if changed else "unchanged → ") + note_url) + + +if __name__ == "__main__": + main() diff --git a/install-timer.sh b/install-timer.sh new file mode 100755 index 0000000..daea77d --- /dev/null +++ b/install-timer.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env sh +# Install the 30-minute HedgeDoc refresh as systemd *user* units. +# Run it as the user that owns ctf_status.env, on the machine that should do the +# polling. Re-running it is safe; it just overwrites the units. +set -eu + +dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd -P) +units="${XDG_CONFIG_HOME:-$HOME/.config}/systemd/user" + +[ -x "$dir/ctf_status.py" ] || { echo "no executable ctf_status.py in $dir" >&2; exit 1; } +[ -f "$dir/ctf_status.env" ] || { + echo "no $dir/ctf_status.env — copy ctf_status.env.example to it first" >&2; exit 1; } +grep -q '^HEDGEDOC_NOTE=..*' "$dir/ctf_status.env" || { + echo "set HEDGEDOC_NOTE= in $dir/ctf_status.env first" >&2; exit 1; } +chmod 600 "$dir/ctf_status.env" + +mkdir -p "$units" +for u in ctf-status.service ctf-status.timer; do + sed "s|__DIR__|$dir|g" "$dir/$u" > "$units/$u" + echo "installed $units/$u" +done + +systemctl --user daemon-reload +systemctl --user enable --now ctf-status.timer + +# without lingering, user units are torn down as soon as you log out +if ! loginctl show-user "$(id -un)" -p Linger --value 2>/dev/null | grep -qx yes; then + loginctl enable-linger "$(id -un)" 2>/dev/null \ + || echo "could not enable linger; run: sudo loginctl enable-linger $(id -un)" >&2 +fi + +echo +echo "one run now, to prove it works:" +systemctl --user start ctf-status.service && echo " ok" +systemctl --user list-timers ctf-status.timer --no-pager