attack-defense-platform/panel/main.py

#!/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 httpx
from pathlib import Path
from fastapi import FastAPI, Request, HTTPException
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")

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):
    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())

@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)
    return {"challenges": CHALLENGES}

@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):
    require_login(req)
    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)."""
    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")
    created = []
    for i in range(1, n + 1):
        td = orch.TEAMS_DIR / f"team{i}"
        if not td.exists():
            st = orch.create_team(i, data.get("label_prefix", "Tim"))
            created.append(st["index"])
    return {"created": created, "total": len(orch.list_teams())}

@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.gemastik.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/flag/submit")
async def api_flag_submit(req: Request):
    """Public endpoint: teams submit flags. No login needed."""
    data = await req.json()
    team = int(data.get("team", 0))
    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(team, chall, flag, team_name)


@app.get("/api/leaderboard")
async def api_leaderboard(req: Request):
    """Leaderboard of solves so far."""
    lb_path = orch.TEAMS_DIR / "leaderboard.json"
    if lb_path.exists():
        lb = json.loads(lb_path.read_text())
    else:
        lb = {"solves": []}
    # aggregate per team
    teams = {}
    for e in lb["solves"]:
        t = teams.setdefault(e["team"], {"team": e["team"], "name": e["team_name"], "solves": 0, "challs": []})
        t["solves"] += 1
        t["challs"].append(e["challenge"])
    return {"solves": lb["solves"], "teams": sorted(teams.values(), key=lambda x: -x["solves"])}


@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()]}