#!/usr/bin/env python3 """Star Trek themed GET-only CTF questionnaire server.""" from __future__ import annotations import argparse import base64 import mimetypes import os import random import re import secrets import ssl import sys import time import unicodedata from dataclasses import dataclass from difflib import SequenceMatcher from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Callable, Mapping from urllib.parse import parse_qs, unquote, urlparse DEFAULT_QUESTIONS_FILE = Path(__file__).with_name("questions.md") DEFAULT_STATIC_QUEST_FILE = Path(__file__).with_name("static_quest.txt") DEFAULT_FILE_ROOT = Path(__file__).resolve().parent DEFAULT_FLAG = "Vigenere key: JDUWZCBAOWZFBYIGVWSNY" DEFAULT_QUEST_LENGTH = 10 DEFAULT_SESSION_TTL_SECONDS = 30 * 60 DEFAULT_SESSION_START_LIMIT = 6 DEFAULT_SESSION_START_WINDOW_SECONDS = 10 * 60 DEFAULT_USER_AGENT_PREFIX = "curl" DECOY_STATUS_CODE = 266 USER_AGENT_PREFIX_LENGTH = 4 QUEST_PATH = "/q" QUESTION_ENCODING_STEPS = 5 QUESTION_ENCODINGS = ("b64", "b85") FILLER_WORDS = { "a", "admiral", "an", "and", "answer", "as", "class", "captain", "chief", "cmdr", "colonel", "commander", "crewman", "doctor", "dr", "ensign", "episode", "film", "from", "general", "grand", "gul", "i", "in", "iss", "is", "it", "lt", "lieutenant", "ltcmdr", "major", "movie", "mr", "mrs", "ms", "my", "nagus", "ncc", "number", "nx", "of", "on", "part", "professor", "series", "show", "species", "star", "starship", "st", "subcommander", "the", "think", "trek", "uss", "was", } NUMBER_WORDS = { "zero": "0", "one": "1", "two": "2", "three": "3", "four": "4", "five": "5", "six": "6", "seven": "7", "eight": "8", "nine": "9", "ten": "10", } NUMBER_PHRASES = { "thirty one": "31", } SERIES_ALIASES = { "tos": "original", "st tos": "original", "star trek tos": "original", "tng": "next generation", "st tng": "next generation", "star trek tng": "next generation", "ds9": "deep space 9", "dsn": "deep space 9", "st ds9": "deep space 9", "star trek ds9": "deep space 9", "voy": "voyager", "st voy": "voyager", "star trek voy": "voyager", "ent": "enterprise", "st ent": "enterprise", "star trek ent": "enterprise", "dsc": "discovery", "dis": "discovery", "disco": "discovery", "st dsc": "discovery", "star trek dsc": "discovery", "star trek disco": "discovery", "snw": "strange new worlds", "st snw": "strange new worlds", "star trek snw": "strange new worlds", "ld": "lower decks", "lds": "lower decks", "st ld": "lower decks", "st lds": "lower decks", "star trek ld": "lower decks", "star trek lds": "lower decks", } @dataclass(frozen=True) class Question: difficulty: str prompt: str answers: tuple[str, ...] fixed: bool = False @dataclass class Session: session_id: str questions: list[Question] encoded_prompts: list[str] index: int created_at: float last_seen_at: float client_key: str | None = None def normalize_answer(value: str) -> str: """Normalize an answer while keeping enough structure to avoid loose matches.""" value = unicodedata.normalize("NFKD", value) value = value.encode("ascii", "ignore").decode("ascii") value = value.lower().replace("&", " and ") value = re.sub(r"[^a-z0-9]+", " ", value) value = " ".join(value.split()) for phrase, replacement in NUMBER_PHRASES.items(): value = re.sub(rf"\b{re.escape(phrase)}\b", replacement, value) tokens = [NUMBER_WORDS.get(token, token) for token in value.split()] value = " ".join(tokens) return SERIES_ALIASES.get(value, value) def compact(value: str) -> str: return value.replace(" ", "") def core_tokens(value: str) -> list[str]: return [token for token in normalize_answer(value).split() if token not in FILLER_WORDS] def answer_matches(response: str, question: Question) -> bool: response_normalized = normalize_answer(response) response_compact = compact(response_normalized) response_core = core_tokens(response) if not response_normalized: return False for accepted in question.answers: accepted_normalized = normalize_answer(accepted) accepted_compact = compact(accepted_normalized) accepted_core = core_tokens(accepted) if response_normalized == accepted_normalized: return True if response_compact == accepted_compact: return True if response_core and accepted_core and response_core == accepted_core: return True if response_core and accepted_core and sorted(response_core) == sorted(accepted_core) and len(accepted_core) > 1: return True # A typo-tolerant path for longer answers. Short answers such as "Data" # and "Odo" stay exact so near-misses do not become free guesses. if len(accepted_compact) >= 7 and len(response_compact) >= 7: if SequenceMatcher(None, response_compact, accepted_compact).ratio() >= 0.90: return True return False def load_questions(path: Path) -> list[Question]: questions: list[Question] = [] use_markdown = path.suffix.lower() in {".md", ".markdown"} with path.open("r", encoding="utf-8") as handle: for line_number, raw_line in enumerate(handle, start=1): line = raw_line.strip() if not line or line.startswith("#"): continue if use_markdown or line.startswith("|"): parsed = parse_markdown_question_row(path, line_number, line) if parsed is None: continue else: parsed = parse_tsv_question_row(path, line_number, line) questions.append(parsed) if not questions: raise ValueError(f"{path}: no questions loaded") return questions def parse_tsv_question_row(path: Path, line_number: int, line: str) -> Question: parts = line.split("\t") if len(parts) not in {3, 4}: raise ValueError( f"{path}:{line_number}: expected 3 or 4 tab-separated columns: " "difficulty, question, answers, optional fixed" ) difficulty, prompt, answer_blob = (part.strip() for part in parts[:3]) fixed = parse_fixed_marker(path, line_number, parts[3].strip()) if len(parts) == 4 else False return make_question(path, line_number, difficulty, prompt, answer_blob.split("|"), fixed) def parse_markdown_question_row(path: Path, line_number: int, line: str) -> Question | None: if not line.startswith("|"): return None cells = split_markdown_table_row(line) if is_markdown_separator(cells): return None normalized_header = tuple(cell.lower() for cell in cells) if normalized_header in { ("difficulty", "question", "answers"), ("difficulty", "fixed", "question", "answers"), }: return None if len(cells) not in {3, 4}: raise ValueError( f"{path}:{line_number}: expected a 3- or 4-column Markdown table row: " "Difficulty, optional Fixed, Question, Answers" ) if len(cells) == 3: difficulty, prompt, answer_blob = cells fixed = False else: difficulty, fixed_blob, prompt, answer_blob = cells fixed = parse_fixed_marker(path, line_number, fixed_blob) return make_question(path, line_number, difficulty, prompt, answer_blob.split(";"), fixed) def split_markdown_table_row(line: str) -> list[str]: stripped = line.strip() if not stripped.startswith("|"): return [] stripped = stripped[1:] if stripped.endswith("|"): stripped = stripped[:-1] cells: list[str] = [] current: list[str] = [] escaped = False for char in stripped: if escaped: current.append(char) escaped = False elif char == "\\": escaped = True elif char == "|": cells.append("".join(current).strip()) current = [] else: current.append(char) if escaped: current.append("\\") cells.append("".join(current).strip()) return cells def is_markdown_separator(cells: list[str]) -> bool: if not cells: return False return all(re.fullmatch(r":?-{3,}:?", cell.replace(" ", "")) for cell in cells) def make_question( path: Path, line_number: int, difficulty: str, prompt: str, answer_parts: list[str], fixed: bool = False, ) -> Question: answers = tuple(answer.strip() for answer in answer_parts if answer.strip()) if not difficulty or not prompt or not answers: raise ValueError(f"{path}:{line_number}: difficulty, question, and answers are required") return Question(difficulty=difficulty, prompt=prompt, answers=answers, fixed=fixed) def parse_fixed_marker(path: Path, line_number: int, value: str) -> bool: normalized = value.strip().lower() if normalized in {"", "no", "n", "false", "0", "-"}: return False if normalized in {"yes", "y", "true", "1", "fixed", "deterministic", "always"}: return True raise ValueError(f"{path}:{line_number}: invalid fixed marker: {value!r}") def encode_question_prompt( prompt: str, rng: random.Random | random.SystemRandom, steps: int = QUESTION_ENCODING_STEPS, ) -> str: payload = prompt.encode("utf-8") for step_index in range(steps): encoding = "b64" if step_index == steps - 1 else rng.choice(QUESTION_ENCODINGS) if encoding == "b64": payload = base64.b64encode(payload) elif encoding == "b85": payload = base64.b85encode(payload) else: raise ValueError(f"unsupported question encoding: {encoding}") return payload.decode("ascii") class QuestGame: def __init__( self, questions: list[Question], quest_length: int, flag: str, session_ttl_seconds: int = DEFAULT_SESSION_TTL_SECONDS, session_start_limit: int = DEFAULT_SESSION_START_LIMIT, session_start_window_seconds: int = DEFAULT_SESSION_START_WINDOW_SECONDS, rng: random.Random | random.SystemRandom | None = None, clock: Callable[[], float] = time.time, ) -> None: if quest_length < 1: raise ValueError("quest length must be at least 1") if session_start_limit < 0: raise ValueError("session start limit cannot be negative") if session_start_window_seconds < 0: raise ValueError("session start window cannot be negative") self.fixed_questions = [question for question in questions if question.fixed] self.random_questions = [question for question in questions if not question.fixed] if self.fixed_questions and len(self.fixed_questions) != 2: raise ValueError("question pool must mark exactly 2 fixed questions when fixed questions are used") if self.fixed_questions and quest_length < 2: raise ValueError("quest length must be at least 2 when fixed questions are used") random_slots = quest_length - len(self.fixed_questions) if random_slots < 0: raise ValueError("quest length cannot be smaller than the fixed question count") if random_slots > len(self.random_questions): raise ValueError( f"quest length ({quest_length}) needs {random_slots} random questions, " f"but only {len(self.random_questions)} are available" ) self.questions = questions self.quest_length = quest_length self.flag = flag self.session_ttl_seconds = session_ttl_seconds self.session_start_limit = session_start_limit self.session_start_window_seconds = session_start_window_seconds self.rng = rng or random.SystemRandom() self.clock = clock self.sessions: dict[str, Session] = {} self.session_starts_by_client: dict[str, list[float]] = {} def handle( self, params: Mapping[str, list[str]], client_key: str | None = None, ) -> tuple[int, str]: self.cleanup_expired_sessions() session_id = first_param(params, "sid", "session", "session_id") answer = first_param(params, "answer", "a", "response") if not session_id: return self.handle_session_start(client_key) if session_id not in self.sessions: self.reset_sessions() return 410, goto_begin("Invalid or expired session. State reset.") session = self.sessions[session_id] session.last_seen_at = self.clock() if answer is None: return 200, self.render_current_question(session, "Repeat current question.") if not answer_matches(answer, session.questions[session.index]): self.delete_session(session_id) return 403, goto_begin("Wrong answer. Session invalidated.") session.index += 1 if session.index >= len(session.questions): self.delete_session(session_id) return 200, "\n".join( [ "Correct.", "Quest complete.", f"FLAG: {self.flag}", "", ] ) return 200, self.render_current_question(session, "Correct.") def handle_session_start(self, client_key: str | None) -> tuple[int, str]: if client_key: allowed, retry_after = self.record_session_start(client_key) if not allowed: self.reset_sessions() return 429, slow_down(retry_after) self.reset_sessions() return 200, self.start_session(client_key) def start_session(self, client_key: str | None = None) -> str: session_id = secrets.token_urlsafe(18) selected = self.select_session_questions() encoded_prompts = [encode_question_prompt(question.prompt, self.rng) for question in selected] now = self.clock() self.sessions[session_id] = Session( session_id=session_id, questions=selected, encoded_prompts=encoded_prompts, index=0, created_at=now, last_seen_at=now, client_key=client_key, ) return self.render_current_question(self.sessions[session_id], include_rules=True) def record_session_start(self, client_key: str) -> tuple[bool, int]: if self.session_start_limit <= 0 or self.session_start_window_seconds <= 0: return True, 0 now = self.clock() cutoff = now - self.session_start_window_seconds starts = [ started_at for started_at in self.session_starts_by_client.get(client_key, []) if started_at > cutoff ] if len(starts) >= self.session_start_limit: self.session_starts_by_client[client_key] = starts retry_after = max(1, int(starts[0] + self.session_start_window_seconds - now)) return False, retry_after starts.append(now) self.session_starts_by_client[client_key] = starts return True, 0 def delete_session(self, session_id: str) -> None: self.sessions.pop(session_id, None) def reset_sessions(self) -> None: self.sessions.clear() def select_session_questions(self) -> list[Question]: if not self.fixed_questions: return self.rng.sample(self.questions, self.quest_length) random_count = self.quest_length - len(self.fixed_questions) return self.rng.sample(self.random_questions, random_count) + self.fixed_questions def render_current_question( self, session: Session, status: str | None = None, include_rules: bool = False, ) -> str: question = session.questions[session.index] encoded_prompt = session.encoded_prompts[session.index] lines = [] if include_rules: lines.extend( [ "STAR TREK QUEST FOR VIGENERE KEY", "", "Rules:", "- Use HTTP GET requests only.", f"- Endpoint: {QUEST_PATH}", f"- Reply with: {QUEST_PATH}?sid=&answer=", "- All questions must be answered correctly.", "- Question text uses b64/b85 nested encoding.", f"- Decode exactly {QUESTION_ENCODING_STEPS} steps. The outer layer is b64; the inner layers are randomized.", "- Send answers in plaintext, not encoded.", "- Any request without a valid session ID resets the current quest.", "- Be warned: repeated bare /q requests from the same client may return HTTP 429.", "- Do not try to write a catalog (we have more than 10_000 questions ;-)).", "- One mistake invalidates the session: GOTO BEGIN.", "- Timeout is around 20 sec / question. Exceptions: unlimited for the first question / only 5 sec for the last question.", "- curl please, no python requests.", "", f"Session ID: {session.session_id}", "", ] ) if status: lines.extend([status, ""]) elif status: lines.extend([status, ""]) lines.extend( [ f"Progress: {session.index + 1}/{len(session.questions)}", f"Difficulty: {question.difficulty}", f"Question: {encoded_prompt}", "", ] ) return "\n".join(lines) def cleanup_expired_sessions(self) -> None: if self.session_ttl_seconds > 0: deadline = self.clock() - self.session_ttl_seconds expired = [ session_id for session_id, session in self.sessions.items() if session.last_seen_at < deadline ] for session_id in expired: self.delete_session(session_id) self.cleanup_session_start_history() def cleanup_session_start_history(self) -> None: if self.session_start_window_seconds <= 0: self.session_starts_by_client.clear() return cutoff = self.clock() - self.session_start_window_seconds for client_key in list(self.session_starts_by_client): starts = [ started_at for started_at in self.session_starts_by_client[client_key] if started_at > cutoff ] if starts: self.session_starts_by_client[client_key] = starts else: del self.session_starts_by_client[client_key] def first_param(params: Mapping[str, list[str]], *names: str) -> str | None: for name in names: values = params.get(name) if values: return values[0].strip() return None def goto_begin(reason: str) -> str: return "\n".join( [ "GOTO BEGIN", reason, f"Start a new quest with GET {QUEST_PATH}.", "", ] ) def slow_down(retry_after_seconds: int) -> str: return "\n".join( [ "SLOW DOWN", "Too many new sessions from this client.", "Continue your active session if you have one.", f"Try a new session again in {retry_after_seconds} seconds.", "", ] ) def validate_user_agent_prefix(prefix: str) -> str: if len(prefix) != USER_AGENT_PREFIX_LENGTH: raise ValueError( f"user-agent prefix must be exactly {USER_AGENT_PREFIX_LENGTH} characters" ) return prefix def user_agent_allowed(user_agent: str | None, required_prefix: str) -> bool: return (user_agent or "")[:USER_AGENT_PREFIX_LENGTH] == required_prefix def route_quest_request( user_agent: str | None, params: Mapping[str, list[str]], game: QuestGame, required_user_agent_prefix: str, static_quest_body: str, client_key: str | None = None, ) -> tuple[int, str]: if not user_agent_allowed(user_agent, required_user_agent_prefix): return DECOY_STATUS_CODE, static_quest_body return game.handle(params, client_key=client_key) def route_static_file_request( user_agent: str | None, request_path: str, required_user_agent_prefix: str, static_quest_body: str, file_root: Path, ) -> tuple[int, bytes, str]: if not user_agent_allowed(user_agent, required_user_agent_prefix): return DECOY_STATUS_CODE, static_quest_body.encode("utf-8"), "text/plain; charset=utf-8" requested_file = resolve_static_file(file_root, request_path) if requested_file is None: return 404, b"File not found.\n", "text/plain; charset=utf-8" if requested_file == "forbidden": return 403, b"Forbidden.\n", "text/plain; charset=utf-8" return 200, requested_file.read_bytes(), guess_content_type(requested_file) def resolve_static_file(file_root: Path, request_path: str) -> Path | str | None: relative_path = unquote(request_path).lstrip("/") if not relative_path: return None root = file_root.resolve() requested = (root / relative_path).resolve() try: requested.relative_to(root) except ValueError: return "forbidden" if not requested.is_file(): return None return requested def guess_content_type(path: Path) -> str: content_type, _ = mimetypes.guess_type(str(path)) if content_type is None: return "application/octet-stream" if content_type.startswith("text/") or content_type in { "application/javascript", "application/json", "application/xml", }: return f"{content_type}; charset=utf-8" return content_type def load_static_quest(path: Path) -> str: return path.read_text(encoding="utf-8") def make_handler( game: QuestGame, required_user_agent_prefix: str, static_quest_body: str, file_root: Path, ) -> type[BaseHTTPRequestHandler]: class QuestRequestHandler(BaseHTTPRequestHandler): server_version = "StarTrekQuest/1.0" def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path != QUEST_PATH: status_code, body, content_type = route_static_file_request( self.headers.get("User-Agent"), parsed.path, required_user_agent_prefix, static_quest_body, file_root, ) self.send_bytes(status_code, body, content_type) return params = parse_qs(parsed.query, keep_blank_values=True) status_code, body = route_quest_request( self.headers.get("User-Agent"), params, game, required_user_agent_prefix, static_quest_body, client_key=self.client_address[0], ) self.send_text(status_code, body) def do_POST(self) -> None: self.send_text(405, f"GET only. Use {QUEST_PATH} with query parameters.\n") def send_text(self, status_code: int, body: str) -> None: self.send_bytes(status_code, body.encode("utf-8"), "text/plain; charset=utf-8") def send_bytes(self, status_code: int, body: bytes, content_type: str) -> None: self.send_response(status_code) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def version_string(self) -> str: return self.server_version def log_message(self, fmt: str, *args: object) -> None: sys.stderr.write("%s - - [%s] %s\n" % (self.client_address[0], self.log_date_time_string(), fmt % args)) return QuestRequestHandler def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the Star Trek GET-only quest server.") parser.add_argument("--host", default=os.getenv("QUEST_HOST", "0.0.0.0")) parser.add_argument("--port", type=int, default=int(os.getenv("QUEST_PORT", "443"))) parser.add_argument( "--questions", type=Path, default=Path(os.getenv("QUEST_QUESTIONS", DEFAULT_QUESTIONS_FILE)), ) parser.add_argument( "--static-quest", type=Path, default=Path(os.getenv("QUEST_STATIC_QUEST", DEFAULT_STATIC_QUEST_FILE)), help="Static text served with HTTP 266 when the User-Agent gate fails.", ) parser.add_argument( "--file-root", type=Path, default=Path(os.getenv("QUEST_FILE_ROOT", DEFAULT_FILE_ROOT)), help="Root directory for gated static file serving.", ) parser.add_argument( "--user-agent-prefix", default=os.getenv("QUEST_USER_AGENT_PREFIX", DEFAULT_USER_AGENT_PREFIX), help="Required first 4 User-Agent characters for the real quest.", ) parser.add_argument( "--length", type=int, default=int(os.getenv("QUEST_LENGTH", str(DEFAULT_QUEST_LENGTH))), help="Number of questions per session.", ) parser.add_argument("--flag", default=os.getenv("FLAG", DEFAULT_FLAG)) parser.add_argument( "--session-ttl", type=int, default=int(os.getenv("SESSION_TTL_SECONDS", str(DEFAULT_SESSION_TTL_SECONDS))), help="Session TTL in seconds. Use 0 to disable expiry.", ) parser.add_argument( "--session-start-limit", type=int, default=int(os.getenv("SESSION_START_LIMIT", str(DEFAULT_SESSION_START_LIMIT))), help="Maximum new sessions per client inside the start window. Use 0 to disable.", ) parser.add_argument( "--session-start-window", type=int, default=int(os.getenv("SESSION_START_WINDOW_SECONDS", str(DEFAULT_SESSION_START_WINDOW_SECONDS))), help="Sliding window in seconds for new-session throttling. Use 0 to disable.", ) parser.add_argument( "--certfile", type=Path, default=env_path("QUEST_CERTFILE"), help="TLS certificate file. Requires --keyfile.", ) parser.add_argument( "--keyfile", type=Path, default=env_path("QUEST_KEYFILE"), help="TLS private key file. Requires --certfile.", ) return parser.parse_args() def env_path(name: str) -> Path | None: value = os.getenv(name) return Path(value) if value else None def build_tls_context(certfile: Path | None, keyfile: Path | None) -> ssl.SSLContext | None: if bool(certfile) != bool(keyfile): raise ValueError("--certfile and --keyfile must be provided together") if certfile is None or keyfile is None: return None context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain(certfile=str(certfile), keyfile=str(keyfile)) return context def main() -> int: args = parse_args() try: required_user_agent_prefix = validate_user_agent_prefix(args.user_agent_prefix) static_quest_body = load_static_quest(args.static_quest) questions = load_questions(args.questions) file_root = args.file_root.resolve() if not file_root.is_dir(): raise ValueError(f"file root does not exist or is not a directory: {file_root}") except (OSError, ValueError) as error: print(f"error: {error}", file=sys.stderr) return 2 game = QuestGame( questions=questions, quest_length=args.length, flag=args.flag, session_ttl_seconds=args.session_ttl, session_start_limit=args.session_start_limit, session_start_window_seconds=args.session_start_window, ) try: tls_context = build_tls_context(args.certfile, args.keyfile) except ValueError as error: print(f"error: {error}", file=sys.stderr) return 2 server = ThreadingHTTPServer( (args.host, args.port), make_handler(game, required_user_agent_prefix, static_quest_body, file_root), ) scheme = "http" if tls_context is not None: server.socket = tls_context.wrap_socket(server.socket, server_side=True) scheme = "https" print(f"Serving Star Trek quest on {scheme}://{args.host}:{args.port}{QUEST_PATH}") print(f"Serving gated files from {file_root}") print(f"Loaded {len(questions)} questions; using {args.length} per session.") print(f"Real quest User-Agent prefix: {required_user_agent_prefix!r}") print( "New-session guard: " f"{args.session_start_limit} starts per {args.session_start_window} seconds per client." ) try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down.") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())