add chall tracker to deployment
This commit is contained in:
+78
-7
@@ -11,6 +11,7 @@ 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
|
||||
@@ -31,11 +32,13 @@ import requests
|
||||
# ---------------------------------------------------------------- config ----
|
||||
|
||||
CFG_KEYS = ("GITLAB_GROUP_URL", "GITLAB_TOKEN")
|
||||
OPT_KEYS = ("HEDGEDOC_NOTE", "HEDGEDOC_TAGS", "TITLE")
|
||||
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:
|
||||
@@ -68,13 +71,15 @@ def load_env() -> dict:
|
||||
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:
|
||||
"""Read-only access to one group. The token needs Project: Read (to list the
|
||||
group) and Repository: Read (to download files)."""
|
||||
"""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)
|
||||
@@ -85,7 +90,12 @@ class GitLab:
|
||||
self.s.headers["PRIVATE-TOKEN"] = token
|
||||
|
||||
def _get(self, url: str, **params) -> requests.Response:
|
||||
r = self.s.get(url, params=params, timeout=self.timeout)
|
||||
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:
|
||||
@@ -121,12 +131,32 @@ class GitLab:
|
||||
|
||||
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 ----
|
||||
|
||||
@@ -383,6 +413,19 @@ def print_terminal(ov: Overview, c: bool) -> None:
|
||||
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("|", "\\|")
|
||||
|
||||
@@ -668,6 +711,12 @@ def main() -> None:
|
||||
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 "
|
||||
@@ -691,6 +740,9 @@ def main() -> None:
|
||||
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"])
|
||||
@@ -712,13 +764,32 @@ def main() -> None:
|
||||
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:
|
||||
sys.exit(f"hedgedoc update of {note_url} failed: {e}")
|
||||
if changed or not args.quiet:
|
||||
print(("updated → " if changed else "unchanged → ") + note_url)
|
||||
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__":
|
||||
|
||||
Reference in New Issue
Block a user