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

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.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:]}