Casually9790andClaude Opus 5.5
4c6cb21cb8
Anyone can set the README status, and several challenges imported from
Nextcloud claimed Testing Ready with an untouched description.yml. For
every challenge claiming Testing Ready or beyond, now also read the repo
tree and description.yml, and block it if description.yml is missing or
still the template, its category disagrees with the README, the flag is
not h4tum{...}, a dynamic flag reaches no container, a deploy_dir or its
Dockerfile is missing, or there is no solve.py.
Blocked challenges keep their claimed status in the counts but are left
out of "Ready for infra testing" and challs.txt, and listed with the
reason in the terminal and the HedgeDoc note. Adds a PyYAML dependency.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
930 lines
39 KiB
Python
Executable File
930 lines
39 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
|
||
|
||
A README status of Testing Ready or beyond is only believed if the repo backs it
|
||
up: description.yml is filled in (not the template), its flag and containers make
|
||
sense, and the deploy dirs and solve.py exist. Otherwise the challenge is shown
|
||
as blocked and kept out of challs.txt.
|
||
|
||
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
|
||
import yaml
|
||
|
||
# ---------------------------------------------------------------- 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
|
||
id: int = 0
|
||
branch: str | None = None
|
||
# only fetched for challenges whose README claims Testing Ready or beyond
|
||
files: set[str] | None = None # every file and dir in the repo
|
||
desc: str | None = None # description.yml, None if there is none
|
||
|
||
|
||
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
|
||
return self._raw(p["id"], url.split(f"/-/blob/{branch}/", 1)[1], branch)
|
||
|
||
def _raw(self, project_id: int, path: str, branch: str) -> str:
|
||
r = self._get(f"{self.api}/projects/{project_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), id=p["id"], branch=p.get("default_branch"))
|
||
|
||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||
return list(pool.map(fetch, projects))
|
||
|
||
def load_files(self, repo: Repo) -> None:
|
||
"""Fill in repo.files and repo.desc, for check_files()."""
|
||
repo.files = set()
|
||
if not repo.branch:
|
||
return
|
||
page = "1"
|
||
while page:
|
||
r = self._get(f"{self.api}/projects/{repo.id}/repository/tree", ref=repo.branch,
|
||
recursive="true", per_page="100", page=page)
|
||
repo.files |= {t["path"] for t in r.json()}
|
||
page = r.headers.get("X-Next-Page", "")
|
||
if "description.yml" in repo.files:
|
||
repo.desc = self._raw(repo.id, "description.yml", repo.branch)
|
||
|
||
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)
|
||
CATEGORY_SPLIT_RE = re.compile(r"[,/&+]| and ")
|
||
|
||
BLOCKED_ICON = "⛔"
|
||
FLAG_RE = re.compile(r"^h4tum\{[^{}]+\}$")
|
||
# what every repo starts with: the challenge-format example "netcat-check-multi-container"
|
||
TEMPLATE_ID = "netcat-check-multi-container"
|
||
TEMPLATE_NAME = "Netcat Check Multi Container"
|
||
TEMPLATE_AUTHOR = "jh"
|
||
TEMPLATE_FLAG = "h4tum{n3tc4t_1s_c0ol}"
|
||
TEMPLATE_CONTAINERS = {"cont1", "cont2"}
|
||
TEMPLATE_DIRS = {"container-1", "container-2"}
|
||
|
||
|
||
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)
|
||
blockers: list = field(default_factory=list) # from check_files()
|
||
|
||
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 = CATEGORY_SPLIT_RE.split(raw)[0].strip().lower()
|
||
return first or None
|
||
|
||
@property
|
||
def claims_testable(self) -> bool:
|
||
"""The README says Testing Ready or beyond; is_testable() also asks the repo."""
|
||
return bool(self.stage) and STATUS_ORDER.index(self.stage) >= STATUS_ORDER.index("Testing Ready")
|
||
|
||
@property
|
||
def icon(self) -> str:
|
||
return BLOCKED_ICON if self.blockers else STATUS_ICON[self.stage]
|
||
|
||
@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 self.claims_testable 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 c.claims_testable and not c.blockers
|
||
|
||
|
||
def check_files(c: Challenge) -> list[str]:
|
||
"""Reasons not to believe a Testing Ready+ status: cheap signs in the repo that
|
||
the challenge was never finished. Passing is no proof that it builds or solves."""
|
||
repo = c.repo
|
||
if repo.desc is None:
|
||
return ["no description.yml"]
|
||
try:
|
||
doc = yaml.safe_load(repo.desc)
|
||
except yaml.YAMLError as e:
|
||
return [f"description.yml does not parse ({str(e).splitlines()[0]})"]
|
||
if not isinstance(doc, dict) or len(doc) != 1 or not isinstance(next(iter(doc.values())), dict):
|
||
return ["description.yml is not a single `<id>: {...}` block"]
|
||
cid, d = next(iter(doc.items()))
|
||
flag = d.get("flag") if isinstance(d.get("flag"), dict) else {}
|
||
containers = [x for x in d.get("containers") or [] if isinstance(x, dict)]
|
||
dirs = [os.path.normpath(str(x.get("deploy_dir") or ".")) for x in containers]
|
||
b = []
|
||
|
||
template = [what for what, hit in (
|
||
(f"id {cid!r}", cid == TEMPLATE_ID),
|
||
(f"name {d.get('name')!r}", d.get("name") == TEMPLATE_NAME),
|
||
(f"author {d.get('author')!r}", d.get("author") == TEMPLATE_AUTHOR),
|
||
("flag", flag.get("content") == TEMPLATE_FLAG),
|
||
("containers", any(x.get("name") in TEMPLATE_CONTAINERS for x in containers)
|
||
or bool(TEMPLATE_DIRS & set(dirs))),
|
||
) if hit]
|
||
if template:
|
||
b.append("description.yml still has the template's " + ", ".join(template))
|
||
|
||
cat = str(d.get("category") or "").strip().lower()
|
||
readme_cats = {x.strip().lower() for x in CATEGORY_SPLIT_RE.split(c.get("Category", ""))}
|
||
if c.category and cat not in readme_cats:
|
||
b.append(f"category {cat or 'missing'} in description.yml, README says {c.get('Category')}")
|
||
|
||
content = str(flag.get("content") or "")
|
||
if not FLAG_RE.match(content):
|
||
b.append(f"flag {content} is not h4tum{{...}}" if content else "no flag in description.yml")
|
||
if flag.get("dynamic") and not any(isinstance(x.get("flag"), dict) for x in containers):
|
||
b.append("dynamic flag, but no container is handed one")
|
||
|
||
for dd in dirs:
|
||
if dd != "." and dd not in repo.files:
|
||
b.append(f"deploy_dir {dd} does not exist")
|
||
elif ("Dockerfile" if dd == "." else f"{dd}/Dockerfile") not in repo.files:
|
||
b.append(f"no Dockerfile in {dd}")
|
||
if not any(p == "solve.py" or p.endswith("/solve.py") for p in repo.files):
|
||
b.append("no solve.py")
|
||
return b
|
||
|
||
|
||
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=self._ready_order)
|
||
|
||
@property
|
||
def blocked(self) -> list:
|
||
"""Testing Ready or beyond according to the README, but not the repo."""
|
||
return sorted((c for c in self.challenges if c.claims_testable and c.blockers),
|
||
key=self._ready_order)
|
||
|
||
def _ready_order(self, c: Challenge) -> tuple:
|
||
return (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)
|
||
|
||
def check(c: Challenge) -> None:
|
||
gl.load_files(c.repo)
|
||
c.blockers = check_files(c)
|
||
|
||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||
list(pool.map(check, [c for c in chs if c.claims_testable]))
|
||
return Overview(chs, sorted(blank, key=lambda c: c.repo.path.lower()))
|
||
|
||
|
||
def icons(chs: list) -> str:
|
||
return "".join(c.icon 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)}")
|
||
|
||
if ov.blocked:
|
||
print(f"\n{col(BOLD)}{col(RED)}=== {BLOCKED_ICON} claims {STATUS_ORDER[2]}+, "
|
||
f"but the repo disagrees ({len(ov.blocked)}) ==={col(RST)}")
|
||
for ch in ov.blocked:
|
||
print(f" {STATUS_ICON[ch.stage]} {ch.title[:34]:<35}{col(DIM)}"
|
||
+ "; ".join(ch.blockers) + 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) + f" {BLOCKED_ICON} blocked")
|
||
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 and not blocked."""
|
||
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 whose repo backs that up — "
|
||
"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("")
|
||
|
||
if ov.blocked:
|
||
L += [f"## {BLOCKED_ICON} Claims Testing Ready+, but the repo disagrees "
|
||
f"({len(ov.blocked)})", "",
|
||
"*the README says so, but description.yml or the repo shows it is not "
|
||
"finished — kept out of challs.txt until this is fixed*", ""]
|
||
for ch in ov.blocked:
|
||
L.append(f"- {STATUS_ICON[ch.stage]} [{md_cell(ch.title)}]({ch.repo.web_url}) "
|
||
f"({ch.stage}): " + "; ".join(ch.blockers))
|
||
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)
|
||
+ f" {BLOCKED_ICON} blocked*", "",
|
||
"| 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 (
|
||
ch.icon, 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(target: str):
|
||
"""Keep a slow or hung cron run from colliding with the next one. One lock per
|
||
target, so the daily challs.txt run and the HedgeDoc refresh can overlap."""
|
||
fh = open(Path(tempfile.gettempdir()) / f"ctf_status-{target}.lock", "w")
|
||
try:
|
||
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except OSError:
|
||
sys.exit(f"another ctf_status.py {target} 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 unblocked 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")
|
||
# held for the lifetime of the process
|
||
locks = [single_run_lock(t) for t, on in (("hedgedoc", note_url), ("challs", args.challs))
|
||
if on] # noqa: F841
|
||
|
||
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()
|