MythEclipse/attack-defense-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
43ed3df557e4e4d879976fdb27a6e661974a9432
attack-defense-platform/panel/teams.py
T
root 43ed3df557 feat: team portal with own login, SSH web terminal, targets & guide 
- Portal tim punya login sendiri (password = ssh_pass, session team_token)
- SSH Web GUI: /api/team/{idx}/ssh/ws (WebSocket+paramiko) → xterm.js terminal
  (fix: ssh_to_ws non-blocking poll, chall_passwords per-container auth)
- Root <slug>.gemastik.imrnes.team → portal tim (bukan login admin)
- Host validation: /team/{idx} & /team/{idx}/guide 403 kalau host != team domain
- /api/team/{idx}/targets: daftar service tim musuh (attack target)
- guide.html: panduan SSH/attack/defense untuk peserta
- fix esc() String(s) (bug: (s||'').replace is not a function saat port number)
- set_ssh_passwords retry loop (container boot race)
2026-09-23 16:37:58 +08:00

423 lines
17 KiB
Python
Raw Blame History

#!/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 = 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 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):
"""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}"
slug = slugify(label)
state = {
"index": idx,
"label": label,
"slug": slug,
"domain": f"{slug}.gemastik.imrnes.team",
"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: <name>
# args:
# - PASSWORD=$PASSWORD_XXXXX
# Replace the whole block with " image: services-<name>".
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" -> "<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", "-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):
"""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 ensure_team_domains() -> str:
"""Write Traefik dynamic config for each team domain -> panel (:18081).
Each team gets <slug>.gemastik.imrnes.team. The panel routes
/team/<idx> to that team's portal page (public, no panitia login),
and /api/team/<idx>/* 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(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))
Reference in New Issue View Git Blame Copy Permalink