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 in the container.  Protocols:
  - TCP/netcat   : asmr, bit-canvas, go-green (xinetd banner) & ticketer (socat)
  - HTTP GET     : anti-alchemy, fjb, gift-card, gift-voucher, gleam-drive,
                   kode-viewer, more-less, tempest-poc
"""
import os
import socket
import subprocess

import requests

from .Challenge import Challenge


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 -f {path} && cat {path} || echo MISSING")
    return bool(r and "MISSING" not in (r.stdout or "") and r.returncode == 0)


def _tcp_banner(port: int, timeout: float = 4.0, expect: bytes = None) -> bool:
    try:
        s = socket.create_connection(("127.0.0.1", port), timeout=timeout)
        s.settimeout(timeout)
        data = s.recv(256)
        s.close()
        if expect:
            return expect.lower() in data.lower()
        return len(data) > 0
    except Exception:
        return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code in (200, 302, 500) or len(r.text) > 0
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_ANTI_ALCHEMY", "anti-alchemy_container"))
        except Exception:
            return False


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

    def check(self):
        try:
            return _tcp_banner(self.port, expect=None)
        except Exception:
            return False


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

    def check(self):
        try:
            return _tcp_banner(self.port, expect=None)
        except Exception:
            return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_FJB", "fjb_container"))
        except Exception:
            return False


class GiftCard(Challenge):
    flag_location = "flags/gift-card.txt"
    history_location = "history/gift-card.txt"

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_GIFT_CARD", "gift-card_container"),
                                      "/ctf/gift-card/flag.txt")
        except Exception:
            return False


class GiftVoucher(Challenge):
    flag_location = "flags/gift-voucher.txt"
    history_location = "history/gift-voucher.txt"

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_GIFT_VOUCHER", "gift-voucher_container"),
                                      "/ctf/gift-voucher/flag.txt")
        except Exception:
            return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_GLEAM_DRIVE", "gleam-drive_container"))
        except Exception:
            return False


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

    def check(self):
        try:
            return _tcp_banner(self.port, expect=None)
        except Exception:
            return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_KODE_VIEWER", "kode-viewer_container"))
        except Exception:
            return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_MORE_LESS", "more-less_container"))
        except Exception:
            return False


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

    def check(self):
        try:
            url = f"http://localhost:{self.port}/"
            r = requests.get(url, timeout=6)
            assert r.status_code < 500
            return _flag_in_container(os.environ.get("CHALLENGE_CONTAINER_TEMPEST_POC", "tempest-poc_container"))
        except Exception:
            return False


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

    def check(self):
        try:
            return _tcp_banner(self.port, expect=None)
        except Exception:
            return False