815 lines
36 KiB
Python
815 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
CWF-DR 调度器 —— TSec Benchmark 自动化跑分的确定性编排脚本
|
||
============================================================
|
||
替代 tec_benchmark 编排 agent，用确定性算法调度 3 个靶场名额，
|
||
把「编排」从解题里剥离，让难题拿到不中断的长连续窗口。
|
||
|
||
方案全称：Continuous-Window Feasibility with Dynamic Reflow
|
||
· 双记账保证不超 6h / 1080 槽·分钟（3 槽 × 360min）
|
||
· 阶段1：54 易题三槽滚动收割，跑到解出即回收，看门狗防死槽
|
||
· 阶段2：9 硬题连续长跑，事件驱动余量瀑布回流
|
||
- b 族(b01/b02/b03)：按历史耗时定制 floor、只延不切、深链延窗预留
|
||
- 6 赌注题：等额起步(自由池÷剩余赌注题数)，按进度信号动态加码
|
||
· 墙钟可行性闸：任何窗口 ≤ 该槽到软截止能连续消费的量（防幽灵预算）
|
||
|
||
用法：
|
||
# 真实跑分（需 VPN 已连、ARTEX 已 seed、平台 token 就绪）
|
||
BENCHMARK_TOKEN=xxx BENCHMARK_BASE_URL=http://platform \\
|
||
ADMIN_PASSWORD=artexbench python3 scheduler.py
|
||
|
||
# 本地模拟（不碰靶场/平台/ARTEX，用历史耗时演练调度逻辑）
|
||
python3 scheduler.py --simulate # 顺利走法
|
||
python3 scheduler.py --simulate --scenario unlucky # 不顺利走法
|
||
|
||
所有阈值见下方 CONFIG，可按实际网络状况调档。
|
||
"""
|
||
import os
|
||
import sys
|
||
import csv
|
||
import json
|
||
import time
|
||
import argparse
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime
|
||
|
||
# ============================ CONFIG（可调） ============================
|
||
TOTAL_BUDGET = 1080.0 # 槽·分钟总预算 = 3 槽 × 360min（硬天花板）
|
||
SLOTS = 3 # 平台硬上限：同时最多 3 个活跃容器
|
||
DEADLINE_MIN = 360.0 # 6h 硬停
|
||
SOFT_DEADLINE = 350.0 # 软截止：此后不新起长窗，留 10min 优雅收尾
|
||
POLL_SEC = 20 # 轮询周期（真实模式）
|
||
|
||
# —— 易题阶段 ——
|
||
EASY_WALL_CAP = 95.0 # 墙钟到此强制收口、剩余易题 park、全槽转硬
|
||
EASY_SPENT_CAP = 260.0 # 易题累计吃到此槽·分钟同样强制收口
|
||
EASY_STALL_MIN = 22.0 # 易题连续无进展 → 干净重启一次
|
||
EASY_ABS_KILL = 30.0 # 易题绝对上限 → park 让槽
|
||
EASY_SOFT_BUDGET = 240.0 # 易题软预算，省下的回流给硬题
|
||
|
||
# —— 硬题阶段 ——
|
||
W_MIN = 70.0 # 赌注题最小有效窗，低于此不新起（转并行/跳过）
|
||
RESTART_RESERVE = 60.0 # 干净重启应急池
|
||
HARD_STALL_MIN = 30.0 # 硬题连续无进展 → 触发看门狗（换思路/放弃）
|
||
B_CAP_MULT = 2.0 # b 族单题窗封顶 = B_CAP_MULT × floor
|
||
GAMBLE_HARD_CAP = 130.0 # 单道赌注题累计封顶（防病态占用）
|
||
PROGRESS_SIGNAL_BONUS = 1.6 # 冒出 foothold/半flag 信号的赌注题权重加成
|
||
|
||
# 9 硬题（用户指定）。b 族按历史耗时+margin 定制 floor；6 赌注题等额起步。
|
||
B_FLOOR = {"b-01": 75.0, "b-02": 110.0, "b-03": 88.0}
|
||
# hardQ 顺序 = 波1 b族 → 波2 最需长窗的赌注题 → 波3 其余（历史碎片较短的）
|
||
HARD_ORDER = ["b-01", "b-02", "b-03", # 波1：高确定性、早解早释放余量
|
||
"f2-05", "e3-04", "a-18", # 波2：最需要长连续窗，早起才拿得到
|
||
"a-03", "a-13", "a-16"] # 波3：历史碎片较短、需求相对短
|
||
B_FAMILY = set(B_FLOOR)
|
||
GAMBLES = set(HARD_ORDER) - B_FAMILY
|
||
HARD_SET = set(HARD_ORDER)
|
||
|
||
# 家族取题优先级（易题阶段选题用；同族无技术关联，仅作性价比排序）
|
||
FAMILY_PRIORITY = ["c", "d", "e1", "a", "e2", "b", "e3", "f1", "f2"]
|
||
|
||
# ARTEX / 平台接入
|
||
ARTEX_API = os.environ.get("ARTEX_API", "http://127.0.0.1:8787")
|
||
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "artexbench")
|
||
BENCH_TOKEN = os.environ.get("BENCHMARK_TOKEN", "")
|
||
BENCH_BASE = os.environ.get("BENCHMARK_BASE_URL", "").rstrip("/")
|
||
VPN_HEALTH_URL = os.environ.get("VPN_HEALTH_URL", "http://10.0.100.58")
|
||
PLAN_PROFILE_ID = os.environ.get("PLAN_PROFILE_ID") # 可选：给硬题任务 pin 强模型
|
||
EVENTS_CSV = os.environ.get("EVENTS_CSV", "scheduler-events.csv")
|
||
|
||
# ============================ 日志 ============================
|
||
class EventLog:
|
||
"""事件 CSV，列对齐历史 run-*.csv 便于回归对比。"""
|
||
COLS = ["时间", "事件类型", "关联题目", "题目编码", "附加信息"]
|
||
|
||
def __init__(self, path, clock):
|
||
self.path = path
|
||
self.clock = clock
|
||
self.fh = open(path, "w", newline="", encoding="utf-8-sig")
|
||
self.w = csv.writer(self.fh)
|
||
self.w.writerow(self.COLS)
|
||
self.fh.flush()
|
||
|
||
def emit(self, kind, code="", info=""):
|
||
ts = self.clock.wallclock_str()
|
||
self.w.writerow([ts, kind, code, code, info])
|
||
self.fh.flush()
|
||
print(f"[{self.clock.now():6.1f}m] {kind:6s} {code:7s} {info}", file=sys.stderr)
|
||
|
||
def close(self):
|
||
self.fh.close()
|
||
|
||
|
||
# ============================ 时钟（真实/模拟统一抽象） ============================
|
||
class RealClock:
|
||
def __init__(self):
|
||
self.t0 = time.monotonic()
|
||
self.wall0 = datetime.now()
|
||
|
||
def now(self): # 距开始的分钟数
|
||
return (time.monotonic() - self.t0) / 60.0
|
||
|
||
def sleep(self, sec):
|
||
time.sleep(sec)
|
||
|
||
def wallclock_str(self):
|
||
return datetime.now().strftime("%Y/%m/%d %H:%M:%S")
|
||
|
||
|
||
class VirtualClock:
|
||
"""模拟模式：虚拟分钟，每 tick 步进 POLL，不真正 sleep。"""
|
||
def __init__(self, step_min):
|
||
self._now = 0.0
|
||
self.step = step_min
|
||
|
||
def now(self):
|
||
return self._now
|
||
|
||
def sleep(self, sec):
|
||
self._now += self.step # 虚拟推进
|
||
|
||
def wallclock_str(self):
|
||
base = datetime(2026, 1, 1)
|
||
from datetime import timedelta
|
||
return (base + timedelta(minutes=self._now)).strftime("%Y/%m/%d %H:%M:%S")
|
||
|
||
|
||
# ============================ HTTP 助手 ============================
|
||
def _http(method, url, headers=None, body=None, timeout=30):
|
||
data = None
|
||
headers = dict(headers or {})
|
||
if body is not None:
|
||
data = json.dumps(body).encode()
|
||
headers.setdefault("Content-Type", "application/json")
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
raw = r.read().decode()
|
||
return r.status, (json.loads(raw) if raw.strip() else None)
|
||
except urllib.error.HTTPError as e:
|
||
raw = e.read().decode()
|
||
try:
|
||
return e.code, json.loads(raw)
|
||
except Exception:
|
||
return e.code, {"error": raw}
|
||
except Exception as e:
|
||
return 0, {"error": str(e)}
|
||
|
||
|
||
# ============================ 平台客户端 ============================
|
||
class BenchEnded(Exception):
|
||
"""跑分任务已结束（token 失效 task_not_found / 总时限到 invalid_state）→ 全局停机。"""
|
||
|
||
|
||
class BenchClient:
|
||
def __init__(self):
|
||
self.h = {"BENCHMARK_TOKEN": BENCH_TOKEN}
|
||
|
||
def vpn_ok(self):
|
||
st, body = _http("GET", VPN_HEALTH_URL, timeout=10)
|
||
return st == 200 and isinstance(body, dict) and body.get("status") == "ok"
|
||
|
||
def challenges(self):
|
||
st, body = _http("GET", f"{BENCH_BASE}/openapi/v1/challenges", self.h)
|
||
if isinstance(body, list):
|
||
return body
|
||
# 非数组=错误响应：token 失效或总时限到 → 任务结束，抛出让主循环停机
|
||
code = (body or {}).get("code") if isinstance(body, dict) else None
|
||
if code in ("task_not_found", "invalid_state"):
|
||
raise BenchEnded(code)
|
||
return [] # 其它瞬时错误（网络抖动等）：当空处理，下一轮重试
|
||
|
||
def start(self, code):
|
||
# 返回 (http_status, error_code|None, container_addr_list)
|
||
st, body = _http("POST", f"{BENCH_BASE}/openapi/v1/challenges/start?unique_code={code}", self.h)
|
||
ecode = (body or {}).get("code") if isinstance(body, dict) else None
|
||
addrs = (body or {}).get("container_addr") if isinstance(body, dict) else None
|
||
return st, ecode, (addrs if isinstance(addrs, list) else [])
|
||
|
||
def close(self, code):
|
||
return _http("POST", f"{BENCH_BASE}/openapi/v1/challenges/close?unique_code={code}", self.h)
|
||
|
||
def hint(self, code):
|
||
return _http("GET", f"{BENCH_BASE}/openapi/v1/challenges/hint?unique_code={code}", self.h)
|
||
|
||
|
||
# ============================ ARTEX 客户端 ============================
|
||
class ArtexClient:
|
||
def __init__(self):
|
||
self.token = None
|
||
|
||
def login(self):
|
||
st, body = _http("POST", f"{ARTEX_API}/api/auth/init", body={"password": ADMIN_PASSWORD})
|
||
tok = (body or {}).get("token") if isinstance(body, dict) else None
|
||
if not tok:
|
||
st, body = _http("POST", f"{ARTEX_API}/api/auth/login",
|
||
body={"username": "ARTEX", "password": ADMIN_PASSWORD})
|
||
tok = (body or {}).get("token") if isinstance(body, dict) else None
|
||
self.token = tok
|
||
return bool(tok)
|
||
|
||
def _auth(self):
|
||
return {"Authorization": f"Bearer {self.token}"}
|
||
|
||
def create_task(self, description, goal):
|
||
payload = {"description": description, "goal": goal, "seed_first_intent": True}
|
||
if PLAN_PROFILE_ID:
|
||
payload["llm_profile_id"] = int(PLAN_PROFILE_ID)
|
||
st, body = _http("POST", f"{ARTEX_API}/api/tasks", self._auth(), payload)
|
||
return (body or {}).get("id") if isinstance(body, dict) else None
|
||
|
||
def control(self, task_id, action): # action = pause | resume
|
||
return _http("POST", f"{ARTEX_API}/api/tasks/{task_id}/control", self._auth(), {"action": action})
|
||
|
||
def delete_task(self, task_id):
|
||
return _http("DELETE", f"{ARTEX_API}/api/tasks/{task_id}", self._auth())
|
||
|
||
def last_activity(self):
|
||
"""返回 {task_id: last_activity_epoch}，用于进度看门狗。"""
|
||
st, body = _http("GET", f"{ARTEX_API}/api/tasks", self._auth())
|
||
out = {}
|
||
if isinstance(body, dict):
|
||
for t in body.get("tasks", []):
|
||
la = t.get("last_activity") or t.get("lastActivity") or 0
|
||
out[t.get("id")] = _to_epoch(la)
|
||
return out
|
||
|
||
def finding_count(self, task_id):
|
||
st, body = _http("GET", f"{ARTEX_API}/api/exploration/findings?task={task_id}", self._auth())
|
||
return len(body) if isinstance(body, list) else 0
|
||
|
||
|
||
def _to_epoch(v):
|
||
if isinstance(v, (int, float)):
|
||
return float(v)
|
||
if isinstance(v, str) and v:
|
||
for fmt in ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S%z"):
|
||
try:
|
||
return datetime.strptime(v, fmt).timestamp()
|
||
except Exception:
|
||
pass
|
||
return 0.0
|
||
|
||
|
||
# ============================ 槽位状态 ============================
|
||
class Slot:
|
||
def __init__(self, idx):
|
||
self.idx = idx
|
||
self.reset()
|
||
|
||
def reset(self):
|
||
self.code = None
|
||
self.task_id = None
|
||
self.addr = None
|
||
self.kind = None # easy | b | gamble
|
||
self.started = None # 起跑墙钟(min)
|
||
self.window = 0.0 # 已分配窗口(min)；easy 不用
|
||
self.banked_flags = 0
|
||
self.last_progress = None # 上次进展墙钟(min)
|
||
self.last_finding = 0
|
||
self.last_activity = 0.0
|
||
self.restarted = False
|
||
|
||
@property
|
||
def busy(self):
|
||
return self.code is not None
|
||
|
||
def elapsed(self, now):
|
||
return (now - self.started) if self.started is not None else 0.0
|
||
|
||
|
||
# ============================ 调度器核心 ============================
|
||
class Scheduler:
|
||
def __init__(self, bench, artex, clock, simulate=False, sim=None):
|
||
self.b = bench
|
||
self.a = artex
|
||
self.clock = clock
|
||
self.simulate = simulate
|
||
self.sim = sim
|
||
self.log = EventLog(EVENTS_CSV, clock)
|
||
self.slots = [Slot(i) for i in range(SLOTS)]
|
||
self.easyQ = [] # 待跑易题 code
|
||
self.deferQ = [] # 被 park 的易题
|
||
self.blacklist = set()
|
||
self.given_up = set()
|
||
self.solved = set()
|
||
self.hard_started = set()
|
||
self.spent_closed = 0.0 # 已关闭工作累计的真实槽·分钟
|
||
self.easy_spent = 0.0
|
||
self.easy_done = False
|
||
self.flags = {} # code -> correct_flag_count（用于检测新 flag / b 族里程碑）
|
||
self.flag_total = {} # code -> flag_count
|
||
self.diff = {} # code -> difficulty（easy/medium/hard，用于同族内排序）
|
||
self._cmap = {} # 本轮 challenges 快照
|
||
|
||
# ---------- 预算记账 ----------
|
||
def live_spent(self, now):
|
||
return self.spent_closed + sum(s.elapsed(now) for s in self.slots if s.busy)
|
||
|
||
def wall_cap(self, slot, now):
|
||
"""该槽从现在到软截止能连续消费的最大窗口（含已跑部分）。防幽灵预算。"""
|
||
return slot.elapsed(now) + max(0.0, SOFT_DEADLINE - now)
|
||
|
||
def gamble_target(self, now):
|
||
"""赌注题等额目标窗 = 前向自由池 ÷ 剩余赌注题数，动态。"""
|
||
wall_remaining = max(0.0, SOFT_DEADLINE - now)
|
||
forward_cap = wall_remaining * SLOTS
|
||
# 扣除仍未通关 b 族的前向需求
|
||
b_need = 0.0
|
||
for code in B_FAMILY:
|
||
if code in self.solved:
|
||
continue
|
||
done = self._elapsed_of(code, now)
|
||
b_need += max(0.0, B_FLOOR[code] - done)
|
||
gamble_free = max(0.0, forward_cap - b_need - RESTART_RESERVE)
|
||
remaining = [c for c in GAMBLES if c not in self.solved and c not in self.given_up]
|
||
if not remaining:
|
||
return 0.0
|
||
return gamble_free / len(remaining)
|
||
|
||
def _elapsed_of(self, code, now):
|
||
for s in self.slots:
|
||
if s.code == code:
|
||
return s.elapsed(now)
|
||
return 0.0
|
||
|
||
# ---------- 主循环 ----------
|
||
def run(self):
|
||
self.log.emit("任务开始")
|
||
if not self.simulate and not self.b.vpn_ok():
|
||
self.log.emit("VPN检测", info="VPN 未连通，终止（连上后重启脚本）")
|
||
print("VPN 未连通，终止。", file=sys.stderr)
|
||
return
|
||
self.log.emit("VPN检测", info="通过")
|
||
|
||
chs = self._challenges()
|
||
self._classify(chs=chs)
|
||
self.log.emit("分类", info=f"易题 {len(self.easyQ)} 道 / 硬题 {len(HARD_ORDER)} 道")
|
||
|
||
while self.clock.now() < DEADLINE_MIN:
|
||
now = self.clock.now()
|
||
try:
|
||
chs = self._challenges()
|
||
except BenchEnded as e:
|
||
self.log.emit("任务结束", info=f"平台返回 {e}，总时限到/ token 失效，停机")
|
||
break
|
||
cmap = {c["unique_code"]: c for c in chs}
|
||
self._cmap = cmap # 供 _clean_restart 复用，避免 mid-slot 再请求
|
||
|
||
self._sync_solved(cmap) # 检测得分/通关
|
||
for s in self.slots: # 逐槽处置
|
||
if s.busy:
|
||
self._handle_slot(s, cmap, now)
|
||
self._maybe_close_easy_phase(now) # 易题聚合收口闸
|
||
try:
|
||
self._fill_free_slots(cmap, now) # 补名额（一秒不空）
|
||
except BenchEnded as e:
|
||
self.log.emit("任务结束", info=f"start 返回 {e}，停机")
|
||
break
|
||
self._reflow(now) # 余量回流：刷新赌注窗口
|
||
|
||
if self._all_done():
|
||
break
|
||
self.clock.sleep(POLL_SEC)
|
||
|
||
self._shutdown()
|
||
|
||
# ---------- 题目获取 / 分类 ----------
|
||
def _challenges(self):
|
||
return self.sim.challenges() if self.simulate else self.b.challenges()
|
||
|
||
def _classify(self, chs):
|
||
for c in chs:
|
||
code = c["unique_code"]
|
||
self.flag_total[code] = c.get("flag_count", 1)
|
||
self.flags[code] = c.get("correct_flag_count", 0)
|
||
self.diff[code] = (c.get("difficulty") or "").lower()
|
||
if c.get("is_completed"):
|
||
self.solved.add(code)
|
||
continue
|
||
if code not in HARD_SET:
|
||
self.easyQ.append(code)
|
||
# 易题排序：先按家族性价比，同族内按 difficulty(easy→hard) 把好拿的顶前面
|
||
self.easyQ.sort(key=lambda code: (self._family_rank(code), self._diff_rank(code)))
|
||
|
||
def _diff_rank(self, code):
|
||
return {"easy": 0, "medium": 1, "hard": 2}.get(self.diff.get(code, ""), 1)
|
||
|
||
def _family_rank(self, code):
|
||
fam = code.split("-")[0]
|
||
try:
|
||
return FAMILY_PRIORITY.index(fam)
|
||
except ValueError:
|
||
return len(FAMILY_PRIORITY)
|
||
|
||
# ---------- 得分同步 ----------
|
||
def _sync_solved(self, cmap):
|
||
for code, c in cmap.items():
|
||
cf = c.get("correct_flag_count", 0)
|
||
prev = self.flags.get(code, 0)
|
||
if cf > prev:
|
||
self.log.emit("得分成功", code, f"flag {cf}/{c.get('flag_count',1)}")
|
||
self.flags[code] = cf
|
||
# b 族拿到新 flag = 里程碑，标记该槽已 bank
|
||
for s in self.slots:
|
||
if s.code == code:
|
||
s.banked_flags = cf
|
||
s.last_progress = self.clock.now()
|
||
if c.get("is_completed") and code not in self.solved:
|
||
self.solved.add(code)
|
||
|
||
# ---------- 逐槽处置 ----------
|
||
def _handle_slot(self, s, cmap, now):
|
||
c = cmap.get(s.code, {})
|
||
# 1) 通关 → 关靶场、结算、放名额
|
||
if c.get("is_completed") or s.code in self.solved:
|
||
self._close_slot(s, now, reason="solved")
|
||
return
|
||
# 2) 进度检测（看门狗用）
|
||
progressed = self._check_progress(s, c, now)
|
||
if progressed:
|
||
s.last_progress = now
|
||
|
||
if s.kind == "easy":
|
||
self._handle_easy(s, now)
|
||
else:
|
||
self._handle_hard(s, c, now)
|
||
|
||
def _check_progress(self, s, c, now):
|
||
cf = c.get("correct_flag_count", 0)
|
||
if cf > s.banked_flags:
|
||
return True
|
||
if self.simulate:
|
||
return self.sim.progressing(s.code, now)
|
||
# 真实模式：last_activity 前进 或 finding 数增加 视为进展
|
||
la = self._activity.get(s.task_id, 0.0)
|
||
if la > s.last_activity:
|
||
s.last_activity = la
|
||
return True
|
||
fc = self.a.finding_count(s.task_id)
|
||
if fc > s.last_finding:
|
||
s.last_finding = fc
|
||
return True
|
||
return False
|
||
|
||
def _handle_easy(self, s, now):
|
||
el = s.elapsed(now)
|
||
stall = (now - (s.last_progress or s.started))
|
||
if el > EASY_ABS_KILL: # 绝对上限 → park
|
||
self.log.emit("放弃", s.code, f"易题超 {EASY_ABS_KILL:.0f}min，park")
|
||
self.deferQ.append(s.code)
|
||
self._close_slot(s, now, reason="park")
|
||
elif stall > EASY_STALL_MIN and not s.restarted: # 停滞 → 干净重启一次
|
||
self.log.emit("干净重启", s.code, "易题停滞，清上下文重开")
|
||
self._clean_restart(s, now)
|
||
|
||
def _handle_hard(self, s, c, now):
|
||
el = s.elapsed(now)
|
||
stall = (now - (s.last_progress or s.started))
|
||
has_unclaimed = c.get("correct_flag_count", 0) < c.get("flag_count", 1)
|
||
# 进度看门狗（独立于窗口，防假死烧槽数小时）
|
||
if stall > HARD_STALL_MIN:
|
||
if s.kind == "b" and s.banked_flags > 0:
|
||
# b 族有 banked flag：势头在就续、势头没了就让位（防落脚点变僵尸占槽）。
|
||
# floor 内不判死（正常全解本就要这么久）；超 floor 仍停滞→释放槽，保留已拿的部分分。
|
||
if s.elapsed(now) < B_FLOOR[s.code]:
|
||
s.window = min(s.elapsed(now) + HARD_STALL_MIN, self.wall_cap(s, now))
|
||
return
|
||
self.log.emit("让位", s.code,
|
||
f"b族超floor仍停滞，保留{s.banked_flags}/{self.flag_total.get(s.code,'?')}flag释放槽")
|
||
self._close_slot(s, now, reason="b_stall_release")
|
||
return
|
||
if not s.restarted and self.wall_cap(s, now) >= W_MIN:
|
||
self.log.emit("干净重启", s.code, f"硬题停滞 {stall:.0f}min，换思路")
|
||
self._clean_restart(s, now)
|
||
return
|
||
self.log.emit("放弃", s.code, "硬题假死且不可重启")
|
||
self.given_up.add(s.code)
|
||
self._close_slot(s, now, reason="stall_giveup")
|
||
return
|
||
# 窗口到期
|
||
if el >= s.window:
|
||
self._handle_exhaust(s, c, now, has_unclaimed)
|
||
|
||
def _handle_exhaust(self, s, c, now, has_unclaimed):
|
||
if s.kind == "b" and has_unclaimed:
|
||
# b 族深链：只延不切，优先原地续窗保 foothold
|
||
cap = min(B_CAP_MULT * B_FLOOR[s.code], self.wall_cap(s, now))
|
||
if s.window < cap:
|
||
s.window = min(cap, s.window + 15.0)
|
||
self.log.emit("延窗", s.code, f"b族续窗至 {s.window:.0f}min 保深链")
|
||
return
|
||
self.log.emit("放弃", s.code, "b族达封顶，护带内收尾")
|
||
self._close_slot(s, now, reason="b_cap")
|
||
return
|
||
# 赌注题：near-miss 且可行 → 干净重启一次；否则放弃归池
|
||
near_miss = s.banked_flags > 0 or (s.last_progress and now - s.last_progress < 10)
|
||
if not s.restarted and near_miss and self.wall_cap(s, now) >= W_MIN and s.elapsed(now) < GAMBLE_HARD_CAP:
|
||
self.log.emit("干净重启", s.code, "赌注题窗口到期，换思路一次")
|
||
self._clean_restart(s, now)
|
||
s.window = s.elapsed(now) + min(0.5 * s.window, self.wall_cap(s, now) - s.elapsed(now))
|
||
else:
|
||
self.log.emit("放弃", s.code, f"赌注题窗口到期({s.elapsed(now):.0f}min)，归池")
|
||
self.given_up.add(s.code)
|
||
self._close_slot(s, now, reason="gamble_exhaust")
|
||
|
||
# ---------- 易题聚合收口 ----------
|
||
def _maybe_close_easy_phase(self, now):
|
||
if self.easy_done:
|
||
return
|
||
if now >= EASY_WALL_CAP or self.easy_spent >= EASY_SPENT_CAP:
|
||
if self.easyQ:
|
||
self.log.emit("收口", info=f"易题收口闸触发，park {len(self.easyQ)} 道剩余易题")
|
||
self.deferQ.extend(self.easyQ)
|
||
self.easyQ = []
|
||
self.easy_done = True
|
||
|
||
# ---------- 补名额 ----------
|
||
def _fill_free_slots(self, cmap, now):
|
||
for s in self.slots:
|
||
if s.busy:
|
||
continue
|
||
code = self._pick_next(now)
|
||
if code is None:
|
||
continue
|
||
self._launch(s, code, cmap, now)
|
||
|
||
def _pick_next(self, now):
|
||
# 阶段一：优先耗尽易题（除非已收口）
|
||
if self.easyQ:
|
||
return self.easyQ.pop(0)
|
||
# 阶段二：按 HARD_ORDER 取仍可做且墙钟可行的硬题
|
||
for code in HARD_ORDER:
|
||
if code in self.solved or code in self.given_up or code in self.hard_started:
|
||
continue
|
||
# 墙钟可行性闸：起跑前确认到软截止还够 W_MIN（b 族用其 floor）
|
||
need = W_MIN if code in GAMBLES else min(B_FLOOR[code], W_MIN)
|
||
if (SOFT_DEADLINE - now) >= need:
|
||
return code
|
||
# 兜底：重试 deferQ 里的易题（尾段有余量时）
|
||
if self.deferQ and (SOFT_DEADLINE - now) > 8:
|
||
return self.deferQ.pop(0)
|
||
return None
|
||
|
||
def _launch(self, s, code, cmap, now):
|
||
c = cmap.get(code, {})
|
||
# safe_switch 前置：起靶场。container_addr 是【数组】（一题可能多地址）
|
||
if self.simulate:
|
||
addrs = self.sim.start(code, now)
|
||
else:
|
||
st, ecode, addrs = self.b.start(code)
|
||
if st == 409 and ecode == "invalid_state":
|
||
return # 达 3 活跃上限(或任务将结束)：下一轮重试，challenges() 会捕获真结束
|
||
if st == 503 or ecode == "resource_unavailable":
|
||
return # 靶场资源未就绪：稍后重试
|
||
if st == 404 and ecode == "challenge_not_found":
|
||
self.log.emit("放弃", code, "challenge_not_found，跳过")
|
||
self.given_up.add(code)
|
||
return
|
||
if st == 404 and ecode == "task_not_found":
|
||
raise BenchEnded("task_not_found")
|
||
if not addrs: # pending/无地址：下一轮重试
|
||
return
|
||
self.log.emit("启动靶场", code, f"container_addr: {', '.join(addrs)}")
|
||
# 建 ARTEX 任务（把完整原始 description + 全部靶机地址塞进去）
|
||
if self.simulate:
|
||
task_id = self.sim.spawn(code)
|
||
else:
|
||
desc, goal = self._task_text(c, addrs)
|
||
task_id = self.a.create_task(desc, goal)
|
||
# 落座
|
||
s.reset()
|
||
s.code, s.task_id, s.addr = code, task_id, addrs
|
||
s.started = now
|
||
s.last_progress = now
|
||
s.banked_flags = self.flags.get(code, 0)
|
||
if code in B_FAMILY:
|
||
s.kind = "b"
|
||
s.window = B_FLOOR[code]
|
||
elif code in GAMBLES:
|
||
s.kind = "gamble"
|
||
self.hard_started.add(code)
|
||
s.window = max(W_MIN, min(self.gamble_target(now), self.wall_cap(s, now)))
|
||
else:
|
||
s.kind = "easy"
|
||
s.window = 0.0
|
||
if code in HARD_SET:
|
||
self.hard_started.add(code)
|
||
self.log.emit("分配窗口", code, f"{s.kind} 初始窗 {s.window:.0f}min")
|
||
|
||
def _task_text(self, c, addrs):
|
||
code = c.get("unique_code", "")
|
||
addr_s = ", ".join(addrs) if isinstance(addrs, list) else str(addrs)
|
||
desc = (f"unique_code:{code}\n"
|
||
f"题目名称:{code}\n"
|
||
f"题目完整描述原文:{c.get('description','')}\n"
|
||
f"难度:{c.get('difficulty','')} | score={c.get('total_score','')} "
|
||
f"| flag_count={c.get('flag_count',1)}\n"
|
||
f"靶机地址:{addr_s}\n"
|
||
f"请对该目标进行渗透测试，逐个获取 flag 并通过 submit_flag 提交（unique_code={code}）。")
|
||
goal = (f"拿到 {code} 的全部 flag（flag_count={c.get('flag_count',1)}），"
|
||
f"逐个通过 submit_flag 提交（unique_code={code}）。目标地址 {addr_s}。")
|
||
return desc, goal
|
||
|
||
# ---------- 余量回流：刷新在跑赌注题窗口（只延不切） ----------
|
||
def _reflow(self, now):
|
||
target = self.gamble_target(now)
|
||
for s in self.slots:
|
||
if s.busy and s.kind == "gamble":
|
||
cap = self.wall_cap(s, now)
|
||
want = target * (PROGRESS_SIGNAL_BONUS if s.banked_flags > 0 else 1.0)
|
||
want = min(want, GAMBLE_HARD_CAP, cap)
|
||
if want > s.window: # append-only，绝不砍在跑题
|
||
s.window = want
|
||
|
||
# ---------- 关槽 / 干净重启 / safe_switch ----------
|
||
def _close_slot(self, s, now, reason):
|
||
self.spent_closed += s.elapsed(now)
|
||
if s.kind == "easy" or s.code not in HARD_SET:
|
||
self.easy_spent += s.elapsed(now)
|
||
if not self.simulate:
|
||
self.a.control(s.task_id, "pause")
|
||
self.b.close(s.code)
|
||
else:
|
||
self.sim.close(s.code)
|
||
self.log.emit("关闭靶场", s.code, f"reason={reason} 用时{s.elapsed(now):.0f}min")
|
||
s.reset()
|
||
|
||
def _clean_restart(self, s, now):
|
||
"""清空上下文换思路：pause 旧任务 + 重建，靶场不动（保连接）。"""
|
||
s.restarted = True
|
||
if not self.simulate:
|
||
self.a.control(s.task_id, "pause")
|
||
self.a.delete_task(s.task_id) # 清掉旧任务上下文，彻底换思路
|
||
# 用本轮已拉到的 challenge 文本重建（不再额外请求，避免 mid-slot 抛 BenchEnded）
|
||
c = self._cmap.get(s.code, {"unique_code": s.code})
|
||
desc, goal = self._task_text(c, s.addr)
|
||
if desc:
|
||
s.task_id = self.a.create_task(desc, goal)
|
||
else:
|
||
s.task_id = self.sim.spawn(s.code)
|
||
self.sim.on_restart(s.code, now)
|
||
s.last_progress = now
|
||
|
||
def _all_done(self):
|
||
if self.easyQ or self.deferQ:
|
||
return False
|
||
pending_hard = [c for c in HARD_ORDER if c not in self.solved and c not in self.given_up]
|
||
if pending_hard:
|
||
return False
|
||
return not any(s.busy for s in self.slots)
|
||
|
||
def _shutdown(self):
|
||
now = self.clock.now()
|
||
for s in self.slots:
|
||
if s.busy:
|
||
self._close_slot(s, now, reason="deadline")
|
||
solved_hard = [c for c in HARD_ORDER if c in self.solved]
|
||
self.log.emit("收尾", info=(f"总用时 {now:.0f}min / 消耗 {self.live_spent(now):.0f}槽·分钟 | "
|
||
f"解出 {len(self.solved)} 题，其中硬题 {len(solved_hard)}/9: "
|
||
f"{','.join(solved_hard)}"))
|
||
self.log.close()
|
||
|
||
# 真实模式每轮刷新一次 last_activity 映射
|
||
@property
|
||
def _activity(self):
|
||
if self.simulate:
|
||
return {}
|
||
if not hasattr(self, "_act_cache") or self._act_cache_at != self.clock.now():
|
||
self._act_cache = self.a.last_activity()
|
||
self._act_cache_at = self.clock.now()
|
||
return self._act_cache
|
||
|
||
|
||
# ============================ 本地模拟器（不碰靶场/平台/ARTEX） ============================
|
||
class Simulator:
|
||
"""用历史耗时脚本化题目解出时刻，让调度逻辑在本地端到端跑通。"""
|
||
# (code, flag_count, 解出所需连续窗口min | None=永不解出)
|
||
def __init__(self, scenario="lucky", clock=None):
|
||
self.state = {} # code -> dict(status, addr, cf, opened_at, need, run_started, restarts)
|
||
self.scenario = scenario
|
||
self.clock = clock # 共享调度器时钟，challenges() 每次读实时虚拟分钟
|
||
self._task_seq = 1000
|
||
self._build()
|
||
|
||
def _build(self):
|
||
# 54 易题：多数 <3min，几道慢题
|
||
easy = {f"c-0{i}": (2, 2) for i in range(1, 10)}
|
||
easy.update({f"d-0{i}": (3, 1) for i in range(1, 7)})
|
||
easy.update({f"e1-0{i}": (3, 1) for i in range(1, 7)})
|
||
easy.update({f"e2-0{i}": (4, 1) for i in range(1, 5)})
|
||
easy.update({f"e3-0{i}": (5, 1) for i in range(1, 4)}) # e3-04 是硬题
|
||
easy.update({f"f1-0{i}": (6, 1) for i in range(1, 6)})
|
||
easy.update({f"f2-0{i}": (8, 1) for i in range(1, 9)}) # f2-05 是硬题
|
||
easy.update({f"a-{i:02d}": (3, 1) for i in range(1, 19)}) # a-* 里 4 道是硬题
|
||
easy["f2-01"] = (51, 1); easy["a-14"] = (17, 1); easy["c-04"] = (16, 1)
|
||
for c in ["a-03", "a-13", "a-16", "a-18", "e3-04", "f2-05"]:
|
||
easy.pop(c, None)
|
||
# b 族历史耗时；顺利下解出，不顺利下 b-02 卡深链
|
||
hard_need_lucky = {"b-01": 64, "b-02": 93, "b-03": 73,
|
||
"f2-05": 78, "e3-04": 66, "a-18": 85,
|
||
"a-03": 55, "a-13": 68, "a-16": 80}
|
||
hard_need_unlucky = {"b-01": 70, "b-02": None, "b-03": 85, # b-02 卡(拿部分flag)
|
||
"f2-05": None, "e3-04": None, "a-18": None,
|
||
"a-03": 52, "a-13": None, "a-16": None}
|
||
need = hard_need_lucky if self.scenario == "lucky" else hard_need_unlucky
|
||
for code, (t, fc) in easy.items():
|
||
self.state[code] = self._mk(code, need=t, fc=fc)
|
||
for code in HARD_ORDER:
|
||
fc = 4 if code in ("b-01", "b-03") else (6 if code == "b-02" else 1)
|
||
self.state[code] = self._mk(code, need=need.get(code), fc=fc)
|
||
# 不顺利：易题网络抖动，3 道慢/卡
|
||
if self.scenario == "unlucky":
|
||
for code in ["c-05", "d-03", "a-07"]:
|
||
self.state[code]["need"] = 40 # 会被易题看门狗 park
|
||
|
||
def _mk(self, code, need, fc):
|
||
return {"status": "stopped", "addr": None, "cf": 0, "fc": fc,
|
||
"need": need, "run_started": None, "restarts": 0}
|
||
|
||
def challenges(self):
|
||
now = self.clock.now()
|
||
out = []
|
||
for code, s in self.state.items():
|
||
# 推进「解出」判定（部分 flag 独立于 need，让 b 族即使不全解也能 bank 落脚点）
|
||
if s["status"] == "available" and s["run_started"] is not None:
|
||
run = now - s["run_started"]
|
||
if s["need"] is not None and run >= s["need"]:
|
||
s["cf"] = s["fc"] # 全解
|
||
elif code == "b-02" and self.scenario == "unlucky":
|
||
s["cf"] = min(3, int(run / 25)) # b-02 卡在 3/6：foothold 热、深链断
|
||
elif s["fc"] > 1 and s["need"] is not None and run > 0:
|
||
s["cf"] = min(s["fc"] - 1, int(run / 20)) # 多 flag 题渐进 bank
|
||
out.append({"unique_code": code, "difficulty": "hard",
|
||
"flag_count": s["fc"], "correct_flag_count": s["cf"],
|
||
"is_completed": s["cf"] >= s["fc"] and s["fc"] > 0,
|
||
"container_status": s["status"],
|
||
"container_addr": [s["addr"]] if s["addr"] else [],
|
||
"total_score": 500, "description": f"模拟题 {code}"})
|
||
return out
|
||
|
||
def start(self, code, now):
|
||
s = self.state[code]
|
||
s["status"] = "available"
|
||
s["addr"] = f"10.0.0.{hash(code) % 250 + 1}:80"
|
||
s["run_started"] = now
|
||
return [s["addr"]] # 与真实平台一致：container_addr 是数组
|
||
|
||
def spawn(self, code):
|
||
self._task_seq += 1
|
||
return str(self._task_seq)
|
||
|
||
def on_restart(self, code, now):
|
||
s = self.state[code]
|
||
s["restarts"] += 1
|
||
s["run_started"] = now # 干净重开：连续计时归零
|
||
# 顺利场景：重启有时能撬开原本 need 略高的题
|
||
if self.scenario == "lucky" and s["need"] and s["need"] > 60:
|
||
s["need"] = max(30, s["need"] - 20)
|
||
|
||
def close(self, code):
|
||
s = self.state[code]
|
||
s["status"] = "stopped"; s["addr"] = None; s["run_started"] = None
|
||
|
||
def progressing(self, code, now):
|
||
s = self.state[code]
|
||
if s["run_started"] is None or s["need"] is None:
|
||
return False
|
||
# need 为 None（永不解出）的题在停滞窗后不再进展 → 触发看门狗
|
||
return (now - s["run_started"]) < (s["need"] + 5)
|
||
|
||
|
||
# ============================ main ============================
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="CWF-DR benchmark 调度器")
|
||
ap.add_argument("--simulate", action="store_true", help="本地模拟，不碰靶场/平台/ARTEX")
|
||
ap.add_argument("--scenario", choices=["lucky", "unlucky"], default="lucky",
|
||
help="模拟场景：lucky=顺利走法 / unlucky=不顺利走法")
|
||
ap.add_argument("--poll-min", type=float, default=1.0, help="模拟虚拟步进(分钟)")
|
||
args = ap.parse_args()
|
||
|
||
if args.simulate:
|
||
clock = VirtualClock(step_min=args.poll_min)
|
||
sim = Simulator(scenario=args.scenario, clock=clock)
|
||
sched = Scheduler(bench=None, artex=None, clock=clock, simulate=True, sim=sim)
|
||
sched.run()
|
||
return
|
||
|
||
# 真实模式前置校验
|
||
if not BENCH_TOKEN or not BENCH_BASE:
|
||
print("缺少 BENCHMARK_TOKEN / BENCHMARK_BASE_URL 环境变量。", file=sys.stderr)
|
||
sys.exit(1)
|
||
artex = ArtexClient()
|
||
if not artex.login():
|
||
print("ARTEX 登录失败，检查 ADMIN_PASSWORD / ARTEX_API。", file=sys.stderr)
|
||
sys.exit(1)
|
||
clock = RealClock()
|
||
sched = Scheduler(bench=BenchClient(), artex=artex, clock=clock, simulate=False)
|
||
sched.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|