attack-defense-platform/receiver/challenges/xvii/checkers.py

"""SLA checkers for GEMASTIK XVII challenges (imported from
github.com/vidner/gemastik-xvii-final — no upstream receiver was provided).

Each checker validates liveness (+ flag presence where the flag is a file) with
STRICT timeouts so a wedged service can never hang the receiver's check loop.

Protocol classes:
  - WEB     : HTTP GET on the challenge port
  - TCP     : socat/xinetd line service — connect and expect a prompt/banner
  - WEB+FLAG: WEB plus the flag must be mounted inside the container

Env-var convention: systemd Environment= keys are normalized to underscores
(CHALLENGE_PORT_GIFT_CARD for the challenge "gift-card"), so the lookup helper
does the same normalization.
"""
import os
import socket
import subprocess

import requests

from .Challenge import Challenge


def _key(name: str) -> str:
    return name.upper().replace("-", "_")


def _container(challenge: str) -> str:
    return os.environ.get(
        f"CHALLENGE_CONTAINER_{_key(challenge)}", f"{challenge}_container"
    )


def _docker_exec(container: str, *args, timeout: int = 10):
    try:
        return subprocess.run(["docker", "exec", container, *args],
                              capture_output=True, text=True, timeout=timeout)
    except Exception:
        return None


def _flag_in_container(container: str, path: str = "/flag.txt") -> bool:
    r = _docker_exec(container, "sh", "-c", f"test -s {path} && echo FLAG_OK || echo MISSING")
    return bool(r and "FLAG_OK" in (r.stdout or ""))


def _web_alive(port: int, timeout: float = 5.0, scheme: str = None) -> bool:
    """Any HTTP response (even 4xx/5xx) proves the listener is up.

    Some challenges serve TLS (gleam-drive's bandit runs
    `Listening on https://localhost:8000`). A plain http:// GET against a TLS
    port returns an SSLError/wrong-version-number, which reads as "service
    down" even though it is perfectly healthy. The scheme is per-challenge and
    comes from CHALLENGE_SCHEME_<NAME>; when unset, try http first then fall
    back to https so a mis-tagged challenge still checks correctly.
    """
    schemes = [scheme] if scheme else ["http", "https"]
    for sch in schemes:
        if not sch:
            continue
        try:
            # verify=False is required, not lazy: the challenge serves a
            # self-signed cert from inside the contest network, so there is no
            # CA to validate against. This probe only proves the listener
            # answers — it never carries a secret, and a MITM here would gain
            # nothing beyond the liveness bit we already discard.
            r = requests.get(f"{sch}://127.0.0.1:{port}/", timeout=timeout,
                             allow_redirects=False, verify=False)
            if r.status_code < 600:
                return True
        except Exception:
            continue
    return False


def _scheme(challenge: str) -> str | None:
    """Per-challenge URL scheme from CHALLENGE_SCHEME_<NAME> (underscored)."""
    return os.environ.get(f"CHALLENGE_SCHEME_{_key(challenge)}", "") or None


def _tcp_alive(port: int, timeout: float = 5.0, send: bytes = None) -> bool:
    """Connect to a line service and read a prompt/banner (or survive silence).

    Some socat services wait for input before greeting, so a successful connect
    with no data is also treated as alive; a refused connection is not.
    """
    try:
        s = socket.create_connection(("127.0.0.1", port), timeout=timeout)
    except Exception:
        return False
    try:
        s.settimeout(timeout)
        if send:
            s.sendall(send)
        try:
            data = s.recv(256)
        except socket.timeout:
            # connected but silent — service is accepting connections
            return True
        return True if data is not None else True
    finally:
        try:
            s.close()
        except Exception:
            pass


class AntiAlchemy(Challenge):
    flag_location = "flags/anti-alchemy.txt"
    history_location = "history/anti-alchemy.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("anti-alchemy")) and _flag_in_container(_container("anti-alchemy"))


class Asmr(Challenge):
    flag_location = "flags/asmr.txt"
    history_location = "history/asmr.txt"

    def check(self):
        return _tcp_alive(self.port)


class BitCanvas(Challenge):
    flag_location = "flags/bit-canvas.txt"
    history_location = "history/bit-canvas.txt"

    def check(self):
        return _tcp_alive(self.port)


class Fjb(Challenge):
    flag_location = "flags/fjb.txt"
    history_location = "history/fjb.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("fjb")) and _flag_in_container(_container("fjb"))


class GiftCard(Challenge):
    """socat TCP line service (python main.py on :5000), NOT http."""
    flag_location = "flags/gift-card.txt"
    history_location = "history/gift-card.txt"

    def check(self):
        return _tcp_alive(self.port, send=b"1\n") and _flag_in_container(
            _container("gift-card"), "/ctf/gift-card/flag.txt")


class GiftVoucher(Challenge):
    """socat TCP line service, NOT http."""
    flag_location = "flags/gift-voucher.txt"
    history_location = "history/gift-voucher.txt"

    def check(self):
        return _tcp_alive(self.port, send=b"1\n") and _flag_in_container(
            _container("gift-voucher"), "/ctf/gift-voucher/flag.txt")


class GleamDrive(Challenge):
    flag_location = "flags/gleam-drive.txt"
    history_location = "history/gleam-drive.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("gleam-drive")) and _flag_in_container(_container("gleam-drive"))


class GoGreen(Challenge):
    flag_location = "flags/go-green.txt"
    history_location = "history/go-green.txt"

    def check(self):
        return _tcp_alive(self.port)


class KodeViewer(Challenge):
    flag_location = "flags/kode-viewer.txt"
    history_location = "history/kode-viewer.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("kode-viewer")) and _flag_in_container(_container("kode-viewer"))


class MoreLess(Challenge):
    flag_location = "flags/more-less.txt"
    history_location = "history/more-less.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("more-less")) and _flag_in_container(_container("more-less"))


class TempestPoc(Challenge):
    flag_location = "flags/tempest-poc.txt"
    history_location = "history/tempest-poc.txt"

    def check(self):
        return _web_alive(self.port, scheme=_scheme("tempest-poc")) and _flag_in_container(_container("tempest-poc"))


class Ticketer(Challenge):
    flag_location = "flags/ticketer.txt"
    history_location = "history/ticketer.txt"

    def check(self):
        return _tcp_alive(self.port)