Cyrene
ae50acfe40
Passwords failed on 10/16 challenges while state.json looked correct:
- only the 6 native GEMASTIK XVIII images provision 'ctfuser'; every imported
XVI/XVII image does 'echo root:${PASSWORD} | chpasswd' and logs in as root.
set_ssh_passwords() hardcoded ctfuser, so chpasswd set a password on an
account nobody uses -> 'Permission denied' everywhere.
Registry gains a per-challenge 'ssh_user'; chpasswd now targets the real
login (and ctfuser/ctf when present) and reports failures loudly.
- phew checker: chall.py block-buffers stdout through the docker exec pipe
(PYTHONUNBUFFERED now set) and leaks chall.py inside the container on
timeout (26 orphans, container saturated) -> reaps the whole exec process
group. Startup does a fresh Pailier keygen (~12 s) so crypto reads need
_CRYPTO_TIMEOUT, not the 5 s prompt default.
Adds panel/verify_ssh_creds.py (proves the state->container binding from
inside via a real login), audit_ssh_users.sh, reset_runtime.sh.
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Probe each team receiver's /check/<challenge> directly and report SLA.
|
|
|
|
Probes are SEQUENTIAL per team on purpose: the team receivers are sync Flask
|
|
apps, so 8 concurrent /check requests make them time out and report false
|
|
failures. Keep max_workers=1. This is still far faster than the panel's
|
|
/api/scoreboard, which probes every team on one shared refresher thread.
|
|
|
|
python3 panel/sla_probe.py # all teams
|
|
python3 panel/sla_probe.py 1 2 # specific teams
|
|
"""
|
|
import base64
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import teams as orch
|
|
|
|
def check(recv_port, au, ap, name):
|
|
url = f"http://127.0.0.1:{recv_port}/check/{name}"
|
|
tok = base64.b64encode(f"{au}:{ap}".encode()).decode()
|
|
req = urllib.request.Request(url, headers={"Authorization": f"Basic {tok}"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
body = json.loads(r.read().decode())
|
|
return name, bool(body.get("success")), ""
|
|
except urllib.error.HTTPError as e:
|
|
return name, False, f"HTTP {e.code}"
|
|
except Exception as e:
|
|
return name, False, str(e)[:60]
|
|
|
|
def main():
|
|
want = [int(a) for a in sys.argv[1:]]
|
|
teams = [t for t in orch.list_teams() if not want or t["index"] in want]
|
|
enabled = [c["name"] for c in orch.enabled_challenges()]
|
|
grand_ok = grand_all = 0
|
|
for t in teams:
|
|
idx = t["index"]
|
|
port = t["ports"]["receiver"]
|
|
au, ap = t.get("admin_user", ""), t.get("admin_pass", "")
|
|
res = [check(port, au, ap, n) for n in enabled]
|
|
up = [n for n, ok, _ in res if ok]
|
|
down = [(n, e) for n, ok, e in res if not ok]
|
|
grand_ok += len(up); grand_all += len(res)
|
|
print(f"team{idx} ({t.get('label')}) SLA {len(up)}/{len(res)}")
|
|
if down:
|
|
for n, e in down:
|
|
print(f" DOWN {n}: {e}")
|
|
print(f"\nTOTAL {grand_ok}/{grand_all} "
|
|
f"({100.0 * grand_ok / grand_all if grand_all else 0:.1f}%)")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|