MythEclipse/attack-defense-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
attack-defense-platform/panel/main.py
T
root 1461c874c3 Render the topology graph with PixiJS v8 
Replaces the hand-rolled inline-SVG topology with a PixiJS 8 scene graph.

Why: the old renderer rebuilt all 37 nodes / 36 edges as one innerHTML string
every 10s, which tore down and recreated every DOM node. That restarted CSS
animations mid-flight and made dragging fight the browser's own hit-testing.
The scene graph gives per-node transforms, so pan/zoom is a single container
transform instead of getScreenCTM() matrix math.

Changes:
- static/topo_pixi.js: new self-contained renderer. Owns its Application and
  tears it down on tab exit so a second WebGL context cannot leak.
- static/index.html: the <svg id=topoSvg> host becomes a <div id=topoHost>;
  the 188-line SVG renderer is replaced by a bridge to the module.
- static/vendor/pixi.mjs: PixiJS 8.21.0 self-hosted (MIT). The .mjs build is
  required; the .js build exports no global. See vendor/README.md.
- main.py: mount /static. Pages were served as inline HTMLResponse, so the
  directory was never mounted and the module had no URL to load from.

Two real bugs found by measuring pixels rather than trusting init():
- preserveDrawingBuffer: without it WebGL clears the back buffer after
  compositing, so any readback or screenshot of the canvas is a coin flip
  depending on which frame it lands on. The graph rendered intermittently
  blank. Now enabled: cheap for a 2D scene, and it makes the view capturable.
- Layout was centred on the SCROLLABLE width (nodes.length * 130), not the
  viewport, so with 37 nodes every team and challenge node landed at
  x=2230-2650 on a 1310px canvas: entirely off-screen. Layout now centres on
  the visible width and reset() frames the whole graph to fit.

test_topo_pixels.js documents three wrong test designs it replaces, all of
which reported false failures against a working graph: counting scene-graph
children (passes on a blank canvas), diffing against the background colour
(the theme is dark by design, so a perfect render measures ~0%), and diffing
two Playwright screenshots (both can be captured after the scene was mutated).
The check now reads the GL back buffer via readPixels in one evaluate.

Verified: 37 nodes / 36 edges drawn (7.94% of frame, max channel delta 225),
graph bbox [437,46,881,476] inside the 1310x520 canvas, glGetError=0, no page
errors, zoom and frame-to-fit reset working. Platform unregressed: SLA 32/32.
2026-09-27 00:31:47 +08:00

