π Panduan SSH & Attack
@@ -254,6 +265,7 @@ function showView(v) {
if (v === 'targets') loadTargets();
if (v === 'activity') loadActivity();
if (v === 'leaderboard') loadLeaderboard();
+ if (v === 'sla') loadSla();
}
// ---------- auth ----------
@@ -425,6 +437,41 @@ async function loadActivity() {
}
}
+// ---------- SLA & scoreboard (public endpoint) ----------
+async function loadSla() {
+ const box = document.getElementById('slaList');
+ try {
+ const d = await api('/api/public/scoreboard');
+ const teams = d.teams || [];
+ if (!teams.length) {
+ box.innerHTML = '
Belum ada data SLA.
';
+ return;
+ }
+ box.innerHTML = teams.map(t => {
+ const isMe = t.team === TEAM_ID;
+ const pct = Math.round(t.sla_pct || 0);
+ const color = pct === 100 ? '#2ecc71' : (pct >= 60 ? '#f1c40f' : '#e74c3c');
+ const bar = `
`;
+ return `
+ ${esc(t.badge || t.rank)}
+ ${esc(t.label)} ${isMe ? '(kamu)' : ''}
+
+ ${t.alive}/${t.total} (${pct}%)
+
β ${t.solves} flag Β· π ${t.points} pts
+
+
${bar}`;
+ }).join('');
+ } catch (e) {
+ box.innerHTML = '
Gagal memuat SLA.
';
+ }
+}
+
+// auto-refresh SLA while its tab is open (every 30s)
+setInterval(async () => {
+ const s = document.getElementById('view-sla');
+ if (s && s.classList.contains('active')) loadSla();
+}, 30000);
+
// ---------- leaderboard (public endpoint) ----------
async function loadLeaderboard() {
const box = document.getElementById('lbList');
@@ -442,7 +489,7 @@ async function loadLeaderboard() {
return `
${rank}
${esc(t.name)} ${isMe ? '(kamu)' : ''}
- β ${t.solves} solve${t.solves > 1 ? 's' : ''}
+ β ${t.solves} solve${t.solves > 1 ? 's' : ''} Β· π ${t.points ?? 0} pts
`;
}).join('');
const found = teams.some(t => t.team === TEAM_ID);
diff --git a/panel/teams.py b/panel/teams.py
index fc31491..b7448d2 100644
--- a/panel/teams.py
+++ b/panel/teams.py
@@ -46,6 +46,88 @@ CHALLENGES = [
("warmup", 5, 27),
]
+# 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)
+ if alive >= SLA_BONUS_MIN_ALIVE and total >= SLA_BONUS_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
@@ -446,7 +528,12 @@ def submit_flag(target_idx: int, chall: str, flag: str,
and e.get("target") == target_idx for e in lb["solves"]):
lb["solves"].append(entry)
lb_path.write_text(json.dumps(lb, indent=2))
- return {"success": True, "team": attacker_idx, "target": target_idx, "challenge": chall}
+ # 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)."""
@@ -469,9 +556,10 @@ def _log_attack(attacker_team: int, target_team: int, target_chall: str, flag_hi
pass
def reset_scores() -> dict:
- """Wipe the leaderboard (all solves removed)."""
+ """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:
@@ -512,6 +600,130 @@ def reset_environment() -> dict:
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"