#!/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()