#!/usr/bin/env python3 """ Gemastik A/D multi-team orchestrator. Each team gets an isolated stack: its own docker-compose (unique ports + SSH passwords), its own receiver (unique admin creds + ports), and its own flags. The orchestrator clones the base services dir, rewrites ports and passwords, and manages lifecycle via docker compose project per team. Team N layout: /opt/gemastik18-final/teams/team/ services/ (docker-compose.yml with team ports + passwords) receiver/ (.env with team admin creds + container overrides) receiver/flags/ (per-team flag files) state.json (team metadata: ports, admin user/pass, ssh creds, created ts) Port scheme (base offset per team index, index 1-based): team i: chall ports = 30000 + i*1000 + 0..5 (blogpost,carbeat,cdn,phew,sheesh,warmup) ssh ports = 30000 + i*1000 + 22..27 (10022-style) receiver = 30000 + i*1000 + 80 (receiver API, e.g. 31080, 32080) """ import json import os import re import secrets import shutil import subprocess import sys import time import uuid from datetime import datetime from pathlib import Path BASE = Path("/opt/gemastik18-final") TEAMS_DIR = BASE / "teams" SERVICES_SRC = BASE / "services" RECEIVER_SRC = BASE / "receiver" # Challenge definitions: (service name, chall port offset 0-5, ssh offset 22-27) CHALLENGES = [ ("blogpost", 0, 22), ("carbeat", 1, 23), ("cdn", 2, 24), ("phew", 3, 25), ("sheesh", 4, 26), ("warmup", 5, 27), ] PORT_BASE = 30000 STEP = 1000 def team_ports(idx: int) -> dict: """Return {service_name: {'chall': port, 'ssh': port}} for 1-based team idx.""" base = PORT_BASE + idx * STEP out = {} for name, coff, soff in CHALLENGES: out[name] = {"chall": base + coff, "ssh": base + soff} out["receiver"] = base + 80 out["panel"] = base + 81 # reserved, not used return out def gen_password(n=16): return uuid.uuid4().hex[:n] def slugify(s: str) -> str: """'Tim Satu Beta!' -> 'tim-satu-beta'""" s = re.sub(r"[^a-zA-Z0-9\s-]", "", s.lower()).strip() s = re.sub(r"[\s_]+", "-", s) return s or "team" def create_team(idx: int, label: str = None, domain: str = None): """Build a full team stack dir with unique ports/passwords. label -> team display name (leaderboard/topology) domain -> custom subdomain host, e.g. "cyber-warriors" or "cyber-warriors.gemastik.imrnes.team" (full host accepted). Defaults to slugify(label).gemastik.imrnes.team. """ ports = team_ports(idx) team_dir = TEAMS_DIR / f"team{idx}" label = label or f"Tim {idx}" slug = slugify(label) # normalize custom domain -> host under *.gemastik.imrnes.team host = (domain or f"{slug}.gemastik.imrnes.team").strip().lower() host = host.replace("https://", "").replace("http://", "").rstrip("/") if not host.endswith(".gemastik.imrnes.team"): host = f"{host}.gemastik.imrnes.team" slug = host.split(".")[0] state = { "index": idx, "label": label, "slug": slug, "domain": host, "ports": ports, "admin_user": f"admin_team{idx}", "admin_pass": gen_password(20), "ssh_user": "ctfuser", "ssh_pass": gen_password(20), "chall_passwords": {name: gen_password(20) for name, *_ in CHALLENGES}, "container_suffix": f"team{idx}", "created": __import__("datetime").datetime.now().isoformat(), "status": "created", } # --- copy services with rewrite --- svc_dir = team_dir / "services" if svc_dir.exists(): shutil.rmtree(svc_dir) shutil.copytree(SERVICES_SRC, svc_dir, ignore=shutil.ignore_patterns("__pycache__", ".git", "exploits", "exploit")) # compose references ../utils/bashrc etc — copy utils next to services utils_src = BASE / "utils" utils_dst = team_dir / "utils" if utils_src.exists(): if utils_dst.exists(): shutil.rmtree(utils_dst) shutil.copytree(utils_src, utils_dst) # redirect preexec URL to the team's receiver port recv_port = ports["receiver"] bashrc = utils_dst / "bashrc" if bashrc.exists(): bashrc.write_text(bashrc.read_text().replace( "host.docker.internal:18080", f"host.docker.internal:{recv_port}")) compose = svc_dir / "docker-compose.yml" text = compose.read_text() # --- replace build: blocks with image: so teams reuse the base images (no rebuild) --- # Each service's build block looks like: # build: # context: # args: # - PASSWORD=$PASSWORD_XXXXX # Replace the whole block with " image: services-". for name, coff, soff in CHALLENGES: text = re.sub( rf" build:\n context: {name}\n args:\n - PASSWORD=\$PASSWORD_[0-9]+\n", f" image: services-{name}\n", text) compose.write_text(text) # rewrite container names + ports per challenge for name, coff, soff in CHALLENGES: cont_old = f"{name}_container" cont_new = f"{name}_container_team{idx}" text = text.replace(f"container_name: {cont_old}", f"container_name: {cont_new}") text = text.replace(f"hostname: {name}", f"hostname: {name}_team{idx}") # ports mapping: "10000:8000" -> ":8000" old_chall = str(10000 + coff * 1000) # 10000,11000,12000,13000,14000,15000 old_ssh = str(10022 + coff * 1000) # 10022,11022,... text = re.sub(rf'"({old_chall}):', f'"{ports[name]["chall"]}:', text) text = re.sub(rf'"({old_ssh}):', f'"{ports[name]["ssh"]}:', text) # PASSWORD_* args -> team passwords for name, coff, soff in CHALLENGES: old_env = f"PASSWORD_{10000 + coff * 1000}" text = re.sub(rf"\${{{old_env}}}", state["chall_passwords"][name], text) # compose file references services/.env — we'll create it below compose.write_text(text) # team services/.env (PASSWORD_* in same shape as starter.py) env_lines = [f"ADMIN_USERNAME={state['admin_user']}", f"ADMIN_PASSWORD={state['admin_pass']}"] env_lines.append(f"COMPOSE_LOCATION={svc_dir}/docker-compose.yml") for i in range(20): env_lines.append(f"PASSWORD_{(i*1000)+10000}={gen_password(20)}") # force the six used passwords to team ones for name, coff, soff in CHALLENGES: for j, line in enumerate(env_lines): if line.startswith(f"PASSWORD_{10000+coff*1000}="): env_lines[j] = f"PASSWORD_{10000+coff*1000}={state['chall_passwords'][name]}" (svc_dir / ".env").write_text("\n".join(env_lines) + "\n") # --- copy receiver with team env --- recv_dir = team_dir / "receiver" if recv_dir.exists(): shutil.rmtree(recv_dir) shutil.copytree(RECEIVER_SRC, recv_dir, ignore=shutil.ignore_patterns("__pycache__", ".venv", ".env", "history")) (recv_dir / "history").mkdir(exist_ok=True) # receiver .env: same admin + passwords + container overrides recv_env = [f"ADMIN_USERNAME={state['admin_user']}", f"ADMIN_PASSWORD={state['admin_pass']}", f"COMPOSE_LOCATION={svc_dir}/docker-compose.yml"] for i in range(20): recv_env.append(f"PASSWORD_{(i*1000)+10000}={gen_password(20)}") for name, coff, soff in CHALLENGES: for j, line in enumerate(recv_env): if line.startswith(f"PASSWORD_{10000+coff*1000}="): recv_env[j] = f"PASSWORD_{10000+coff*1000}={state['chall_passwords'][name]}" recv_env.append(f"CHALLENGE_PORT_{name.upper()}={ports[name]['chall']}") recv_env.append(f"CHALLENGE_CONTAINER_{name.upper()}={name}_container_team{idx}") (recv_dir / ".env").write_text("\n".join(recv_env) + "\n") # --- flags (randomized per team so each team has unique flags) --- flags_dir = team_dir / "receiver" / "flags" flags_dir.mkdir(parents=True, exist_ok=True) flags_map = {} for name, coff, soff in CHALLENGES: flag = f"GEMASTIK18{{TEAM{idx}_{name.upper()}_{secrets.token_hex(6)}}}" (flags_dir / f"{name}.txt").write_text(flag) flags_map[name] = flag state["flags"] = flags_map (team_dir / "state.json").write_text(json.dumps(state, indent=2)) return state def update_team(idx: int, label: str = None, domain: str = None) -> dict: """Update a team's display name and/or custom domain (live re-route). - label updates state.json['label'] (leaderboard/topology use this). - domain updates slug + domain and rewrites the Traefik team route. Returns updated state. """ team_dir = TEAMS_DIR / f"team{idx}" if not (team_dir / "state.json").exists(): raise FileNotFoundError(f"Team {idx} not created") st = json.loads((team_dir / "state.json").read_text()) if label: st["label"] = str(label).strip()[:48] or st["label"] if domain: host = domain.strip().lower().replace("https://", "").replace("http://", "").rstrip("/") if not host.endswith(".gemastik.imrnes.team"): host = f"{host}.gemastik.imrnes.team" st["domain"] = host st["slug"] = host.split(".")[0] (team_dir / "state.json").write_text(json.dumps(st, indent=2)) # re-route domain in Traefik immediately ensure_team_domains() return st def start_team(idx: int): team_dir = TEAMS_DIR / f"team{idx}" if not (team_dir / "state.json").exists(): raise FileNotFoundError(f"Team {idx} not created") svc_dir = team_dir / "services" subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "up", "-d", "--build"], cwd=str(svc_dir), check=False, capture_output=True) _start_receiver(idx) st = json.loads((team_dir / "state.json").read_text()) st["status"] = "running" (team_dir / "state.json").write_text(json.dumps(st, indent=2)) # Set per-team SSH passwords at runtime (images are shared across teams, # so chpasswd ensures each team's containers have the team's own password). set_ssh_passwords(idx) return st def set_ssh_passwords(idx: int): """Set SSH password for ctfuser in each challenge container of a team.""" team_dir = TEAMS_DIR / f"team{idx}" st = json.loads((team_dir / "state.json").read_text()) for name, coff, soff in CHALLENGES: cont = f"{name}_container_team{idx}" pw = st["chall_passwords"][name] cmd = f"echo 'ctfuser:{pw}' | chpasswd" # retry a few times — right after `compose up`, container may still be booting ok = False for attempt in range(5): r = subprocess.run(["docker", "exec", cont, "sh", "-c", cmd], capture_output=True, text=True, timeout=30) if r.returncode == 0: ok = True break time.sleep(3) if not ok: print(f"[set_ssh_passwords] {cont}: FAILED after retries ({r.stderr.strip()[:100]})") else: print(f"[set_ssh_passwords] {cont}: OK") def stop_team(idx: int): team_dir = TEAMS_DIR / f"team{idx}" svc_dir = team_dir / "services" subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "down"], cwd=str(svc_dir), check=False, capture_output=True) _stop_receiver(idx) st = json.loads((team_dir / "state.json").read_text()) st["status"] = "stopped" (team_dir / "state.json").write_text(json.dumps(st, indent=2)) return st def _start_receiver(idx: int): """Start team's receiver via systemd service (independent of panel cgroup).""" team_dir = TEAMS_DIR / f"team{idx}" st = json.loads((team_dir / "state.json").read_text()) port = st["ports"]["receiver"] pidfile = team_dir / "receiver.pid" # ensure the per-team systemd unit exists with current env subprocess.run([sys.executable, str(BASE / "panel" / "gen_receiver_services.py"), "start"], check=False, capture_output=True) subprocess.run(["systemctl", "restart", f"gemastik-receiver-team{idx}.service"], check=False, capture_output=True) # best-effort pid bookkeeping for ps try: out = subprocess.run(["systemctl", "show", "-p", "MainPID", f"gemastik-receiver-team{idx}.service"], capture_output=True, text=True).stdout pid = out.strip().split("=")[1] pidfile.write_text(pid) except Exception: pass def _stop_receiver(idx: int): """Stop team's receiver via systemd service.""" team_dir = TEAMS_DIR / f"team{idx}" pidfile = team_dir / "receiver.pid" subprocess.run(["systemctl", "stop", f"gemastik-receiver-team{idx}.service"], check=False, capture_output=True) pidfile.unlink(missing_ok=True) def team_logs(idx: int, service: str = None, tail: int = 100): team_dir = TEAMS_DIR / f"team{idx}" svc_dir = team_dir / "services" data = {} services = [s for s, *_ in CHALLENGES] if service is None else [service] for s in services: try: out = subprocess.run( ["docker", "logs", "--tail", str(tail), f"{s}_container_team{idx}"], capture_output=True, text=True, timeout=15) data[s] = (out.stdout + out.stderr)[-4000:] except Exception as e: data[s] = f"ERR: {e}" return data def ensure_team_domains() -> str: """Write Traefik dynamic config for each team domain -> panel (:18081). Each team gets .gemastik.imrnes.team. The panel routes /team/ to that team's portal page (public, no panitia login), and /api/team//* serves team-scoped API. Writing this into the same dynamic dir Traefik watches means domains appear automatically. Returns a status string for logging. """ traefik_dir = Path("/data/coolify/proxy/dynamic") traefik_dir.mkdir(parents=True, exist_ok=True) teams = list_teams() out = [] changed = False for t in teams: slug = t.get("slug") or slugify(t.get("label", "")) if not slug: continue # backfill slug/domain for teams created before this feature if not t.get("slug"): td = TEAMS_DIR / f"team{t['index']}" st = json.loads((td / "state.json").read_text()) st["slug"] = slug st["domain"] = f"{slug}.gemastik.imrnes.team" (td / "state.json").write_text(json.dumps(st, indent=2)) changed = True host = f"{slug}.gemastik.imrnes.team" out.append(f""" team-{slug}-http: rule: Host(`{host}`) entryPoints: - http service: gemastik-panel-service middlewares: - redirect-to-https team-{slug}-https: rule: Host(`{host}`) entryPoints: - https service: gemastik-panel-service tls: certResolver: letsencrypt domains: - main: {host} """) if not out: return "no teams to route" # Keep the team domain block in its own file so the main gemastik.yaml stays untouched (traefik_dir / "gemastik-teams.yaml").write_text("http:\n routers:\n" + "".join(out)) return f"wrote {len(out)} team domain(s) in gemastik-teams.yaml" def list_teams() -> list: out = [] if not TEAMS_DIR.exists(): return out for d in sorted(TEAMS_DIR.glob("team*")): if (d / "state.json").exists(): st = json.loads((d / "state.json").read_text()) # compute alive status quickly st["alive_count"] = 0 st["up_count"] = 0 out.append(st) return out # --------------------------------------------------------------------------- # Flag randomization + team submission tracking # --------------------------------------------------------------------------- def randomize_flags(idx: int) -> dict: """Generate fresh unique flags for every challenge of a team.""" team_dir = TEAMS_DIR / f"team{idx}" if not (team_dir / "state.json").exists(): raise FileNotFoundError(f"Team {idx} not created") flags_dir = team_dir / "receiver" / "flags" flags_dir.mkdir(parents=True, exist_ok=True) mapping = {} for name, coff, soff in CHALLENGES: token = secrets.token_hex(6) flag = f"GEMASTIK18{{TEAM{idx}_{name.upper()}_{token}}}" (flags_dir / f"{name}.txt").write_text(flag) mapping[name] = flag # rotate into containers: compose mounts the flag files read-only, so # drop+recreate the challenge containers to pick up new flags svc_dir = team_dir / "services" subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "down"], cwd=str(svc_dir), check=False, capture_output=True) subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "up", "-d"], cwd=str(svc_dir), check=False, capture_output=True) _start_receiver(idx) st = json.loads((team_dir / "state.json").read_text()) st["flags"] = mapping (team_dir / "state.json").write_text(json.dumps(st, indent=2)) return mapping def submit_flag(target_idx: int, chall: str, flag: str, attacker_idx: int = None, attacker_name: str = "") -> dict: """Validate a submitted flag against the TARGET team's challenge flag. A/D semantics: attacker_idx scores by stealing target_idx's flag. Back-compat: attacker_idx defaults to target_idx (self-submit). """ attacker_idx = attacker_idx if attacker_idx else target_idx team_dir = TEAMS_DIR / f"team{target_idx}" if not (team_dir / "state.json").exists(): return {"success": False, "error": "unknown team"} flags_dir = team_dir / "receiver" / "flags" expected = (flags_dir / f"{chall}.txt").read_text().strip() if (flags_dir / f"{chall}.txt").exists() else None if not expected: return {"success": False, "error": "challenge not found"} if flag.strip() != expected: _log_attack(attacker_idx, target_idx, chall, flag[:24], success=False) return {"success": False, "error": "wrong flag"} _log_attack(attacker_idx, target_idx, chall, flag, success=True) # record leaderboard entry — score goes to the ATTACKER lb_path = TEAMS_DIR / "leaderboard.json" lb = json.loads(lb_path.read_text()) if lb_path.exists() else {"solves": []} entry = { "team": attacker_idx, "team_name": attacker_name or f"Team {attacker_idx}", "challenge": chall, "target": target_idx, "flag": flag, "ts": time.time(), "ts_human": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } # avoid double-solve duplicates (same flag + same attacker team) if not any(e["team"] == attacker_idx and e["challenge"] == chall and e.get("target") == target_idx for e in lb["solves"]): lb["solves"].append(entry) lb_path.write_text(json.dumps(lb, indent=2)) return {"success": True, "team": attacker_idx, "target": target_idx, "challenge": chall} def _log_attack(attacker_team: int, target_team: int, target_chall: str, flag_hint: str, success: bool): """Append a row to the attack log (used by the topology attack visualizer).""" try: ap = TEAMS_DIR / "attacks.json" log = json.loads(ap.read_text()) if ap.exists() else {"events": []} log["events"].append({ "attacker": attacker_team, "target": target_team, "challenge": target_chall, "flag_hint": flag_hint, "success": success, "ts": time.time(), "ts_human": datetime.now().strftime("%H:%M:%S"), }) # keep last 200 log["events"] = log["events"][-200:] ap.write_text(json.dumps(log, indent=2)) except Exception: pass def reset_scores() -> dict: """Wipe the leaderboard (all solves removed).""" lb_path = TEAMS_DIR / "leaderboard.json" lb_path.write_text(json.dumps({"solves": []}, indent=2)) return {"ok": True, "cleared": True} def reset_environment() -> dict: """Full env reset: stop all teams, remove containers + receiver services, delete team dirs (fresh for re-create). Keeps panel + images.""" results = [] for d in sorted(TEAMS_DIR.glob("team*")): sf = d / "state.json" if not sf.exists(): continue st = json.loads(sf.read_text()) idx = st["index"] try: stop_team(idx) # compose down + stop receiver service + set status except Exception as e: results.append({"team": idx, "ok": False, "err": str(e)}) continue # remove the per-team systemd receiver unit subprocess.run(["systemctl", "disable", f"gemastik-receiver-team{idx}.service"], check=False, capture_output=True) subprocess.run(["systemctl", "stop", f"gemastik-receiver-team{idx}.service"], check=False, capture_output=True) unit = Path("/etc/systemd/system") / f"gemastik-receiver-team{idx}.service" if unit.exists(): unit.unlink() results.append({"team": idx, "ok": True}) subprocess.run(["systemctl", "daemon-reload"], check=False) # remove team dirs entirely (fresh for re-create) for d in sorted(TEAMS_DIR.glob("team*")): shutil.rmtree(d, ignore_errors=True) # wipe leaderboard too reset_scores() # drop team domains from Traefik (regenerate — no teams left) try: ensure_team_domains() except Exception: pass return {"ok": True, "stopped": results, "teams_dir": str(TEAMS_DIR)} if __name__ == "__main__": import sys cmd = sys.argv[1] if len(sys.argv) > 1 else "list" if cmd == "create" and len(sys.argv) > 2: st = create_team(int(sys.argv[2])) print(json.dumps(st, indent=2)) elif cmd == "list": print(json.dumps(list_teams(), indent=2)) elif cmd == "start" and len(sys.argv) > 2: print(json.dumps(start_team(int(sys.argv[2])), indent=2)) elif cmd == "stop" and len(sys.argv) > 2: print(json.dumps(stop_team(int(sys.argv[2])), indent=2))