1086 lines
44 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
Gemastik A/D Panel — web UI for the gemastik18-final receiver.
Serves a dashboard at / and proxies receiver API calls server-side so the
admin credentials stay out of the browser.
"""
import os
import json
import time
import asyncio
import threading
import subprocess
import sys
import httpx
from pathlib import Path
from fastapi import FastAPI, Request, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from typing import Optional
import teams as orch
RECEIVER_URL = os.environ.get("RECEIVER_URL", "http://127.0.0.1:18080")
ADMIN_USER = os.environ.get("PANEL_ADMIN_USER", "admin")
ADMIN_PASS = os.environ.get("PANEL_ADMIN_PASS", "admin")
BASE_DIR = Path(__file__).parent
app = FastAPI(title="Gemastik A/D Panel")
# The panel used to serve every page as an inline HTMLResponse, so static/ was
# never mounted. The PixiJS topology needs real asset URLs: topo_pixi.js is an ES
# module and vendor/pixi.mjs is an 810 KB bundle — neither can be inlined.
# Mounted at the root so the module's own `import './vendor/pixi.mjs'` and
# `import('./topo_pixi.js')` resolve without rewriting paths.
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")
@app.on_event("startup")
async def _start_background():
"""Warm the SLA scoreboard cache in the background."""
import threading
threading.Thread(target=orch._build_scoreboard, daemon=True,
name="sla-warmup").start()
orch.start_sla_refresher()
# Toggle jobs: Docker builds take minutes, so PATCH /api/challenges runs the
# work on a background thread and the client polls /api/challenges/jobs/<id>.
_DELETE_JOBS = {}
_TOGGLE_JOBS: dict[str, dict] = {}
CHALLENGES = [
{"name": "blogpost", "port": 10000, "ssh": 10022, "category": "web", "desc": "Flask blog with exiftool + SSTI"},
{"name": "carbeat", "port": 11000, "ssh": 11022, "category": "pwn", "desc": "Binary exploitation menu"},
{"name": "cdn", "port": 12000, "ssh": 12022, "category": "web", "desc": "CDN/image proxy SSTI"},
{"name": "phew", "port": 13000, "ssh": 13022, "category": "crypto","desc": "Paillier crypto oracle"},
{"name": "sheesh", "port": 14000, "ssh": 14022, "category": "crypto","desc": "AES padding oracle"},
{"name": "warmup", "port": 15000, "ssh": 15022, "category": "warmup","desc": "Go file viewer (path traversal)"},
]
# Simple in-memory session tokens (good enough for a CTF ops panel)
_sessions = {}
def _check_basic(req: Request):
auth = req.headers.get("authorization", "")
if not auth.startswith("Basic "):
return None
import base64
try:
decoded = base64.b64decode(auth.split(" ", 1)[1]).decode()
user, _, pw = decoded.partition(":")
return (user, pw)
except Exception:
return None
def _authorized(req: Request) -> bool:
creds = _check_basic(req)
if creds and creds[0] == ADMIN_USER and creds[1] == ADMIN_PASS:
return True
# session token via cookie
token = req.cookies.get("panel_token")
return token in _sessions and _sessions[token] > time.time()
def _receiver_auth() -> tuple:
# Load receiver admin creds from its .env (single source of truth)
env_path = Path("/opt/gemastik18-final/receiver/.env")
u = p = ""
try:
for line in env_path.read_text().splitlines():
if line.startswith("ADMIN_USERNAME="):
u = line.split("=", 1)[1]
elif line.startswith("ADMIN_PASSWORD="):
p = line.split("=", 1)[1]
except Exception:
pass
return (u, p)
async def _proxy(method: str, path: str, body: dict = None):
u, p = _receiver_auth()
async with httpx.AsyncClient(timeout=20) as client:
resp = await client.request(method, f"{RECEIVER_URL}{path}",
auth=(u, p), json=body if body is not None else None)
return resp
@app.get("/", response_class=HTMLResponse)
async def index(req: Request):
host = (req.headers.get("host") or "").split(":")[0]
# Team portal domains: <slug>.attackdefense.imrnes.team -> their team portal (no admin login)
if host.endswith(".attackdefense.imrnes.team") and host != "attackdefense.imrnes.team" and host != "panel.attackdefense.imrnes.team":
slug = host.split(".")[0]
for t in orch.list_teams():
if t.get("slug") == slug:
html = (BASE_DIR / "static" / "team.html").read_text()
html = html.replace('name="team-id" content="0"', f'name="team-id" content="{t["index"]}"')
html = html.replace('href="/team/0/guide"', f'href="/team/{t["index"]}/guide"')
return HTMLResponse(html)
return HTMLResponse("<h2 style='font-family:sans-serif;color:#888;padding:40px'>Team tidak ditemukan: " + slug + "</h2>", status_code=404)
if not _authorized(req):
return RedirectResponse("/login")
html = (BASE_DIR / "static" / "index.html").read_text()
return HTMLResponse(html)
@app.get("/login", response_class=HTMLResponse)
async def login_page(req: Request):
if _authorized(req):
return RedirectResponse("/")
return HTMLResponse((BASE_DIR / "static" / "login.html").read_text())
@app.get("/submit", response_class=HTMLResponse)
async def submit_page(req: Request):
"""Public flag submission page for teams (no login)."""
return HTMLResponse((BASE_DIR / "static" / "submit.html").read_text())
# ============ Per-team portal (public via <slug>.attackdefense.imrnes.team) ============
@app.get("/team/{idx}", response_class=HTMLResponse)
async def team_portal(idx: int, req: Request):
"""Public portal page for a team. Must be accessed via that team's own domain."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
if _team_authorized(req, idx): # logged in -> show portal directly
html = (BASE_DIR / "static" / "team.html").read_text()
html = html.replace('name="team-id" content="0"', f'name="team-id" content="{idx}"')
# server-side: fix guide link so non-JS / pre-JS clicks never hit /team/0
html = html.replace('href="/team/0/guide"', f'href="/team/{idx}/guide"')
return HTMLResponse(html)
# not logged in -> show a stripped landing/login page (no admin info)
html = (BASE_DIR / "static" / "team.html").read_text()
html = html.replace('name="team-id" content="0"', f'name="team-id" content="{idx}"')
html = html.replace('href="/team/0/guide"', f'href="/team/{idx}/guide"')
return HTMLResponse(html)
@app.get("/team/{idx}/guide", response_class=HTMLResponse)
async def team_guide(idx: int, req: Request):
"""Public SSH/attack guide for a team. Must be accessed via that team's own domain."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
html = (BASE_DIR / "static" / "guide.html").read_text()
html = html.replace('name="team-id" content="0"', f'name="team-id" content="{idx}"')
return HTMLResponse(html)
# ---- Team portal login (separate from admin; password = team ssh_pass) ----
_team_sessions: dict[str, tuple[int, float]] = {} # token -> (idx, expiry)
@app.post("/api/team/{idx}/login")
async def api_team_login(idx: int, req: Request):
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
data = await req.json()
pw = data.get("pass", "")
if pw != st.get("ssh_pass"):
raise HTTPException(401, "Password salah")
token = os.urandom(16).hex()
_team_sessions[token] = (idx, time.time() + 12 * 3600)
resp = JSONResponse({"ok": True, "team": idx})
resp.set_cookie("team_token", token, httponly=True, samesite="lax", max_age=12 * 3600)
return resp
@app.post("/api/team/logout")
async def api_team_logout(req: Request):
token = req.cookies.get("team_token")
if token:
_team_sessions.pop(token, None)
return {"ok": True}
def _team_authorized(req: Request, idx: int) -> bool:
token = req.cookies.get("team_token")
if not token:
return False
e = _team_sessions.get(token)
if not e or e[0] != idx or e[1] < time.time():
_team_sessions.pop(token, None)
return False
return True
def _check_team_host(req: Request, st: dict) -> bool:
"""IDOR guard: host must be this team's own domain (or localhost).
panel.gemastik / attackdefense.imrnes.team only allowed with a valid ADMIN session."""
host = (req.headers.get("host") or "").split(":")[0]
if host == st.get("domain"):
return True
if host.startswith("127.0.0.1") or host.startswith("localhost"):
return True
if host in ("panel.attackdefense.imrnes.team", "attackdefense.imrnes.team"):
return _authorized(req) # admin preview only
return False
@app.get("/api/team/{idx}/session")
async def api_team_session(idx: int, req: Request):
"""True when this browser has a valid team session for idx."""
return {"authed": _team_authorized(req, idx)}
@app.get("/api/team/{idx}/own-challenges")
async def api_team_own_challenges(idx: int, req: Request):
"""The challenges THIS team runs, each with the SSH login it provisions.
The terminal's challenge picker used to be a hardcoded list of only the 6
native GEMASTIK XVIII entries, so the 10 imported XVI/XVII challenges could
not be selected at all. Names + per-challenge ssh_user only -- no passwords.
"""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
if not _team_authorized(req, idx):
raise HTTPException(401, "Login portal team dulu")
users = orch.challenge_ssh_users()
out = []
for name, _coff, _soff in orch.CHALLENGES:
p = st.get("ports", {}).get(name)
if not p:
continue
out.append({"name": name, "ssh_user": users.get(name, "ctfuser"),
"port": p.get("chall"), "ssh_port": p.get("ssh")})
return {"challenges": out}
@app.get("/api/team/{idx}/targets")
async def api_team_targets(idx: int, req: Request):
"""Team targets — requires team login + own host. Only domain + port."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st_self = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st_self):
raise HTTPException(403, "Akses team lain tidak diizinkan")
if not _team_authorized(req, idx):
raise HTTPException(401, "Login portal team dulu")
out = []
ssh_users = orch.challenge_ssh_users()
for d in sorted(orch.TEAMS_DIR.glob("team*")):
if not (d / "state.json").exists():
continue
st = json.loads((d / "state.json").read_text())
if st.get("index") == idx:
continue # skip self
for name, coff, soff in orch.CHALLENGES:
p = st["ports"].get(name)
if not p:
continue
out.append({
"team_idx": st.get("index"),
"team_label": st.get("label", f"Team {st.get('index')}"),
"challenge": name,
"domain": st.get("domain") or (st.get("slug", f"team{st.get('index')}") + ".attackdefense.imrnes.team"),
"port": p["chall"],
# attackers need the same login the victim container provisions;
# it is per-challenge, so surface it instead of assuming ctfuser
"ssh_user": ssh_users.get(name, "ctfuser"),
})
return {"targets": out}
@app.get("/api/team/{idx}/info")
async def api_team_info(idx: int, req: Request):
"""Team portal info — requires team login + own host; no admin secrets."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
if not _team_authorized(req, idx):
raise HTTPException(401, "Login portal team dulu")
return {"team": {
"index": st.get("index"),
"label": st.get("label"),
"slug": st.get("slug"),
"domain": st.get("domain"),
"status": st.get("status"),
"ports": st.get("ports"),
"ssh_user": st.get("ssh_user"),
"ssh_pass": st.get("ssh_pass"), # same password used to login portal + SSH
}}
@app.get("/api/team/{idx}/status")
async def api_team_status(idx: int, req: Request):
"""SLA status for a team's challenges (read-only health, own-host only)."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st = json.loads ((td / "state.json").read_text())
if not _check_team_host(req, st):
raise HTTPException(403, "Akses team lain tidak diizinkan")
recv_port = st["ports"]["receiver"]
# use team's receiver admin creds (server-side only; never sent to browser)
au, ap = st.get("admin_user", ""), st.get("admin_pass", "")
results = []
for name, coff, soff in orch.CHALLENGES:
try:
async with httpx.AsyncClient(timeout=8) as client:
resp = await client.get(f"http://127.0.0.1:{recv_port}/check/{name}",
auth=(au, ap))
ok = bool(resp.json().get("success")) if resp.status_code == 200 else False
except Exception:
ok = False
results.append({"name": name, "port": st["ports"][name]["chall"],
"ssh": st["ports"][name]["ssh"], "alive": ok})
# award SLA bonus when all services are UP (throttled, see add_sla_bonus)
alive = sum(1 for r in results if r["alive"])
bonus = orch.add_sla_bonus(idx, alive, len(results))
return {"results": results, "sla": {"alive": alive, "total": len(results),
"pct": round(100 * alive / max(len(results), 1), 1),
"points": orch.get_team_points(idx),
"rank": orch.team_rank(idx), "badge": orch.team_badge(idx),
"bonus": bonus}}
@app.get("/api/team/{idx}/activity")
async def api_team_activity(idx: int, req: Request):
"""Team-scoped activity feed: attack events where this team is attacker or
target (i.e. "ada serangan ke service kita" / "kita menyerang"), plus a
snapshot of own service health. Requires team login + own host."""
td = orch.TEAMS_DIR / f"team{idx}"
if not (td / "state.json").exists():
raise HTTPException(404, "Team not found")
st_self = json.loads((td / "state.json").read_text())
if not _check_team_host(req, st_self):
raise HTTPException(403, "Akses team lain tidak diizinkan")
if not _team_authorized(req, idx):
raise HTTPException(401, "Login portal team dulu")
# live label lookup (rename team = activity updates everywhere)
def label(i):
try:
s = json.loads((orch.TEAMS_DIR / f"team{i}" / "state.json").read_text())
return s.get("label") or f"Team {i}"
except Exception:
return f"Team {i}"
ap = orch.TEAMS_DIR / "attacks.json"
try:
events = json.loads(ap.read_text()).get("events", []) if ap.exists() else []
except Exception:
events = []
mine = []
for e in reversed(events): # newest first
if e.get("attacker") == idx or e.get("target") == idx:
e = dict(e)
e["attacker_label"] = label(e.get("attacker"))
e["target_label"] = label(e.get("target"))
mine.append(e)
return {"events": mine[:100], "team_label": st_self.get("label") or f"Team {idx}"}
@app.post("/api/login")
async def api_login(req: Request):
data = await req.json()
if data.get("user") == ADMIN_USER and data.get("pass") == ADMIN_PASS:
token = os.urandom(16).hex()
_sessions[token] = time.time() + 8 * 3600
resp = JSONResponse({"ok": True})
resp.set_cookie("panel_token", token, httponly=True, samesite="lax", max_age=8 * 3600)
return resp
raise HTTPException(401, "Invalid credentials")
@app.post("/api/logout")
async def api_logout(req: Request):
token = req.cookies.get("panel_token")
if token:
_sessions.pop(token, None)
return {"ok": True}
def require_login(req: Request):
if not _authorized(req):
raise HTTPException(401, "Not authorized")
# ---- receiver proxy endpoints (server-side, keeps admin creds secret) ----
@app.get("/api/challenges")
async def api_challenges(req: Request):
require_login(req)
# Full registry (all sets) with enabled status, plus count of live teams
reg = orch.load_registry()
challs = []
for c in reg.get("challenges", []):
challs.append({
"name": c["name"],
"set": c.get("set"),
"category": c.get("category"),
"desc": c.get("desc"),
"enabled": bool(c.get("enabled")),
"chall_offset": c.get("chall_offset"),
"ssh_offset": c.get("ssh_offset"),
"org_port": c.get("org_port"),
"service_dir": c.get("service_dir"),
})
return {"challenges": challs, "hint": "PATCH /api/challenges/<name> with {\"enabled\": bool} to toggle"}
@app.patch("/api/challenges/{challenge}")
async def api_challenge_toggle(challenge: str, req: Request):
require_login(req)
data = await req.json()
enabled = bool(data.get("enabled"))
reg = orch.load_registry()
found = any(c.get("name") == challenge for c in reg.get("challenges", []))
if not found:
raise HTTPException(404, "Unknown challenge")
changed = orch.set_challenge_enabled(challenge, enabled)
# The registry is the source of truth, but a team's state.json must also
# carry ports/password/flag for the new challenge before the compose can be
# rendered. reconcile_team_state() fills in anything missing.
try:
orch.reconcile_team_state()
except Exception as e:
raise HTTPException(500, f"Rekonsiliasi state tim gagal: {e}")
# apply to live teams (build/up or stop/remove + receiver restart);
# skip rebuild when the flag didn't actually change
if "unchanged" in changed:
return {"ok": True, "name": challenge, "enabled": enabled, "applied": [], "unchanged": True}
# Applying a challenge means a Docker build per team, which takes minutes.
# Run it in the background so the admin UI stays responsive, and let the
# client poll /api/challenges/jobs/<job_id> for the per-team report.
job_id = f"{challenge}-{'on' if enabled else 'off'}-{int(time.time())}"
_TOGGLE_JOBS[job_id] = {
"job": job_id, "name": challenge, "enabled": enabled,
"state": "running", "applied": [], "detail": "queued",
"started": int(time.time()),
}
def _worker():
try:
report = orch.sync_challenge_runtime(challenge, enabled)
_TOGGLE_JOBS[job_id].update(
state="done", applied=report.get("teams", []), detail="complete")
except Exception as e: # surfaced to the client via the job record
_TOGGLE_JOBS[job_id].update(state="error", detail=str(e))
finally:
_TOGGLE_JOBS[job_id]["finished"] = int(time.time())
threading.Thread(target=_worker, name=f"toggle-{job_id}", daemon=True).start()
return {"ok": True, "name": challenge, "enabled": enabled,
"job": job_id, "state": "running"}
@app.get("/api/challenges/jobs/{job_id}")
async def api_challenge_job(job_id: str, req: Request):
"""Poll the result of an async challenge toggle started by PATCH above."""
require_login(req)
job = _TOGGLE_JOBS.get(job_id)
if not job:
raise HTTPException(404, "Unknown job")
return job
@app.get("/api/status")
async def api_status(req: Request):
require_login(req)
results = []
for ch in CHALLENGES:
try:
resp = await _proxy("GET", f"/check/{ch['name']}")
ok = bool(resp.json().get("success")) if resp.status_code == 200 else False
except Exception as e:
ok = False
# read host flag file
flag = ""
try:
fp = Path(f"/opt/gemastik18-final/receiver/flags/{ch['name']}.txt")
if fp.exists():
flag = fp.read_text().strip()
except Exception:
pass
results.append({**ch, "alive": ok, "flag": flag})
return {"results": results, "ts": int(time.time())}
@app.post("/api/flag")
async def api_flag(req: Request):
require_login(req)
data = await req.json()
challenge = data.get("challenge", "")
flag = data.get("flag", "")
if challenge not in [c["name"] for c in CHALLENGES]:
raise HTTPException(400, "Unknown challenge")
if not flag:
raise HTTPException(400, "Flag is empty")
resp = await _proxy("POST", "/flag", {"challenge": challenge, "flag": flag})
return {"receiver_status": resp.status_code, "receiver_body": resp.text}
@app.post("/api/restart/{challenge}")
async def api_restart(challenge: str, req: Request):
require_login(req)
resp = await _proxy("GET", f"/restart/{challenge}")
return {"receiver_status": resp.status_code, "receiver_body": resp.text}
@app.post("/api/rollback/{challenge}")
async def api_rollback(challenge: str, req: Request):
require_login(req)
resp = await _proxy("GET", f"/rollback/{challenge}")
return {"receiver_status": resp.status_code, "receiver_body": resp.text}
@app.post("/api/activate/{challenge}")
async def api_activate(challenge: str, req: Request):
require_login(req)
resp = await _proxy("GET", f"/activate/{challenge}")
return {"receiver_status": resp.status_code, "receiver_body": resp.text}
@app.post("/api/deactivate/{challenge}")
async def api_deactivate(challenge: str, req: Request):
require_login(req)
resp = await _proxy("GET", f"/deactivate/{challenge}")
return {"receiver_status": resp.status_code, "receiver_body": resp.text}
@app.get("/api/credential/{challenge}")
async def api_credential(challenge: str, req: Request, team: int = None):
require_login(req)
# The global receiver on :18080 only knows the 6 native GEMASTIK XVIII
# challenges, so proxying everything there returns "Invalid challenge" for the
# 10 imported XVI/XVII ones -- participants saw no SSH creds at all. A team's
# state.json is the authority for BOTH the password and the SSH login user
# (the same `ssh_user` the panel chpasswds), plus the team-specific port.
# `?team=N` selects the team; team portal hosts imply their own index.
idx = team
if idx is None:
host = (req.headers.get("host") or "").split(":")[0]
for t in orch.all_teams():
dom = (t.get("domain") or "").split(":")[0]
if dom and dom == host:
idx = t.get("index")
break
if idx is not None:
cred = orch.challenge_credential(idx, challenge)
if cred:
return cred
resp = await _proxy("GET", f"/credential/{challenge}")
if resp.status_code == 200:
return resp.json()
return {"error": resp.text}
@app.get("/api/history")
async def api_history(req: Request):
require_login(req)
try:
lines = (BASE_DIR.parent / "history" / "command.txt").read_text().splitlines()
except Exception:
lines = []
return {"lines": lines[-200:]}
# ============ Multi-team orchestrator endpoints ============
@app.get("/api/teams")
async def api_teams(req: Request):
require_login(req)
return {"teams": orch.list_teams()}
@app.post("/api/teams/set")
async def api_teams_set(req: Request):
"""Set/create N teams (idempotent: creates missing, keeps existing).
body: {count, labels: {"1": "Tim Satu"}, domains: {"1": "tim-satu"}}"""
require_login(req)
data = await req.json()
n = int(data.get("count", 0))
if n < 0 or n > 50:
raise HTTPException(400, "Team count must be 0-50")
labels = data.get("labels") or {} # { "1": "Tim Satu", ... }
domains = data.get("domains") or {} # { "1": "mycustom", ... }
created = []
ufw = []
for i in range(1, n + 1):
td = orch.TEAMS_DIR / f"team{i}"
if not td.exists():
label = labels.get(str(i)) or labels.get(i) or f"Tim {i}"
dom = domains.get(str(i)) or domains.get(i) or None
st = orch.create_team(i, label, domain=dom)
created.append(st["index"])
# UFW defaults to deny(incoming) on this host: without these rules
# the team's challenge/SSH/receiver ports are silently blackholed.
ufw.append(orch.sync_team_ufw(i))
else:
# team exists: apply any label/domain overrides
label = labels.get(str(i)) or labels.get(i)
dom = domains.get(str(i)) or domains.get(i)
if label or dom:
orch.update_team(i, label=label, domain=dom)
orch.ensure_team_domains()
return {"created": created, "total": len(orch.list_teams()), "ufw": ufw}
@app.put("/api/teams/{idx}")
async def api_team_update(idx: int, req: Request):
"""Update a team's custom name and/or domain (applies live)."""
require_login(req)
if not (orch.TEAMS_DIR / f"team{idx}" / "state.json").exists():
raise HTTPException(404, "Team not found")
data = await req.json()
try:
st = orch.update_team(idx, label=data.get("label"), domain=data.get("domain"))
return {"ok": True, "team": st}
except FileNotFoundError as e:
raise HTTPException(404, str(e))
@app.delete("/api/teams/{idx}")
async def api_team_delete(idx: int, req: Request):
"""Permanently delete ONE team: containers, network, receiver unit, ports,
directory, score records and its Traefik domain.
Body: {"purge_scores": true} (default) — set false to keep the team's
leaderboard/points history. Destructive and irreversible, so the UI gates
it behind a confirm dialog.
"""
require_login(req)
raw = {}
try:
raw = await req.json()
except Exception:
pass # DELETE with no body is fine
if not (orch.TEAMS_DIR / f"team{idx}" / "state.json").exists():
raise HTTPException(404, f"Team {idx} not found")
try:
# compose down for 16 services takes a while — keep the event loop free
result = await asyncio.to_thread(orch.delete_team, idx,
bool(raw.get("purge_scores", True)))
# refresh the remaining teams' generated artifacts (receiver main.py,
# systemd units) so nothing points at the deleted team
subprocess.run([sys.executable, str(orch.BASE / "panel" / "gen_receiver_services.py"), "start"],
check=False, capture_output=True)
return result
except FileNotFoundError as e:
raise HTTPException(404, str(e))
except Exception as e:
raise HTTPException(500, str(e))
@app.post("/api/teams/bulk-delete")
async def api_teams_bulk_delete(req: Request):
"""Delete SEVERAL teams in one background job.
Body: {"indices": [5, 6], "purge_scores": true}
Why a job and not a loop of DELETE /api/teams/{idx}: a single team takes
~100 s (compose down of 16 services), so deleting four teams inline would
hold the request open for ~7 minutes and trip every proxy/browser timeout
in front of the panel. Each team is therefore deleted sequentially inside
ONE background thread, and the client polls a single job id.
Teams are processed one at a time on purpose. delete_team() runs
`docker compose -p teamN down` and regenerates shared artifacts
(receiver main.py, systemd units) afterwards, so running several in
parallel would race on those shared files.
A team that fails does NOT abort the rest: the job records the error and
moves on, because the point of a bulk delete is to clear stale teams and
one broken compose shouldn't strand the others.
"""
require_login(req)
data = await req.json()
raw = data.get("indices") or data.get("indices[]") or []
try:
indices = sorted({int(i) for i in raw})
except (TypeError, ValueError):
raise HTTPException(400, "indices must be a list of team numbers")
if not indices:
raise HTTPException(400, "No team selected")
if len(indices) > 20:
raise HTTPException(400, "Refusing to delete more than 20 teams at once")
purge = bool(data.get("purge_scores", True))
existing = [i for i in indices
if (orch.TEAMS_DIR / f"team{i}" / "state.json").exists()]
skipped = [i for i in indices if i not in existing]
if not existing:
raise HTTPException(404, "None of the selected teams exist")
job_id = f"bulkdel-{int(time.time())}-{len(existing)}"
_DELETE_JOBS[job_id] = {
"job": job_id, "indices": existing, "skipped": skipped,
"state": "running", "done": [], "errors": {},
"detail": "queued", "started": int(time.time()),
}
def _worker():
job = _DELETE_JOBS[job_id]
try:
for n, i in enumerate(existing, 1):
job["detail"] = f"menghapus team {i} ({n}/{len(existing)})"
try:
# delete_team is blocking (subprocess + shutil), and this
# runs in a plain thread, so call it directly.
res = orch.delete_team(i, purge)
job["done"].append({
"team": i,
"label": res.get("label", f"Team {i}"),
"steps": res.get("steps", []),
"purged": res.get("purged", {}),
})
except Exception as e:
job["errors"][str(i)] = str(e)
# regenerate shared artifacts once at the end, so the remaining
# teams' receiver main.py / systemd units stop referencing deleted ones
subprocess.run(
[sys.executable, str(orch.BASE / "panel" / "gen_receiver_services.py"), "start"],
check=False, capture_output=True)
job["state"] = "done" if not job["errors"] else "partial"
job["detail"] = "complete" if not job["errors"] else "completed with errors"
except Exception as e:
job["state"] = "error"
job["detail"] = str(e)
finally:
job["finished"] = int(time.time())
threading.Thread(target=_worker, name=f"bulkdel-{job_id}", daemon=True).start()
return {"ok": True, "job": job_id, "state": "running",
"indices": existing, "skipped": skipped}
@app.get("/api/teams/bulk-delete/{job_id}")
async def api_teams_bulk_delete_job(job_id: str, req: Request):
"""Poll a bulk team delete started by POST /api/teams/bulk-delete."""
require_login(req)
job = _DELETE_JOBS.get(job_id)
if not job:
raise HTTPException(404, "Unknown job")
return job
@app.post("/api/teams/{idx}/ufw")
async def api_team_ufw(idx: int, req: Request):
"""Reconcile UFW rules for a team's port block.
UFW defaults to deny(incoming) on this host, so a team whose ports were
never opened is blackholed. Use this after creating a team out-of-band, or
to close the ports of a team you just deleted by hand.
Body: {"remove": true} deletes the rules instead.
"""
require_login(req)
raw = {}
try:
raw = await req.json()
except Exception:
pass
return await asyncio.to_thread(orch.sync_team_ufw, idx, bool(raw.get("remove", False)))
@app.post("/api/teams/start")
async def api_teams_start(req: Request):
require_login(req)
data = await req.json()
idx = data.get("index")
if idx is None:
# start all
results = []
for t in orch.list_teams():
try:
results.append({"team": t["index"], "ok": True})
orch.start_team(t["index"])
except Exception as e:
results.append({"team": t["index"], "ok": False, "err": str(e)})
return {"results": results}
try:
st = orch.start_team(int(idx))
return {"ok": True, "team": st}
except Exception as e:
raise HTTPException(500, str(e))
@app.post("/api/teams/stop")
async def api_teams_stop(req: Request):
require_login(req)
data = await req.json()
idx = data.get("index")
if idx is None:
results = []
for t in orch.list_teams():
try:
orch.stop_team(t["index"])
results.append({"team": t["index"], "ok": True})
except Exception as e:
results.append({"team": t["index"], "ok": False, "err": str(e)})
return {"results": results}
try:
st = orch.stop_team(int(idx))
return {"ok": True, "team": st}
except Exception as e:
raise HTTPException(500, str(e))
@app.get("/api/teams/{idx}/logs")
async def api_team_logs(idx: int, req: Request, service: Optional[str] = None, tail: int = 100):
require_login(req)
try:
logs = orch.team_logs(idx, service, tail)
return {"team": idx, "logs": logs}
except Exception as e:
raise HTTPException(500, str(e))
@app.get("/api/teams/{idx}/creds")
async def api_team_creds(idx: int, req: Request):
require_login(req)
try:
td = orch.TEAMS_DIR / f"team{idx}"
st = json.loads((td / "state.json").read_text())
return {"team": st}
except Exception as e:
raise HTTPException(404, str(e))
@app.get("/api/topology")
async def api_topology(req: Request):
"""Return topology graph data (nodes + edges) for the UI."""
require_login(req)
teams = orch.list_teams()
nodes = [
{"id": "panel", "label": "Panel A/D", "type": "panel", "url": "https://panel.attackdefense.imrnes.team"},
{"id": "traefik", "label": "Traefik / Coolify", "type": "infra"},
{"id": "dns", "label": "*.imrnes.team → 43.134.105.109", "type": "infra"},
]
edges = [{"from": "dns", "to": "traefik"}, {"from": "traefik", "to": "panel"}]
for t in teams:
nid = f"team{t['index']}"
nodes.append({
"id": nid, "label": t.get("label", f"Team {t['index']}"),
"type": "team", "index": t["index"], "status": t.get("status", "unknown"),
"receiver_port": t["ports"]["receiver"], "ssh_pass": t.get("ssh_pass", ""),
})
edges.append({"from": "traefik", "to": nid, "label": f":{t['ports']['receiver']}"})
for name, coff, soff in orch.CHALLENGES:
cn = f"team{t['index']}-{name}"
nodes.append({"id": cn, "label": f"{name}", "type": "challenge",
"port": t["ports"][name]["chall"], "ssh": t["ports"][name]["ssh"]})
edges.append({"from": nid, "to": cn, "label": f":{t['ports'][name]['chall']}"})
return {"nodes": nodes, "edges": edges}
# ============ Flag randomization + team submission ============
@app.post("/api/teams/{idx}/randomize")
async def api_randomize(idx: int, req: Request):
"""Randomize all flags for a team (recreates containers to pick up new flags)."""
require_login(req)
try:
mapping = orch.randomize_flags(idx)
return {"ok": True, "team": idx, "flags": mapping}
except Exception as e:
raise HTTPException(500, str(e))
@app.post("/api/reset/scores")
async def api_reset_scores(req: Request):
"""Admin: wipe leaderboard (all solves)."""
require_login(req)
return orch.reset_scores()
@app.post("/api/reset/environment")
async def api_reset_environment(req: Request):
"""Admin: full environment reset (stop all, remove teams/containers/receivers)."""
require_login(req)
return orch.reset_environment()
@app.post("/api/flag/submit")
async def api_flag_submit(req: Request):
"""Public endpoint: teams submit flags. No login needed.
body: {team: attacker, target?: victim, challenge, flag}"""
data = await req.json()
team = int(data.get("team", 0)) # attacker (tim yang submit)
target = int(data.get("target") or 0) or team # victim team (flag dicuri dari sini)
chall = data.get("challenge", "")
flag = data.get("flag", "").strip()
team_name = data.get("team_name", "").strip()[:64] or f"Team {team}"
if team <= 0:
return {"success": False, "error": "Select your team"}
if chall not in [c[0] for c in orch.CHALLENGES]:
return {"success": False, "error": "Challenge not found"}
if not flag:
return {"success": False, "error": "Flag is required"}
return orch.submit_flag(target, chall, flag, attacker_idx=team, attacker_name=team_name)
@app.get("/api/attacks")
async def api_attacks(req: Request):
"""Admin: recent attack events (attacker -> target) for the topology visualizer."""
require_login(req)
ap = orch.TEAMS_DIR / "attacks.json"
if ap.exists():
log = json.loads(ap.read_text())
else:
log = {"events": []}
return {"events": log.get("events", [])}
@app.get("/api/leaderboard")
async def api_leaderboard(req: Request):
"""Leaderboard of solves so far. Team names resolved LIVE from state.json
so renaming a team updates leaderboard + topology everywhere."""
lb_path = orch.TEAMS_DIR / "leaderboard.json"
if lb_path.exists():
lb = json.loads(lb_path.read_text())
else:
lb = {"solves": []}
# live name lookup: state.json label fallback to snapshot
def team_name(idx):
try:
st = json.loads((orch.TEAMS_DIR / f"team{idx}" / "state.json").read_text())
return st.get("label") or f"Team {idx}"
except Exception:
return f"Team {idx}"
# aggregate per team
teams = {}
for e in lb["solves"]:
name = team_name(e["team"])
t = teams.setdefault(e["team"], {"team": e["team"], "name": name, "solves": 0, "challs": [], "points": 0})
t["name"] = name
t["solves"] += 1
t["challs"].append(e["challenge"])
# attach live points from points.json
for tid, t in teams.items():
t["points"] = orch.get_team_points(tid)
return {"solves": lb["solves"], "teams": sorted(teams.values(), key=lambda x: (-x["points"], -x["solves"]))}
@app.get("/api/scoreboard")
async def api_scoreboard(req: Request):
"""Admin + team: full scoreboard with points, SLA, rank, badges."""
require_login(req)
return orch.sla_status_all()
@app.get("/api/public/scoreboard")
async def api_public_scoreboard():
"""Public scoreboard (no login) — used by the team portal status tab."""
d = orch.sla_status_all()
# strip receiver creds / ports internals for the public view
for t in d["teams"]:
t.pop("domain", None)
return d
@app.get("/api/public/teams")
async def api_public_teams():
"""Public list of team names (for the submit dropdown)."""
return {"teams": [{"index": t["index"], "label": t.get("label", f"Team {t['index']}")} for t in orch.list_teams()]}
@app.get("/api/targets")
async def api_admin_targets(req: Request):
"""Admin: matrix of every team's service domain:port (public targets)."""
require_login(req)
out = []
for d in sorted(orch.TEAMS_DIR.glob("team*")):
if not (d / "state.json").exists():
continue
st = json.loads((d / "state.json").read_text())
dom = st.get("domain") or (st.get("slug", f"team{st.get('index')}") + ".attackdefense.imrnes.team")
for name, coff, soff in orch.CHALLENGES:
p = st["ports"].get(name)
if not p:
continue
out.append({
"team": st.get("index"),
"team_label": st.get("label", f"Team {st.get('index')}"),
"challenge": name,
"domain": dom,
"port": p["chall"],
})
return {"targets": out}
# ============ WebSocket SSH terminal (team portal) ============
import paramiko
import websockets
@app.websocket("/api/team/{idx}/ssh/ws")
async def team_ssh_ws(ws: WebSocket, idx: int):
chall = ws.query_params.get("chall", "")
# auth: team session cookie must match this team
token = ws.cookies.get("team_token")
e = _team_sessions.get(token or "")
if not e or e[0] != idx or e[1] < time.time():
await ws.close(code=4001, reason="unauthorized")
return
td = orch.TEAMS_DIR / f"team{idx}"
st = json.loads((td / "state.json").read_text())
# host check (IDOR): only this team's domain can open its SSH
host = (ws.headers.get("host") or "").split(":")[0]
if not (host == st.get("domain") or host.startswith("127.0.0.1") or host.startswith("localhost")):
await ws.close(code=4003, reason="wrong host")
return
if chall not in st.get("ports", {}):
await ws.close(code=4002, reason="unknown challenge")
return
recv_port = st["ports"][chall]["ssh"]
# The SSH login is PER-CHALLENGE, not per-team. Only the 6 native GEMASTIK
# XVIII images provision `ctfuser`; every imported XVI/XVII image does
# `RUN echo root:${PASSWORD} | chpasswd`. Reading st["ssh_user"] (a single
# team-wide value, default ctfuser) made the web terminal log in as ctfuser
# for all 16 challenges, so 10 of them always failed with "Permission denied".
# challenge_ssh_users() reads the registry -- the same field set_ssh_passwords()
# chpasswds -- so the login and the password can never drift apart.
user = orch.challenge_ssh_users().get(chall) or st.get("ssh_user", "ctfuser")
# each challenge container has its own password (chall_passwords);
# ssh_pass is the portal login password (may differ).
pw = st.get("chall_passwords", {}).get(chall) or st.get("ssh_pass", "")
await ws.accept()
chan = client = None
try:
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None, lambda: client.connect("127.0.0.1", port=recv_port, username=user,
password=pw, timeout=10, allow_agent=False,
look_for_keys=False))
chan = client.invoke_shell(term="xterm-256color", width=120, height=32)
async def ws_to_ssh():
while True:
try:
msg = await ws.receive_text()
except Exception:
break
if msg.startswith("__resize__"):
try:
_, cols, rows = msg.split(":", 2)
chan.resize_pty(int(cols), int(rows))
except Exception:
pass
else:
try:
chan.send(msg)
except Exception:
break
async def ssh_to_ws():
# non-blocking poll: chan.recv() di dalam async task akan
# memblokir seluruh event loop (deadlock) — jadi poll recv_ready.
while True:
try:
if chan.recv_ready():
data = chan.recv(4096)
if not data:
break
await ws.send_text(data.decode("utf-8", "replace"))
elif chan.closed:
break
except Exception:
break
await asyncio.sleep(0.03)
t1 = asyncio.create_task(ws_to_ssh())
t2 = asyncio.create_task(ssh_to_ws())
done, pending = await asyncio.wait({t1, t2}, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()
except Exception as e:
try:
await ws.send_text(f"\r\n[ssh error] {e}\r\n")
except Exception:
pass
finally:
try:
if chan: chan.close()
except Exception: pass
try:
if client: client.close()
except Exception: pass
try:
await ws.close()
except Exception: pass
Reference in New Issue View Git Blame Copy Permalink