#!/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 threading 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 registry — single source of truth across all distributed sets. # Loaded from teams/challenge_registry.json (admin can toggle enabled). # # CHALLENGES below is a DERIVED list: (name, chall_offset, ssh_offset) for # every ENABLED challenge, in registry order. All team logic uses this. # --------------------------------------------------------------------------- _REGISTRY_PATH = TEAMS_DIR / "challenge_registry.json" def _default_registry() -> dict: return {"sets": {}, "challenges": []} def load_registry() -> dict: try: return json.loads(_REGISTRY_PATH.read_text()) except Exception: return _default_registry() def save_registry(reg: dict): _REGISTRY_PATH.write_text(json.dumps(reg, indent=2)) def registry_challenges() -> list: """All challenge dicts from the registry (enabled or not), in order.""" return load_registry().get("challenges", []) def enabled_challenges() -> list: return [c for c in registry_challenges() if c.get("enabled")] def set_challenge_enabled(name: str, enabled: bool) -> dict: """Flip a challenge's enabled flag in the registry (global toggle).""" reg = load_registry() for c in reg.get("challenges", []): if c["name"] == name: if bool(c.get("enabled")) == bool(enabled): return {"unchanged": True, **c} c["enabled"] = bool(enabled) save_registry(reg) return c raise KeyError(f"Challenge {name} tidak ada di registry") # Serializes sync_challenge_runtime(): each run rewrites every team's # docker-compose.yml and shells out to docker compose in those same dirs. _SYNC_LOCK = threading.Lock() def reconcile_team_state() -> list[str]: """Make every team's state.json agree with the registry's enabled set. Needed whenever the registry is edited directly (panel/set_enabled.py) or a challenge is enabled but a team was offline/stopped while it happened: state.json must carry ports + a chall password + a flag for every enabled challenge, otherwise render_team_compose() raises KeyError on ports[name]. Returns a list of human-readable actions taken. """ reg = load_registry() enabled = enabled_challenges() # rebuild the derived CHALLENGES for fresh creates global CHALLENGES CHALLENGES = [(c["name"], c["chall_offset"], c["ssh_offset"]) for c in enabled] notes: list[str] = [] 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"] dirty = False st.setdefault("ports", {}) st.setdefault("chall_passwords", {}) st.setdefault("flags", {}) for ch in enabled: name = ch["name"] if name not in st["ports"]: st["ports"][name] = {"chall": 30000 + idx * 1000 + ch["chall_offset"], "ssh": 30000 + idx * 1000 + ch["ssh_offset"]} dirty = True notes.append(f"team{idx}: allocated ports for {name}") if not st["chall_passwords"].get(name): st["chall_passwords"][name] = f"chall{idx}_{name}_{secrets.token_hex(4)}" dirty = True if not st["flags"].get(name): flag = f"GEMASTIK18{{TEAM{idx}_{name.upper()}_{secrets.token_hex(6)}}}" st["flags"][name] = flag fd = d / "receiver" / "flags" fd.mkdir(parents=True, exist_ok=True) (fd / f"{name}.txt").write_text(flag) dirty = True notes.append(f"team{idx}: minted flag for {name}") # make sure the source tree is present for a later `build:` ts = d / "services" / name if not ts.exists(): src = SERVICES_SRC / name if src.exists(): shutil.copytree(src, ts, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".git")) notes.append(f"team{idx}: copied source for {name}") if dirty: sf.write_text(json.dumps(st, indent=2)) return notes def sync_challenge_runtime(name: str, enabled: bool) -> dict: """Apply one challenge's enabled flag to every live team. Serialized via _SYNC_LOCK: multiple toggle requests rewrite the same per-team docker-compose.yml and run docker compose in the same directories, so concurrent syncs would corrupt each other's output. """ with _SYNC_LOCK: return _sync_challenge_runtime_locked(name, enabled) def _sync_challenge_runtime_locked(name: str, enabled: bool) -> dict: """Apply an enable/disable toggle to every RUNNING team: - regenerate that team's compose from the registry (compose_gen) - for ENABLE: docker compose up -d (builds image first if needed) - for DISABLE: docker compose rm -sf (stop+remove the container) - restart the team receiver so its challenge dict matches (main.py) Returns a per-team report. """ import compose_gen reg = load_registry() ch = next((c for c in reg.get("challenges", []) if c["name"] == name), None) if not ch: raise KeyError(f"Challenge {name} tidak ada di registry") # rebuild the derived CHALLENGES for fresh creates global CHALLENGES CHALLENGES = [(c["name"], c["chall_offset"], c["ssh_offset"]) for c in enabled_challenges()] report = {"challenge": name, "enabled": bool(enabled), "teams": []} 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"] svc_dir = d / "services" try: if enabled: # fresh flag file for this challenge flags_dir = d / "receiver" / "flags" flags_dir.mkdir(parents=True, exist_ok=True) flag = f"GEMASTIK18{{TEAM{idx}_{name.upper()}_{secrets.token_hex(6)}}}" (flags_dir / f"{name}.txt").write_text(flag) st.setdefault("flags", {})[name] = flag st["ports"][name] = {"chall": 30000 + idx*1000 + ch["chall_offset"], "ssh": 30000 + idx*1000 + ch["ssh_offset"]} st.setdefault("chall_passwords", {})[name] = st["chall_passwords"].get( name) or f"chall{idx}_{name}_{secrets.token_hex(4)}" # write state BEFORE rendering (render needs ports[name]) (sf).write_text(json.dumps(st, indent=2)) # Copy the challenge source into the team's services tree so a # `build: context: .` resolves (team dirs only hold challenges # that were enabled at create_team time). team_svc_src = svc_dir / name if not team_svc_src.exists(): src = SERVICES_SRC / name if not src.exists(): raise FileNotFoundError(f"sumber service tidak ada: {src}") shutil.copytree(src, team_svc_src, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".git")) # apt-insecure.conf is needed by every challenge build shared_apt = SERVICES_SRC / "apt-insecure.conf" if shared_apt.exists() and not (team_svc_src / "apt-insecure.conf").exists(): shutil.copy2(shared_apt, team_svc_src / "apt-insecure.conf") # regenerate whole compose (so enabled challenge included), # then bring up just this service new_text = compose_gen.render_team_compose(idx, st) (svc_dir / "docker-compose.yml").write_text(new_text) # inject the team-specific password env if compose uses ${PASSWORD_*} envp = svc_dir / ".env" if not envp.exists(): env_lines = [f"ADMIN_USERNAME={st['admin_user']}", f"ADMIN_PASSWORD={st['admin_pass']}", f"COMPOSE_LOCATION={svc_dir}/docker-compose.yml"] for i in range(20): env_lines.append(f"PASSWORD_{(i*1000)+10000}=placeholder") (envp).write_text("\n".join(env_lines) + "\n") r = subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "up", "-d", "--build", name], cwd=str(svc_dir), capture_output=True, text=True, timeout=1800) ok = r.returncode == 0 report["teams"].append({"team": idx, "ok": ok, "detail": (r.stdout or r.stderr)[-300:]}) if ok: # set the per-team SSH password after container boot try: pw = st["chall_passwords"][name] subprocess.run(["docker", "exec", f"{name}_container_team{idx}", "sh", "-c", f"echo 'ctfuser:{pw}' | chpasswd"], capture_output=True, timeout=60) except Exception: pass else: # stop + remove the container FIRST (old compose), then regenerate r = subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", svc_dir / "docker-compose.yml", "rm", "-sf", name], cwd=str(svc_dir), capture_output=True, text=True, timeout=120) # remove from state ports/flags st["ports"].pop(name, None) st.setdefault("flags", {}).pop(name, None) (sf).write_text(json.dumps(st, indent=2)) # regenerate compose WITHOUT this challenge new_text = compose_gen.render_team_compose(idx, st) (svc_dir / "docker-compose.yml").write_text(new_text) report["teams"].append({"team": idx, "ok": True, "detail": (r.stdout or r.stderr)[-300:] or "removed"}) # restart receiver so its challenge set matches _start_receiver(idx) except Exception as e: report["teams"].append({"team": idx, "ok": False, "detail": str(e)[-300:]}) return report # Derived list used everywhere (keeps old API: tuples of name, coff, soff) CHALLENGES = [(c["name"], c["chall_offset"], c["ssh_offset"]) for c in enabled_challenges()] # Points system: base points earned by stealing a flag from another team's # challenge. The SLA bonus is earned by keeping your OWN services alive. POINTS_PER_FLAG = 100 SLA_BONUS_POINTS = 50 SLA_BONUS_MIN_ALIVE = 6 # bonus only when ALL 6 services are UP def _load_points() -> dict: p = TEAMS_DIR / "points.json" if p.exists(): try: return json.loads(p.read_text()) except Exception: pass return {"teams": {}} def _save_points(data: dict): (TEAMS_DIR / "points.json").write_text(json.dumps(data, indent=2)) def get_team_points(team_idx: int) -> int: """Total attack points a team has earned (from flag steals).""" data = _load_points() return int(data.get("teams", {}).get(str(team_idx), {}).get("points", 0)) def add_attack_points(team_idx: int, points: int, chall: str = "", target: int = 0, ts: float = None) -> dict: """Award points to a team for stealing a flag. Returns updated tally.""" data = _load_points() tid = str(team_idx) me = data["teams"].setdefault(tid, {"points": 0, "events": []}) me["points"] = int(me.get("points", 0)) + points me["events"].append({ "type": "attack", "challenge": chall, "target": target, "points": points, "ts": ts if ts is not None else time.time(), "ts_human": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) me["events"] = me["events"][-200:] _save_points(data) return {"team": team_idx, "points": me["points"], "awarded": points} def add_sla_bonus(team_idx: int, alive: int, total: int = 6) -> dict: """Award SLA bonus when all services are UP. Applies bonus at most once per 5-minute window so online checks don't spam the ledger.""" data = _load_points() tid = str(team_idx) me = data["teams"].setdefault(tid, {"points": 0, "events": []}) now = time.time() last = me.get("last_sla_bonus", 0) min_alive = max(1, len(enabled_challenges())) if alive >= min_alive and total >= min_alive: if now - last > 300: # 5 min window me["points"] = int(me.get("points", 0)) + SLA_BONUS_POINTS me["last_sla_bonus"] = now me["events"].append({ "type": "sla_bonus", "alive": alive, "total": total, "points": SLA_BONUS_POINTS, "ts": now, "ts_human": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) me["events"] = me["events"][-200:] _save_points(data) return {"team": team_idx, "points": me["points"], "awarded": SLA_BONUS_POINTS, "bonus": True} return {"team": team_idx, "points": me["points"], "awarded": 0, "bonus": False} def team_rank(idx: int) -> int: """1-based rank of team by total points.""" data = _load_points()["teams"] rows = [(int(t), int(v.get("points", 0))) for t, v in data.items() if int(t) > 0] rows.sort(key=lambda x: -x[1]) for i, (t, _) in enumerate(rows, 1): if t == idx: return i return len(rows) + 1 # teams with 0 points rank after all scorers def team_badge(idx: int) -> str: """Emoji badge for podium teams.""" r = team_rank(idx) return {1: "👑 Juara 1", 2: "🥈 Runner-up", 3: "🥉 Peringkat 3"}.get(r, "") 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.attackdefense.imrnes.team" (full host accepted). Defaults to slugify(label).attackdefense.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 *.attackdefense.imrnes.team host = (domain or f"{slug}.attackdefense.imrnes.team").strip().lower() host = host.replace("https://", "").replace("http://", "").rstrip("/") if not host.endswith(".attackdefense.imrnes.team"): host = f"{host}.attackdefense.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) svc_dir.mkdir(parents=True, exist_ok=True) # copy utils next to services (compose references ../utils/bashrc) 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}")) # generate compose from the challenge registry (compose_gen renders the # per-team file: image: services-, team ports, team passwords) import compose_gen compose_text = compose_gen.render_team_compose(idx, state) compose = svc_dir / "docker-compose.yml" compose.write_text(compose_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 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]}" # systemd EnvironmentFile keys must match [A-Za-z_][A-Za-z0-9_]* — a # hyphen (gift-card, bit-canvas, ...) makes systemd log # "Ignoring invalid environment assignment" and DROP the line, so the # checker falls back to a default port/container and reports the # service down. Normalize to underscores on the writer; the reader # (gen_receiver_main._envkey) uses the same normalization. key = name.upper().replace("-", "_") recv_env.append(f"CHALLENGE_PORT_{key}={ports[name]['chall']}") recv_env.append(f"CHALLENGE_CONTAINER_{key}={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(".attackdefense.imrnes.team"): host = f"{host}.attackdefense.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 missing_sidecars(idx: int) -> list[str]: """Sidecar services in the team's compose that are NOT running. A multi-container challenge (anti-alchemy + its postgres, gemas-notes + database/validation, kode-viewer + redis, ...) needs its sidecars to boot: the main service's `create_db_conn()` retries in an infinite loop until the DB hostname resolves, so a missing sidecar leaves the main container "Up" with NO app listening and the checker reports it DOWN. Symptom class: container is running, port is open at the docker level, but the app never starts. Recover with `docker compose -p teamN up -d` (no service name). """ svc_dir = TEAMS_DIR / f"team{idx}" / "services" compose = svc_dir / "docker-compose.yml" if not compose.exists(): return [] declared = _compose_service_names(compose, project=f"team{idx}") if not declared: return [] # _compose_service_names returns the REAL container_name values # (`_container_teamN`), so compare them as-is — do NOT re-wrap them # in the compose default `teamN--1`, which no service here uses. want = set(declared) running = set(subprocess.run(["docker", "ps", "--format", "{{.Names}}"], capture_output=True, text=True).stdout.split()) return sorted(want - running) def _compose_service_names(compose: Path, project: str = "") -> list[str]: """Container names the compose will create, without needing PyYAML. Two shapes exist in these composes and BOTH must be caught: * services with an explicit `container_name:` (`_container_teamN`) — that name is what the SLA checkers and the receiver env key off, so it must be matched as-is; * sidecars WITHOUT a `container_name:` (anti-alchemy-db, gemas-notes-db) — compose names those `teamN--1`, and they are exactly the ones that go missing, because a per-service `up
` never starts them. Matching only the first shape reported "[] missing" while the db sidecar was absent on every team, and matching the compose default for everything reported all 16 running challenges as missing. Both are wrong; return the real name when there is one and the compose default when there isn't. """ names: list[str] = [] in_services = False pending: str | None = None pfx = f"{project}-" if project else "" for line in compose.read_text().splitlines(): if not line.strip() or line.lstrip().startswith("#"): continue if re.match(r"^services:\s*$", line): in_services = True continue if in_services and re.match(r"^[a-zA-Z]", line): break # next top-level key if not in_services: continue m = re.match(r"^ ([A-Za-z0-9_.-]+):\s*$", line) if m: if pending is not None: # previous service had no container_name -> compose default names.append(f"{pfx}{pending}-1") pending = m.group(1) continue m = re.match(r'^\s+container_name:\s*"?([A-Za-z0-9_.-]+)"?\s*$', line) if m and pending is not None: names.append(m.group(1)) pending = None if pending is not None: names.append(f"{pfx}{pending}-1") return names 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) # Self-heal: a per-service `up` (challenge toggle) or a raced start can # leave a sidecar behind while the main container looks fine. One plain # `up -d` reconciles the whole project (already-running ones are no-ops). gone = missing_sidecars(idx) if gone: 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["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) if gone: print(f"[start_team] team{idx}: started missing sidecar(s): {', '.join(gone)}") return st def challenge_ssh_users() -> dict: """{challenge_name: ssh login} from the registry. Only the 6 native GEMASTIK XVIII images provision `ctfuser`. Every imported XVI/XVII image does `RUN echo root:${PASSWORD} | chpasswd` and logs in as `root` (some also create an unprivileged `ctf` for the app). Hardcoding `ctfuser` made chpasswd either fail or set a password on an account nobody uses, so SSH returned "Permission denied" for every team on 10 of 16 challenges while state.json still looked correct. """ out = {} for c in registry_challenges(): out[c["name"]] = c.get("ssh_user") or "root" return out def all_teams() -> list: """Every team that still has a state.json, newest index last.""" out = [] for d in sorted(TEAMS_DIR.glob("team*")): sf = d / "state.json" if not sf.exists(): continue try: st = json.loads(sf.read_text()) except Exception: continue if "index" in st: out.append(st) return out def team_state(idx: int): """state.json for one team, or None if that team doesn't exist.""" sf = TEAMS_DIR / f"team{idx}" / "state.json" if not sf.exists(): return None try: return json.loads(sf.read_text()) except Exception: return None def challenge_credential(idx: int, name: str): """{username, password, port} a participant should actually use for SSH. Reads the TEAM's state.json rather than the receiver: the global receiver on :18080 only knows the 6 native GEMASTIK XVIII challenges, so asking it for an imported XVI/XVII challenge returns "Invalid challenge" and the UI showed participants nothing at all. The registry `ssh_user` is the same field the panel chpasswds, so the two can never drift. """ st = team_state(idx) if not st: return None pwd = (st.get("chall_passwords") or {}).get(name) if not pwd: return None return { "username": challenge_ssh_users().get(name, "ctfuser"), "password": pwd, "port": ((st.get("ports") or {}).get(name) or {}).get("chall"), "team": idx, } def set_ssh_passwords(idx: int): """Set the SSH password for the CORRECT login of each challenge container. The login is per-challenge (registry `ssh_user`), not a global `ctfuser`. Retries because right after `compose up` the container may still be booting. Reports failures loudly instead of writing to a log nobody reads. """ team_dir = TEAMS_DIR / f"team{idx}" st = json.loads((team_dir / "state.json").read_text()) users = challenge_ssh_users() failures = [] for name, coff, soff in CHALLENGES: cont = f"{name}_container_team{idx}" user = users.get(name, "root") pw = st["chall_passwords"].get(name) if not pw: failures.append(f"{cont}: no password in state") continue # Set the password for the real login AND for ctfuser when that account # exists, so either convention works during a transition. cmd = (f"id {user} >/dev/null 2>&1 && echo '{user}:{pw}' | chpasswd; " f"id ctfuser >/dev/null 2>&1 && echo 'ctfuser:{pw}' | chpasswd; " f"id ctf >/dev/null 2>&1 && echo 'ctf:{pw}' | chpasswd; " f"id {user} >/dev/null 2>&1") ok = False r = None 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: failures.append(f"{cont}: {(r.stderr or '').strip()[:100]}") else: print(f"[set_ssh_passwords] {cont} (login={user}): OK") if failures: print(f"[set_ssh_passwords] team{idx} FAILED {len(failures)}/{len(CHALLENGES)}: " + "; ".join(failures)) return failures 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 .attackdefense.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}.attackdefense.imrnes.team" (td / "state.json").write_text(json.dumps(st, indent=2)) changed = True host = f"{slug}.attackdefense.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: # No teams left. The file MUST still be rewritten (to an empty router # set) — returning early leaves the previous content on disk, so a # DELETED team keeps its Traefik router and stays routable to the panel # portal forever. That was the stale-domain bug. (traefik_dir / "attackdefense-teams.yaml").write_text("http:\n routers: {}\n") return "no teams to route (emptied attackdefense-teams.yaml)" # Keep the team domain block in its own file so the main attackdefense.yaml stays untouched (traefik_dir / "attackdefense-teams.yaml").write_text("http:\n routers:\n" + "".join(out)) return f"wrote {len(out)} team domain(s) in attackdefense-teams.yaml" def team_port_block(idx: int) -> list[int]: """Every host port a team occupies: each challenge's chall+ssh port, the receiver, and the reserved panel slot.""" st_path = TEAMS_DIR / f"team{idx}" / "state.json" ports: set[int] = set() if st_path.exists(): st = json.loads(st_path.read_text()) for name, pv in (st.get("ports") or {}).items(): if isinstance(pv, dict): for k in ("chall", "ssh"): if pv.get(k): ports.add(int(pv[k])) elif isinstance(pv, int): ports.add(pv) else: # team dir already gone (mid-delete): derive from the port scheme base = PORT_BASE + idx * STEP for name, coff, soff in CHALLENGES: ports.add(base + coff) ports.add(base + soff) ports.add(base + 80) ports.add(base + 81) return sorted(ports) def sync_team_ufw(idx: int, remove: bool = False) -> dict: """Reconcile UFW rules for one team's port block. UFW here defaults to deny(incoming), so a port that is not explicitly allowed is blackholed — the team is unreachable from the internet AND from its own containers. Team creation never opened these ports (they were opened by hand earlier), so a new team silently gets no connectivity. remove=True DELETES the rules instead of adding them (called from delete_team). The two modes are mutually exclusive: never add-then-delete, that would flap the rules and briefly re-open a dead team's ports. """ ports = team_port_block(idx) if remove: for p in ports: subprocess.run(["ufw", "--force", "delete", "allow", f"{p}/tcp"], check=False, capture_output=True) return {"team": idx, "ports": len(ports), "closed": ports, "failed": []} failed = [] for p in ports: r = subprocess.run(["ufw", "allow", f"{p}/tcp"], check=False, capture_output=True, text=True) # `ufw allow` on an existing rule prints "Skipping adding existing rule" # and still exits 0; a non-zero rc with a skip message is not a failure. if r.returncode != 0 and "Skipping" not in (r.stdout + r.stderr): failed.append(p) return {"team": idx, "ports": len(ports), "opened": [p for p in ports if p not in failed], "failed": failed} def _purge_team_records(idx: int) -> dict: """Drop every ledger row that references a team, so a deleted team leaves no ghost entries on the leaderboard / points / attack topology.""" removed = {"leaderboard": 0, "points": 0, "attacks": 0} lb_path = TEAMS_DIR / "leaderboard.json" if lb_path.exists(): try: lb = json.loads(lb_path.read_text()) before = len(lb.get("solves", [])) lb["solves"] = [e for e in lb.get("solves", []) if e.get("team") != idx and e.get("target") != idx] removed["leaderboard"] = before - len(lb["solves"]) lb_path.write_text(json.dumps(lb, indent=2)) except Exception: pass p_path = TEAMS_DIR / "points.json" if p_path.exists(): try: data = json.loads(p_path.read_text()) if str(idx) in data.get("teams", {}): data["teams"].pop(str(idx)) removed["points"] = 1 p_path.write_text(json.dumps(data, indent=2)) except Exception: pass a_path = TEAMS_DIR / "attacks.json" if a_path.exists(): try: log = json.loads(a_path.read_text()) before = len(log.get("events", [])) log["events"] = [e for e in log.get("events", []) if e.get("attacker") != idx and e.get("target") != idx] removed["attacks"] = before - len(log["events"]) a_path.write_text(json.dumps(log, indent=2)) except Exception: pass return removed def repair_team_receiver_env(idx: int) -> dict: """Rewrite a team receiver .env with systemd-legal keys. Teams created before the hyphen fix have `CHALLENGE_PORT_GIFT-CARD=` in their .env. systemd logs "Ignoring invalid environment assignment" and drops the line, so every hyphenated challenge (gift-card, bit-canvas, gleam-drive, more-less, anti-alchemy, gift-voucher) reports DOWN even though its container is up. This rewrites the file in place, fixing existing teams without forcing a re-create. """ env_path = TEAMS_DIR / f"team{idx}" / "receiver" / ".env" if not env_path.exists(): raise FileNotFoundError(f"team{idx} receiver .env not found") st = json.loads((TEAMS_DIR / f"team{idx}" / "state.json").read_text()) lines = env_path.read_text().splitlines() kept, fixed, dropped = [], [], [] for line in lines: if line.startswith("CHALLENGE_PORT_") or line.startswith("CHALLENGE_CONTAINER_"): k, _, v = line.partition("=") nk = k.replace("-", "_") if nk != k: fixed.append(f"{k}->{nk}") else: kept.append(line) continue kept.append(line) # re-append authoritative values for every challenge in state for name in (st.get("ports") or {}): if name in ("receiver", "panel"): continue key = name.upper().replace("-", "_") kept.append(f"CHALLENGE_PORT_{key}={st['ports'][name]['chall']}") kept.append(f"CHALLENGE_CONTAINER_{key}={name}_container_team{idx}") env_path.write_text("\n".join(kept) + "\n") # also refresh the shared checker packages (team1 was missing xvi.Art) try: sys.path.insert(0, str(BASE / "panel")) from gen_receiver_main import _sync_checker_packages _sync_checker_packages(TEAMS_DIR / f"team{idx}" / "receiver") except Exception as e: dropped.append(f"checker sync failed: {e}") return {"team": idx, "fixed_keys": fixed, "notes": dropped} def delete_team(idx: int, purge_scores: bool = True) -> dict: """Permanently remove ONE team and free everything it owns. Teardown order matters — do the teardown against the OLD compose before removing the directory, or `docker compose` has no file to read and the containers survive as orphans: 1. stop the per-team receiver systemd unit and remove the unit file 2. `docker compose -p teamN down -v` against the team compose (also drops the teamN_default network) 3. force-remove any surviving _container_teamN container 4. delete the team's UFW rules 5. rmtree teams/teamN (compose, receiver, flags, state.json) 6. purge leaderboard / points / attacks rows for that team 7. rewrite the Traefik team-domain file so the domain stops resolving Images (services-*) are SHARED across teams and are never removed here. purge_scores=False keeps the team's leaderboard/points history. """ 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()) label = st.get("label", f"Team {idx}") steps: list[str] = [] # 1. receiver unit unit = f"gemastik-receiver-team{idx}.service" subprocess.run(["systemctl", "disable", unit], check=False, capture_output=True) subprocess.run(["systemctl", "stop", unit], check=False, capture_output=True) unit_path = Path("/etc/systemd/system") / f"{unit}" if unit_path.exists(): unit_path.unlink() steps.append("removed receiver unit") subprocess.run(["systemctl", "daemon-reload"], check=False, capture_output=True) # 2. compose down (containers + network) while the compose file still exists svc_dir = team_dir / "services" compose = svc_dir / "docker-compose.yml" if compose.exists(): r = subprocess.run(["docker", "compose", "-p", f"team{idx}", "-f", str(compose), "down", "-v", "--remove-orphans"], cwd=str(svc_dir), check=False, capture_output=True, text=True, timeout=600) steps.append("compose down" if r.returncode == 0 else f"compose down rc={r.returncode}") # 3. force-remove leftovers (a half-written compose can leave orphans) left = subprocess.run(["docker", "ps", "-aq", "--filter", f"name=_container_team{idx}"], capture_output=True, text=True).stdout.split() if left: subprocess.run(["docker", "rm", "-f", *left], check=False, capture_output=True) steps.append(f"force-removed {len(left)} container(s)") net_rm = subprocess.run(["docker", "network", "rm", f"team{idx}_default"], check=False, capture_output=True, text=True) if net_rm.returncode == 0: steps.append("removed docker network") # 4. UFW ufw = sync_team_ufw(idx, remove=True) steps.append(f"closed {len(ufw.get('closed', []))} ufw port(s)") # 5. directory shutil.rmtree(team_dir, ignore_errors=True) if team_dir.exists(): raise RuntimeError(f"team{idx} directory could not be removed") steps.append("deleted team directory") # 6. ledgers purged = _purge_team_records(idx) if purge_scores else {} if purge_scores: steps.append("purged score records") # 7. Traefik: drop the dead domain try: ensure_team_domains() steps.append("rewrote team domains") except Exception as e: steps.append(f"traefik rewrite failed: {e}") return {"ok": True, "team": idx, "label": label, "domain": st.get("domain", ""), "steps": steps, "purged": purged} 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)) # NEW: award attack points (first solve only) pts = add_attack_points(attacker_idx, POINTS_PER_FLAG, chall=chall, target=target_idx) else: pts = {"team": attacker_idx, "points": get_team_points(attacker_idx), "awarded": 0} return {"success": True, "team": attacker_idx, "target": target_idx, "challenge": chall, "points": pts} 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) and the points ledger.""" lb_path = TEAMS_DIR / "leaderboard.json" lb_path.write_text(json.dumps({"solves": []}, indent=2)) (TEAMS_DIR / "points.json").write_text(json.dumps({"teams": {}}, 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)} # ---- SLA + scoring helpers ---- _SLA_CACHE = {"ts": 0, "data": None} def _probe_one(recv_port: int, au: str, ap: str, name: str, st: dict) -> dict: import urllib.request, urllib.error, base64 url = f"http://127.0.0.1:{recv_port}/check/{name}" ok = False try: req = urllib.request.Request(url) token = base64.b64encode(f"{au}:{ap}".encode()).decode() req.add_header("Authorization", f"Basic {token}") with urllib.request.urlopen(req, timeout=25) as resp: body = resp.read().decode() import json as _j ok = bool(_j.loads(body).get("success")) if resp.status == 200 else False except Exception: ok = False return {"name": name, "port": st["ports"][name]["chall"], "ssh": st["ports"][name]["ssh"], "alive": ok} def probe_team_sla_fast(idx: int) -> dict: """Probe one team's challenges. Sequential per challenge (receivers are sync Flask; parallel probes overload them and cause false timeouts). ~30-60s worst case for 6 challs; results are cached by the refresher.""" td = TEAMS_DIR / f"team{idx}" sf = td / "state.json" if not sf.exists(): return {"team": idx, "alive": 0, "total": 0, "per_challenge": [], "error": "no team"} st = json.loads(sf.read_text()) recv_port = st["ports"]["receiver"] au, ap = st.get("admin_user", ""), st.get("admin_pass", "") results = [] for name, _, _ in CHALLENGES: try: results.append(_probe_one(recv_port, au, ap, name, st)) except Exception: results.append({"name": name, "port": 0, "ssh": 0, "alive": False}) alive = sum(1 for r in results if r["alive"]) return {"team": idx, "alive": alive, "total": len(results), "per_challenge": results} def _scoreboard_row(st: dict) -> dict: """Build one scoreboard row for a team state (runs in a worker thread).""" idx = st["index"] sla = probe_team_sla_fast(idx) pts = get_team_points(idx) solves = 0 lb_path = TEAMS_DIR / "leaderboard.json" if lb_path.exists(): try: lb = json.loads(lb_path.read_text()) solves = sum(1 for e in lb["solves"] if e["team"] == idx) except Exception: pass return { "team": idx, "label": st.get("label") or f"Team {idx}", "domain": st.get("domain") or "", "alive": sla["alive"], "total": sla["total"], "sla_pct": round(100 * sla["alive"] / sla["total"], 1) if sla["total"] else 0, "points": pts, "solves": solves, } def sla_status_all() -> dict: """Per-team SLA (own team view) + aggregate scoreboard with points. Reads a cache that a background thread keeps fresh (~every 30s), so the HTTP endpoint is instant. First call (cold cache) blocks up to ~60s.""" import time as _t if _SLA_CACHE["data"] is not None and _t.time() - _SLA_CACHE["ts"] < 60: return _SLA_CACHE["data"] _build_scoreboard() return _SLA_CACHE["data"] def _build_scoreboard() -> dict: """Build the scoreboard: probes teams 2-at-a-time (6 chall parallel per team) to avoid overwhelming the receivers, then caches the result.""" import time as _t import concurrent.futures states = [] for d in sorted(TEAMS_DIR.glob("team*")): sf = d / "state.json" if sf.exists(): states.append(json.loads(sf.read_text())) teams = [] # probe one team at a time (each team = 6 sequential chall checks) with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex: for row in ex.map(_scoreboard_row, states): teams.append(row) teams.sort(key=lambda x: (-x["points"], -x["solves"], x["team"])) for i, t in enumerate(teams, 1): t["rank"] = i t["badge"] = {1: "👑 Juara 1", 2: "🥈 Runner-up", 3: "🥉 Peringkat 3"}.get(i, "") _SLA_CACHE["ts"] = _t.time() _SLA_CACHE["data"] = {"teams": teams, "ts": _t.time()} return _SLA_CACHE["data"] def start_sla_refresher(): """Background daemon thread: keeps the SLA scoreboard cache warm.""" import threading def _loop(): import time as _t while True: try: _build_scoreboard() except Exception: pass _t.sleep(30) t = threading.Thread(target=_loop, daemon=True, name="sla-refresher") t.start() return t 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))