challenge_status_tracker/ctf_status.py

#!/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. The rules are in tracker/checks.py.

Config: see .env.example -> .env (chmod 600)
"""
from __future__ import annotations

import argparse
import fcntl
import sys
import tempfile
from pathlib import Path

import requests

from tracker.challs import challs_message, render_challs
from tracker.config import DEFAULT_CHALLS_PATH, DEFAULT_TAGS, DEFAULT_TITLE, load_env
from tracker.gitlab import GitLab
from tracker.hedgedoc import HedgeDocNote, push_hedgedoc, update_hedgedoc
from tracker.markdown import render_hedgedoc
from tracker.overview import collect
from tracker.terminal import print_terminal


# the lock is released when its file is closed, so keep them open until we exit
_held_locks = []


def single_run_lock(target: str) -> None:
    """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")
    _held_locks.append(fh)


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")
    if note_url:
        single_run_lock("hedgedoc")
    if args.challs:
        single_run_lock("challs")

    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()