attack-defense-platform/receiver/challenges/xvi/Challenge.py

import logging
import os
import random
import string

from config import get_settings


class Challenge(object):
    name = __name__
    settings = get_settings()
    port = 0
    # Host the checker connects to. The team receiver runs on the same machine
    # as the published team ports, so 127.0.0.1 is correct; override with
    # RECEIVER_HOST if a receiver ever runs off-host.
    host = os.environ.get("RECEIVER_HOST", "127.0.0.1")
    # URL scheme for this challenge. TLS services (gleam-drive/bandit) serve
    # HTTPS only, so a plain http:// request fails against a healthy service.
    # Read from CHALLENGE_SCHEME_<NAME> by the concrete checker via _scheme();
    # this default is overridden per class where needed.
    scheme = os.environ.get("RECEIVER_SCHEME", "http")

    def __init__(self, port, host=None):
        self.port = port
        if host:
            self.host = host
        self.add_logger()

    def url(self, path=""):
        """Absolute URL for the challenge service.

        Every XVI checker (Art, XL, S3, ...) calls self.url(...) — without this
        helper they all fail with "'<Class>' object has no attribute 'url'" and
        the receiver reports the service DOWN while it is actually healthy.

        The scheme is per-challenge: a TLS service (CHALLENGE_SCHEME_<NAME>=https)
        must be reached over https or requests raises an SSLError. The env key is
        derived from the class module name, which the generator renders as the
        challenge name, with hyphens normalized to underscores.
        """
        p = "" if path.startswith("/") else "/"
        key = self.name.upper().replace("-", "_")
        scheme = os.environ.get(f"CHALLENGE_SCHEME_{key}") or self.scheme
        return f"{scheme}://{self.host}:{self.port}{p}{path}"

    def add_logger(self):
        self.logger = logging.getLogger()

    def random_string(self, length):
        charset = string.ascii_uppercase + string.ascii_lowercase + string.digits
        return ''.join(random.choice(charset) for i in range(length))

    def distribute(self, flag):
        raise NotImplementedError

    def check(self):
        raise NotImplementedError

    def credentials(self):
        pwd = os.environ.get(f'PASSWORD_{self.port}')
        if not pwd:
            pwd = getattr(self.settings, f'PASSWORD_{self.port}', '')
        return {
            'username': 'root',
            'password': pwd,
        }