797 lines
33 KiB
Python
Executable File
797 lines
33 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""CTF challenge tracker.
|
||
|
||
Lists every repo in the GitLab challenges group, reads each repo's README.md
|
||
(the filled-out challenge template) and reports:
|
||
* a terminal summary: status per category, category x difficulty, open issues
|
||
* a HedgeDoc-ready markdown overview
|
||
|
||
Config: see ctf_status.env.example -> ctf_status.env (chmod 600)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from collections import Counter, defaultdict
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from urllib.parse import quote, urlsplit
|
||
|
||
import requests
|
||
|
||
# ---------------------------------------------------------------- config ----
|
||
|
||
CFG_KEYS = ("GITLAB_GROUP_URL", "GITLAB_TOKEN")
|
||
OPT_KEYS = ("HEDGEDOC_NOTE", "HEDGEDOC_TAGS", "TITLE",
|
||
"CHALLS_REPO_URL", "CHALLS_PATH", "CHALLS_BRANCH")
|
||
# the names a plain `.env` tends to use for the same thing
|
||
ALIASES = {"GL_PAT": "GITLAB_TOKEN", "GITLAB_URL": "GITLAB_GROUP_URL"}
|
||
DEFAULT_TAGS = "Logistics, h4tum CTF 2026"
|
||
DEFAULT_TITLE = "CTF challenge tracking"
|
||
DEFAULT_CHALLS_PATH = "2026/challs.txt"
|
||
|
||
|
||
def load_env() -> dict:
|
||
cfg = {}
|
||
here = Path(__file__).parent
|
||
# every file that exists contributes; the first one to set a key wins
|
||
for candidate in (here / "ctf_status.env", here / ".env",
|
||
Path.home() / ".config" / "ctf-status" / "env"):
|
||
if not candidate.exists():
|
||
continue
|
||
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)
|
||
k = k.strip().upper()
|
||
cfg.setdefault(ALIASES.get(k, k), v.strip().strip("'\""))
|
||
for k in CFG_KEYS + OPT_KEYS + tuple(ALIASES):
|
||
if os.environ.get(k):
|
||
cfg[ALIASES.get(k, 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
|
||
|
||
|
||
# --------------------------------------------------------------- gitlab -----
|
||
|
||
@dataclass
|
||
class Repo:
|
||
path: str
|
||
web_url: str
|
||
ssh_url: str
|
||
last_activity: str # YYYY-MM-DD
|
||
readme: str | None # None: the repo has no README at all
|
||
|
||
|
||
class GitLab:
|
||
"""Access to one group, plus the one file we write elsewhere. The token needs
|
||
Project: Read (to list the group) and Repository: Read (to download files);
|
||
--challs additionally needs read + write on the deployment repo."""
|
||
|
||
def __init__(self, group_url: str, token: str, timeout: int = 20):
|
||
parts = urlsplit(group_url)
|
||
self.api = f"{parts.scheme}://{parts.netloc}/api/v4"
|
||
self.group = parts.path.strip("/")
|
||
self.timeout = timeout
|
||
self.s = requests.Session()
|
||
self.s.headers["PRIVATE-TOKEN"] = token
|
||
|
||
def _get(self, url: str, **params) -> requests.Response:
|
||
return self._req("GET", url, params=params)
|
||
|
||
def _req(self, method: str, url: str, ok: tuple = (), **kw) -> requests.Response:
|
||
r = self.s.request(method, url, timeout=self.timeout, **kw)
|
||
if r.status_code in ok:
|
||
return r
|
||
if r.status_code == 403:
|
||
# fine-grained tokens say exactly which permission is missing
|
||
try:
|
||
why = r.json().get("error_description") or r.text
|
||
except ValueError:
|
||
why = r.text
|
||
raise RuntimeError(f"gitlab refused {r.url}: {why}")
|
||
r.raise_for_status()
|
||
return r
|
||
|
||
def projects(self) -> list[dict]:
|
||
out, page = [], "1"
|
||
while page:
|
||
r = self._get(f"{self.api}/groups/{quote(self.group, safe='')}/projects",
|
||
include_subgroups="true", per_page="100", page=page,
|
||
order_by="path", sort="asc")
|
||
out += r.json()
|
||
page = r.headers.get("X-Next-Page", "")
|
||
# repos GitLab is about to delete are renamed "...-deletion_scheduled-<id>"
|
||
return [p for p in out if not p.get("archived") and not p.get("marked_for_deletion_on")]
|
||
|
||
def readme(self, p: dict) -> str | None:
|
||
branch, url = p.get("default_branch"), p.get("readme_url")
|
||
if not branch or not url or f"/-/blob/{branch}/" not in url:
|
||
return None
|
||
path = url.split(f"/-/blob/{branch}/", 1)[1]
|
||
r = self._get(f"{self.api}/projects/{p['id']}/repository/files/"
|
||
f"{quote(path, safe='')}/raw", ref=branch)
|
||
return r.content.decode("utf-8", "replace")
|
||
|
||
def repos(self) -> list[Repo]:
|
||
projects = self.projects()
|
||
|
||
def fetch(p: dict) -> Repo:
|
||
return Repo(path=p["path"], web_url=p["web_url"],
|
||
ssh_url=p["ssh_url_to_repo"],
|
||
last_activity=(p.get("last_activity_at") or "")[:10],
|
||
readme=self.readme(p))
|
||
|
||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||
return list(pool.map(fetch, projects))
|
||
|
||
def write_file(self, repo_url: str, path: str, content: str, branch: str | None,
|
||
message) -> bool:
|
||
"""Commit `content` to `path` in the repo at `repo_url` (a web URL), creating
|
||
the file if needed; `message(old_content)` makes the commit message.
|
||
False if the file already said exactly this."""
|
||
proj = quote(urlsplit(repo_url).path.strip("/").split("/-/", 1)[0], safe="")
|
||
base = f"{self.api}/projects/{proj}"
|
||
if not branch:
|
||
branch = self._get(base).json()["default_branch"]
|
||
url = f"{base}/repository/files/{quote(path, safe='')}"
|
||
r = self._req("GET", url, ok=(404,), params={"ref": branch})
|
||
exists = r.status_code != 404
|
||
old = base64.b64decode(r.json()["content"]).decode("utf-8", "replace") if exists else ""
|
||
if exists and old == content:
|
||
return False
|
||
self._req("PUT" if exists else "POST", url,
|
||
json={"branch": branch, "content": content, "commit_message": message(old)})
|
||
return True
|
||
|
||
|
||
# --------------------------------------------------------------- parsing ----
|
||
|
||
FIELDS = ("Category", "Difficulty", "Event", "Author", "Tester", "Status")
|
||
HEADING_RE = re.compile(r"^\s{0,3}#{1,4}\s+(.+?)\s*$", re.M)
|
||
FIELD_RE = re.compile(r"^\s*[-*]\s*\**(?P<key>[A-Za-z ]+?)\**\s*:\s*(?P<val>.*?)\s*$", re.M)
|
||
# a value the author never touched: "[crypto, pwn, ...]" or "[Author Name]"
|
||
PLACEHOLDER_RE = re.compile(r"^\[.*\]$")
|
||
NO_TESTER_RE = re.compile(r"^(\?+|none|tester name|not found\b.*|tbd)(\s.*)?$", re.I)
|
||
|
||
STATUS_ORDER = ["Planning", "PoC Draft", "Testing Ready", "Tested", "Ready"]
|
||
STATUS_RE = re.compile(r"testing\s*ready|tested|ready|poc(?:\s*draft)?|planning", re.I)
|
||
STATUS_ICON = {"Planning": "⚪", "PoC Draft": "🟠", "Testing Ready": "🟡",
|
||
"Tested": "🔵", "Ready": "🟢", None: "❔"}
|
||
CATEGORIES = ["crypto", "pwn", "web", "rev", "misc", "osint", "forensic", "hardware"]
|
||
DIFFICULTIES = ["intro", "easy", "medium", "hard"]
|
||
DIFF_RE = re.compile(r"intro|easy|medium|hard", re.I)
|
||
DIFF_RANGE_RE = re.compile(r"(intro|easy|medium|hard)\s*(?:-|–|/|\bto\b)\s*(intro|easy|medium|hard)", re.I)
|
||
|
||
|
||
def clean(v: str) -> str:
|
||
v = re.sub(r"\\+([\[\]_*])", r"\1", v) # markdown escapes, sometimes doubled
|
||
return v.replace("**", "").strip()
|
||
|
||
|
||
def is_placeholder(v: str) -> bool:
|
||
v = clean(v)
|
||
return not v or bool(PLACEHOLDER_RE.match(v))
|
||
|
||
|
||
def stage_of(raw: str) -> tuple[str | None, bool]:
|
||
"""Map a free-text status onto STATUS_ORDER -> (stage, clear).
|
||
|
||
"Tested with h4tum-chall-format" is clearly Tested. "Planning to PoC Draft"
|
||
or a half-deleted option list is not; we take the least advanced stage it
|
||
names, so nothing looks further along than it is, and call it unclear."""
|
||
if is_placeholder(raw):
|
||
return None, False
|
||
found = set()
|
||
for m in STATUS_RE.finditer(raw):
|
||
w = m.group(0).lower()
|
||
found.add("Testing Ready" if w.startswith("testing") else
|
||
"PoC Draft" if w.startswith("poc") else w.title())
|
||
if not found:
|
||
return None, False
|
||
return min(found, key=STATUS_ORDER.index), len(found) == 1
|
||
|
||
|
||
def difficulty_of(raw: str) -> str | None:
|
||
""""medium/hard", "medium-hard", "easy to medium" -> a range "medium-hard"."""
|
||
if is_placeholder(raw):
|
||
return None
|
||
m = DIFF_RANGE_RE.search(raw)
|
||
if m:
|
||
ends = sorted({m.group(1).lower(), m.group(2).lower()}, key=DIFFICULTIES.index)
|
||
return "-".join(ends) if len(ends) > 1 else ends[0]
|
||
# prose like "currently medium, maybe also an easy and a hard version": the first one
|
||
m = DIFF_RE.search(raw)
|
||
return m.group(0).lower() if m else None
|
||
|
||
|
||
def difficulty_rank(d: str | None) -> float:
|
||
if not d:
|
||
return 99
|
||
ends = d.split("-")
|
||
return sum(DIFFICULTIES.index(e) for e in ends) / len(ends)
|
||
|
||
|
||
@dataclass
|
||
class Challenge:
|
||
repo: Repo
|
||
name: str = ""
|
||
fields: dict = field(default_factory=dict)
|
||
|
||
def get(self, key: str, dash: str = "—") -> str:
|
||
v = clean(self.fields.get(key, ""))
|
||
return dash if is_placeholder(v) else v
|
||
|
||
# -- derived -----------------------------------------------------------
|
||
@property
|
||
def named(self) -> bool:
|
||
return bool(self.name) and not is_placeholder(self.name)
|
||
|
||
@property
|
||
def title(self) -> str:
|
||
return self.name if self.named else self.repo.path
|
||
|
||
@property
|
||
def touched(self) -> bool:
|
||
"""Did anyone put *any* real content in the README template?"""
|
||
return self.named or any(not is_placeholder(self.fields.get(f, "")) for f in FIELDS)
|
||
|
||
@property
|
||
def status_raw(self) -> str:
|
||
return self.get("Status")
|
||
|
||
@property
|
||
def stage(self) -> str | None:
|
||
return stage_of(self.fields.get("Status", ""))[0]
|
||
|
||
@property
|
||
def difficulty(self) -> str | None:
|
||
return difficulty_of(self.fields.get("Difficulty", ""))
|
||
|
||
@property
|
||
def category(self) -> str | None:
|
||
"""The first category named; "rev, misc" counts as rev."""
|
||
raw = self.get("Category", "")
|
||
first = re.split(r"[,/&+]| and ", raw)[0].strip().lower()
|
||
return first or None
|
||
|
||
@property
|
||
def has_tester(self) -> bool:
|
||
t = self.get("Tester", "")
|
||
return bool(t) and not NO_TESTER_RE.match(t)
|
||
|
||
@property
|
||
def problems(self) -> list[str]:
|
||
p = []
|
||
stage, clear = stage_of(self.fields.get("Status", ""))
|
||
if stage is None:
|
||
p.append(f"status unknown ({self.status_raw!r})" if self.status_raw != "—"
|
||
else "no status")
|
||
elif not clear:
|
||
p.append(f"status unclear ({self.status_raw!r}, counted as {stage})")
|
||
if not self.category:
|
||
p.append("no category")
|
||
if not self.difficulty:
|
||
p.append(f"difficulty unclear ({self.get('Difficulty')!r})"
|
||
if self.get("Difficulty") != "—" else "no difficulty")
|
||
if is_testable(self) and not self.has_tester:
|
||
p.append("no tester")
|
||
if not self.named:
|
||
p.append("no challenge name in README")
|
||
return p
|
||
|
||
|
||
def is_testable(c: Challenge) -> bool:
|
||
return bool(c.stage) and STATUS_ORDER.index(c.stage) >= STATUS_ORDER.index("Testing Ready")
|
||
|
||
|
||
def parse_challenge(repo: Repo) -> Challenge:
|
||
c = Challenge(repo=repo)
|
||
text = repo.readme or ""
|
||
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")
|
||
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 Overview:
|
||
challenges: list # repos with a README somebody filled in
|
||
blank: list # repos with no README, or an untouched template
|
||
|
||
@property
|
||
def categories(self) -> list[str]:
|
||
extra = sorted({c.category or "?" for c in self.challenges} - set(CATEGORIES))
|
||
return CATEGORIES + extra
|
||
|
||
@property
|
||
def difficulties(self) -> list[str]:
|
||
seen = {c.difficulty for c in self.challenges if c.difficulty}
|
||
unknown = ["?"] if any(not c.difficulty for c in self.challenges) else []
|
||
return sorted(set(DIFFICULTIES) | seen, key=difficulty_rank) + unknown
|
||
|
||
@property
|
||
def stages(self) -> list:
|
||
"""The five stages, plus None ("?") only if some status is unreadable."""
|
||
return STATUS_ORDER + ([None] if any(not c.stage for c in self.challenges) else [])
|
||
|
||
@property
|
||
def testable(self) -> list:
|
||
"""Everything infra can already deploy and test: Testing Ready or beyond."""
|
||
return sorted((c for c in self.challenges if is_testable(c)),
|
||
key=lambda c: (self.categories.index(c.category or "?"),
|
||
difficulty_rank(c.difficulty), c.title.lower()))
|
||
|
||
def by_category(self) -> dict[str, list]:
|
||
out = defaultdict(list)
|
||
for c in self.challenges:
|
||
out[c.category or "?"].append(c)
|
||
for chs in out.values():
|
||
chs.sort(key=lambda c: (-STATUS_ORDER.index(c.stage) if c.stage else 1,
|
||
difficulty_rank(c.difficulty), c.title.lower()))
|
||
return out
|
||
|
||
|
||
def collect(gl: GitLab) -> Overview:
|
||
chs, blank = [], []
|
||
for repo in gl.repos():
|
||
c = parse_challenge(repo)
|
||
(chs if c.touched else blank).append(c)
|
||
return Overview(chs, sorted(blank, key=lambda c: c.repo.path.lower()))
|
||
|
||
|
||
def icons(chs: list) -> str:
|
||
return "".join(STATUS_ICON[c.stage] for c in chs)
|
||
|
||
|
||
def print_terminal(ov: Overview, c: bool) -> None:
|
||
def col(x): return x if c else ""
|
||
stages = ov.stages
|
||
groups = ov.by_category()
|
||
|
||
total = Counter(ch.stage for ch in ov.challenges)
|
||
print(f"\n{col(BOLD)}=== {len(ov.challenges)} challenges ==={col(RST)} "
|
||
+ " ".join(f"{STATUS_ICON[s]} {s or '?'}:{total[s]}" for s in stages if total[s]))
|
||
|
||
ready = ov.testable
|
||
print(f"\n{col(BOLD)}{col(GRN)}=== ready for infra testing ({len(ready)}) ==={col(RST)}")
|
||
for ch in ready:
|
||
print(f" {STATUS_ICON[ch.stage]} {ch.title[:34]:<35}{ch.category or '?':<10}"
|
||
f"{ch.difficulty or '?':<13}{col(DIM)}{ch.repo.web_url}{col(RST)}")
|
||
|
||
print(f"\n{col(BOLD)}=== status by category ==={col(RST)}")
|
||
print(f" {'':<10}" + "".join(f"{s or '?':>14}" for s in stages) + f"{'total':>7}")
|
||
for cat in ov.categories:
|
||
chs = groups.get(cat, [])
|
||
cnt = Counter(ch.stage for ch in chs)
|
||
tint = "" if chs else DIM
|
||
print(f" {col(tint)}{cat:<10}" + "".join(f"{cnt[s] or '·':>14}" for s in stages)
|
||
+ f"{len(chs):>7}{col(RST)}")
|
||
|
||
print(f"\n{col(BOLD)}=== category × difficulty ==={col(RST)} "
|
||
+ " ".join(f"{STATUS_ICON[s]} {s}" for s in STATUS_ORDER))
|
||
diffs = ov.difficulties
|
||
print(f" {'':<10}" + "".join(f"{d:<13}" for d in diffs))
|
||
for cat in ov.categories:
|
||
chs = groups.get(cat, [])
|
||
cells = [icons([ch for ch in chs if (ch.difficulty or "?") == d]) for d in diffs]
|
||
# emoji are two columns wide in a terminal
|
||
print(f" {cat:<10}" + "".join(f"{x or '·'}{' ' * (13 - (2 * len(x) if x else 1))}"
|
||
for x in cells))
|
||
|
||
print(f"\n{col(BOLD)}=== needs attention ==={col(RST)}")
|
||
for ch in sorted(ov.challenges, key=lambda c: c.repo.path.lower()):
|
||
if ch.problems:
|
||
print(f" {col(YEL)}{ch.repo.path:<32}{col(RST)} {col(DIM)}"
|
||
+ "; ".join(ch.problems) + col(RST))
|
||
for ch in ov.blank:
|
||
why = "no README" if ch.repo.readme is None else "README is still the blank template"
|
||
print(f" {col(RED)}{ch.repo.path:<32}{col(RST)} {col(DIM)}{why}{col(RST)}")
|
||
print()
|
||
|
||
|
||
def render_challs(ov: Overview) -> str:
|
||
"""The deployment repo's challs.txt: one clone URL per line, for everything
|
||
that is Testing Ready or beyond."""
|
||
return "".join(f"{c.repo.ssh_url}\n" for c in ov.testable)
|
||
|
||
|
||
def challs_message(old: str, new: str, path: str) -> str:
|
||
name = lambda u: u.rsplit("/", 1)[-1].removesuffix(".git")
|
||
was, now = set(old.split()), set(new.split())
|
||
parts = [f"+{name(u)}" for u in sorted(now - was)] + [f"-{name(u)}" for u in sorted(was - now)]
|
||
return f"Update {path} from challenge status" + (f": {' '.join(parts)}" if parts else "")
|
||
|
||
|
||
def md_cell(v: str) -> str:
|
||
return v.replace("|", "\\|")
|
||
|
||
|
||
def render_hedgedoc(ov: Overview, title: str, tags: str = "") -> str:
|
||
stamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
|
||
stages = ov.stages
|
||
groups = ov.by_category()
|
||
total = Counter(ch.stage for ch in ov.challenges)
|
||
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` from the README.md of every repo in the "
|
||
f"GitLab challenges group — {len(ov.challenges)} challenges*", "",
|
||
" · ".join(f"{STATUS_ICON[s]} **{s or 'unknown'}** {total[s]}"
|
||
for s in stages if total[s] or s), ""]
|
||
|
||
ready = ov.testable
|
||
L += [f"## ✅ Ready for infra testing ({len(ready)} of {len(ov.challenges)})", "",
|
||
"*every challenge at Testing Ready or beyond — these can be deployed and "
|
||
"tested against the infrastructure now*", ""]
|
||
if ready:
|
||
L += ["| | Challenge | Category | Difficulty | Status | Event | Author | Tester |",
|
||
"| --- | --- | --- | --- | --- | --- | --- | --- |"]
|
||
for ch in ready:
|
||
L.append("| " + " | ".join(md_cell(x) for x in (
|
||
STATUS_ICON[ch.stage], f"[{ch.title}]({ch.repo.web_url})",
|
||
ch.category or "?", ch.difficulty or "?", ch.stage, ch.get("Event"),
|
||
ch.get("Author"), ch.get("Tester") if ch.has_tester else "—")) + " |")
|
||
else:
|
||
L.append("*nothing yet*")
|
||
L.append("")
|
||
|
||
L += ["## Status by category", "",
|
||
"| Category | " + " | ".join(f"{STATUS_ICON[s]} {s or '?'}" for s in stages)
|
||
+ " | total |",
|
||
"| --- |" + " ---: |" * (len(stages) + 1)]
|
||
for cat in ov.categories:
|
||
chs = groups.get(cat, [])
|
||
cnt = Counter(ch.stage for ch in chs)
|
||
L.append(f"| {cat} | " + " | ".join(str(cnt[s] or "·") for s in stages)
|
||
+ f" | **{len(chs)}** |")
|
||
L.append("| **total** | " + " | ".join(f"**{total[s]}**" for s in stages)
|
||
+ f" | **{len(ov.challenges)}** |")
|
||
|
||
diffs = ov.difficulties
|
||
L += ["", "## Category × difficulty", "",
|
||
"*one dot per challenge, coloured by status: "
|
||
+ " ".join(f"{STATUS_ICON[s]} {s}" for s in STATUS_ORDER) + "*", "",
|
||
"| Category | " + " | ".join(diffs) + " | total |",
|
||
"| --- |" + " --- |" * len(diffs) + " ---: |"]
|
||
for cat in ov.categories:
|
||
chs = groups.get(cat, [])
|
||
cells = [icons([ch for ch in chs if (ch.difficulty or "?") == d]) or "·" for d in diffs]
|
||
L.append(f"| {cat} | " + " | ".join(cells) + f" | **{len(chs)}** |")
|
||
dtot = Counter(ch.difficulty or "?" for ch in ov.challenges)
|
||
L.append("| **total** | " + " | ".join(f"**{dtot[d]}**" for d in diffs)
|
||
+ f" | **{len(ov.challenges)}** |")
|
||
|
||
L += ["", "## Challenges"]
|
||
for cat in ov.categories:
|
||
chs = groups.get(cat)
|
||
if not chs:
|
||
continue
|
||
L += ["", f"### {cat} ({len(chs)})", "",
|
||
"| | Challenge | Difficulty | Status | Event | Author | Tester | Last activity |",
|
||
"| --- | --- | --- | --- | --- | --- | --- | --- |"]
|
||
for ch in chs:
|
||
status = ch.stage or "?"
|
||
if ch.status_raw.lower() != status.lower():
|
||
status += f" *({ch.status_raw})*"
|
||
cat_raw = ch.get("Category", "")
|
||
name = f"[{md_cell(ch.title)}]({ch.repo.web_url})"
|
||
if cat_raw.lower() != cat:
|
||
name += f" *({cat_raw})*"
|
||
L.append("| " + " | ".join(md_cell(x) for x in (
|
||
STATUS_ICON[ch.stage], name, ch.difficulty or ch.get("Difficulty"), status,
|
||
ch.get("Event"), ch.get("Author"),
|
||
ch.get("Tester") if ch.has_tester else "—", ch.repo.last_activity)) + " |")
|
||
|
||
L += ["", "## Needs attention", ""]
|
||
issues = [ch for ch in sorted(ov.challenges, key=lambda c: c.repo.path.lower()) if ch.problems]
|
||
for ch in issues:
|
||
L.append(f"- [{md_cell(ch.title)}]({ch.repo.web_url}): " + "; ".join(ch.problems))
|
||
for ch in ov.blank:
|
||
why = "no README" if ch.repo.readme is None else "README is still the blank template"
|
||
L.append(f"- [{ch.repo.path}]({ch.repo.web_url}): {why}")
|
||
if not issues and not ov.blank:
|
||
L.append("*nothing — every README is filled in. 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 /<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:
|
||
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("--challs", action="store_true",
|
||
help="commit the clone URLs of every Testing Ready+ challenge to "
|
||
f"CHALLS_PATH (default {DEFAULT_CHALLS_PATH!r}) in the "
|
||
"deployment repo CHALLS_REPO_URL, if they changed")
|
||
ap.add_argument("--challs-out", metavar="FILE",
|
||
help="write that challs.txt to a local file instead ('-' for stdout)")
|
||
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 GitLab walk
|
||
except ValueError as e:
|
||
sys.exit(str(e))
|
||
if args.challs and not cfg.get("CHALLS_REPO_URL"):
|
||
sys.exit("--challs needs CHALLS_REPO_URL in the config")
|
||
if note_url or args.challs:
|
||
lock = single_run_lock() # noqa: F841 (held for the lifetime of the process)
|
||
|
||
gl = GitLab(cfg["GITLAB_GROUP_URL"], cfg["GITLAB_TOKEN"])
|
||
try:
|
||
ov = collect(gl)
|
||
except (requests.RequestException, RuntimeError) as e:
|
||
sys.exit(f"reading {cfg['GITLAB_GROUP_URL']} failed: {e}")
|
||
if not ov.challenges and not ov.blank:
|
||
sys.exit(f"no repos found in {cfg['GITLAB_GROUP_URL']!r}")
|
||
|
||
md = render_hedgedoc(ov, title, tags)
|
||
if args.out:
|
||
Path(args.out).write_text(md)
|
||
if args.stdout:
|
||
print(md, end="")
|
||
elif not args.quiet:
|
||
print_terminal(ov, 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))
|
||
challs = render_challs(ov)
|
||
if args.challs_out == "-":
|
||
print(challs, end="")
|
||
elif args.challs_out:
|
||
Path(args.challs_out).write_text(challs)
|
||
|
||
# one failing target must not keep the other from being updated
|
||
failed = []
|
||
if note_url:
|
||
try:
|
||
changed = update_hedgedoc(note_url, md, retries=args.retries, quiet=args.quiet)
|
||
if changed or not args.quiet:
|
||
print(("updated → " if changed else "unchanged → ") + note_url)
|
||
except (requests.RequestException, RuntimeError, ValueError, KeyError) as e:
|
||
failed.append(f"hedgedoc update of {note_url} failed: {e}")
|
||
if args.challs:
|
||
repo, path = cfg["CHALLS_REPO_URL"], cfg.get("CHALLS_PATH") or DEFAULT_CHALLS_PATH
|
||
try:
|
||
changed = gl.write_file(repo, path, challs, cfg.get("CHALLS_BRANCH"),
|
||
lambda old: challs_message(old, challs, path))
|
||
if changed or not args.quiet:
|
||
print(("committed → " if changed else "unchanged → ") + f"{repo} {path}")
|
||
except (requests.RequestException, RuntimeError, KeyError) as e:
|
||
failed.append(f"challs.txt update in {repo} failed: {e}")
|
||
if failed:
|
||
sys.exit("\n".join(failed))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|