Files
attack-defense-platform/receiver/challenges/xvii/Challenge.py
T
Cyrene ae50acfe40 fix(ssh): per-challenge SSH login + phew buffering/leak/timeout
Passwords failed on 10/16 challenges while state.json looked correct:
- only the 6 native GEMASTIK XVIII images provision 'ctfuser'; every imported
  XVI/XVII image does 'echo root:${PASSWORD} | chpasswd' and logs in as root.
  set_ssh_passwords() hardcoded ctfuser, so chpasswd set a password on an
  account nobody uses -> 'Permission denied' everywhere.
  Registry gains a per-challenge 'ssh_user'; chpasswd now targets the real
  login (and ctfuser/ctf when present) and reports failures loudly.
- phew checker: chall.py block-buffers stdout through the docker exec pipe
  (PYTHONUNBUFFERED now set) and leaks chall.py inside the container on
  timeout (26 orphans, container saturated) -> reaps the whole exec process
  group. Startup does a fresh Pailier keygen (~12 s) so crypto reads need
  _CRYPTO_TIMEOUT, not the 5 s prompt default.

Adds panel/verify_ssh_creds.py (proves the state->container binding from
inside via a real login), audit_ssh_users.sh, reset_runtime.sh.
2026-09-26 14:40:10 +08:00

48 lines
1.3 KiB
Python

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. Same machine as the published team ports;
# override with RECEIVER_HOST if a receiver ever runs off-host.
host = os.environ.get("RECEIVER_HOST", "127.0.0.1")
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 (see xvi/Challenge.py)."""
p = "" if path.startswith("/") else "/"
return f"http://{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': 'ctfuser',
'password': pwd,
}