root
1ca963b2b7
- panel/teams.py: create/start/stop teams with isolated ports+creds, per-team receivers with CHALLENGE_PORT/CONTAINER env, flag randomization, submit validation + leaderboard
- panel/main.py: /api/teams, /api/teams/{idx}/randomize, /api/flag/submit, /api/leaderboard, /api/public/teams, /submit page
- panel/static/submit.html: public flag submission UI for teams
- receiver: _ch_port/_ch_container read .env per team, container name overrides
- .gitignore: exclude teams/, .venv, __pycache__
324 lines
14 KiB
Python
324 lines
14 KiB
Python
#!/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<N>/
|
|
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 = 20000 + i*1000 + 0..5 (blogpost,carbeat,cdn,phew,sheesh,warmup)
|
|
ssh ports = 20000 + i*1000 + 22..27 (10022-style)
|
|
receiver = 20000 + i*1000 + 80 (receiver API, e.g. 21080, 22080)
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shutil
|
|
import subprocess
|
|
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 = 20000
|
|
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 create_team(idx: int, label: str = None):
|
|
"""Build a full team stack dir with unique ports/passwords."""
|
|
ports = team_ports(idx)
|
|
team_dir = TEAMS_DIR / f"team{idx}"
|
|
label = label or f"Tim {idx}"
|
|
state = {
|
|
"index": idx,
|
|
"label": label,
|
|
"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()
|
|
# 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" -> "<team_chall>: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 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", "-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))
|
|
return st
|
|
|
|
def stop_team(idx: int):
|
|
team_dir = TEAMS_DIR / f"team{idx}"
|
|
svc_dir = team_dir / "services"
|
|
subprocess.run(["docker", "compose", "-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):
|
|
"""Launch team's receiver as a detached uvicorn process with team env."""
|
|
team_dir = TEAMS_DIR / f"team{idx}"
|
|
recv_dir = team_dir / "receiver"
|
|
st = json.loads((team_dir / "state.json").read_text())
|
|
port = st["ports"]["receiver"]
|
|
pidfile = team_dir / "receiver.pid"
|
|
# kill existing
|
|
_stop_receiver(idx)
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = str(recv_dir)
|
|
env["COMPOSE_LOCATION"] = str(team_dir / "services" / "docker-compose.yml")
|
|
# expose team ports/containers so challenge classes bind the right targets
|
|
for ch in st["ports"]:
|
|
if ch in ("receiver", "panel"):
|
|
continue
|
|
env[f"CHALLENGE_PORT_{ch.upper()}"] = str(st["ports"][ch]["chall"])
|
|
env[f"CHALLENGE_CONTAINER_{ch.upper()}"] = f"{ch}_container_team{idx}"
|
|
proc = subprocess.Popen(
|
|
[str(RECEIVER_SRC.parent / "receiver" / ".venv" / "bin" / "python"),
|
|
"-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", str(port)],
|
|
cwd=str(recv_dir), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
start_new_session=True)
|
|
pidfile.write_text(str(proc.pid))
|
|
|
|
def _stop_receiver(idx: int):
|
|
team_dir = TEAMS_DIR / f"team{idx}"
|
|
pidfile = team_dir / "receiver.pid"
|
|
if pidfile.exists():
|
|
try:
|
|
pid = int(pidfile.read_text().strip())
|
|
os.kill(pid, 15)
|
|
except Exception:
|
|
pass
|
|
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 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", "-f", svc_dir / "docker-compose.yml",
|
|
"down"], cwd=str(svc_dir), check=False, capture_output=True)
|
|
subprocess.run(["docker", "compose", "-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(team_idx: int, chall: str, flag: str, team_name: str = "") -> dict:
|
|
"""Validate a submitted flag against the owning team's challenge flag."""
|
|
team_dir = TEAMS_DIR / f"team{team_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:
|
|
return {"success": False, "error": "wrong flag"}
|
|
# record leaderboard entry
|
|
lb_path = TEAMS_DIR / "leaderboard.json"
|
|
lb = json.loads(lb_path.read_text()) if lb_path.exists() else {"solves": []}
|
|
entry = {
|
|
"team": team_idx,
|
|
"team_name": team_name or f"Team {team_idx}",
|
|
"challenge": chall,
|
|
"flag": flag,
|
|
"ts": time.time(),
|
|
"ts_human": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
# avoid double-solve duplicates (same flag + same team)
|
|
if not any(e["team"] == team_idx and e["challenge"] == chall for e in lb["solves"]):
|
|
lb["solves"].append(entry)
|
|
lb_path.write_text(json.dumps(lb, indent=2))
|
|
return {"success": True, "team": team_idx, "challenge": chall}
|
|
|
|
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)) |