Competition Runner 设计文档

版本: v2.0 | 日期: 2026-04-11

1. 背景与动机

问题

tchkiller 是一个单目标渗透引擎：每次执行攻击一个靶机，手动指定 target。这在 TCH 智能渗透挑战赛中不够用：

多目标调度 — 赛区有 20+ 个靶机，需要自动遍历
Flag 提交 — tchkiller 检测到 flag 但不提交到比赛平台
关卡解锁 — 需要监控 flag 提交阈值，解锁后自动切换到新关卡
断点恢复 — 比赛可能中断，需要从上次进度继续
全自主运行 — 比赛期间选手无法 SSH 干预，agent 必须全自动

方案

在 tchkiller 之上增加一层确定性 Python 编排器（Competition Runner），负责：

与比赛平台交互（获取赛题、启停实例、提交 flag）
并行调度 tchkiller 攻击多个靶机（最多 3 个 worker）
基于规则决策目标优先级和时间分配
自动 Hint 策略
持久化状态，支持中断恢复

关键设计决策：编排层不用 LLM。选目标、分配时间、提交 flag 都是确定性逻辑，用规则即可。LLM 只在 tchkiller 内部负责渗透本身。

三层架构

Layer 0: watchdog.sh        — 进程保活
Layer 1: competition.py     — 确定性编排器（本文档）
Layer 2: tchkiller (main.py) — 渗透智能体

2. 架构

总览

┌──────────────────────────────────────────────────────────────────┐
│                   competition.py (编排层)                         │
│                                                                   │
│  ┌──────────────┐  ┌───────────┐  ┌────────────┐  ┌───────────┐  │
│  │  Platform     │  │ Strategy  │  │ WorkerPool │  │  State    │  │
│  │  (平台 API)   │  │ (选目标)   │  │ (并发调度)  │  │ (持久化)  │  │
│  └──────┬───────┘  └─────┬─────┘  └─────┬──────┘  └─────┬─────┘  │
│         │                │              │                │        │
│         ▼                ▼              ▼                ▼        │
│  MockPlatform     pick_next_batch  N workers (1-3)  comp_state   │
│  TCHPlatform      allocate_time    TargetRunner       .json      │
│                   should_use_hint  flag 实时监控                   │
│                   extra_prompt     实例启停管理                    │
└──────────────────────────────────────────────────────────────────┘
                            │ (子进程 × N)
                            ▼
               ┌─────────────────────────┐
               │   tchkiller (main.py)    │
               │   单目标渗透 + flag 检测  │
               │   output/<run_dir>/      │
               └─────────────────────────┘

流式调度模型

v2 采用流式 Worker 池替代 v1 的顺序执行。每个 worker 完成当前任务后立即领取下一个，无需等待其他 worker。

# WorkerPool 核心循环 (每个 worker 独立)
async def _worker_loop(worker_id):
    while not stop:
        challenge = await claim_next(worker_id)  # 领取（刷新赛题、检测解锁）
        if challenge is None:
            return  # 全部完成
        result = await execute_challenge(challenge)  # 启动实例 → 攻击 → 提交 → 停止实例
        process_result(result)
        signal_others()  # 通知: 可能触发关卡解锁

3. 模块设计

3.1 Platform（平台适配器）

comp/platform.py

抽象接口 + 两个实现：

方法	说明
`get_challenges()`	获取赛题列表 + 当前关卡/得分/进度，返回 `PlatformStatus`
`start_challenge(code)`	启动赛题实例，返回 `entrypoint` 列表
`stop_challenge(code)`	停止赛题实例（释放槽位）
`submit_flag(code, flag)`	提交 flag，返回 `SubmitResult`（含 flag_got_count、解锁通知）
`view_hint(code)`	查看提示（-10% 分数）
`stop_all()`	紧急停止所有运行中的实例

TCHPlatform：官方 HTTP API 客户端。

端点: http://<SERVER_HOST>/api
认证: Agent-Token header
限频: 0.4s 间隔（< 3 req/s 上限），429 自动重试
连接失败自动重试 2 次
所有请求/响应记录到 api_debug.log

MockPlatform：本地测试用。支持三种初始化方式：

默认内置 5 道赛题（2 个关卡，含多 flag 题）
从 targets.txt 读取（每行一个 URL/IP）
从 challenges.json 读取（完整赛题定义，含预设 flag）

Mock 模式下模拟实例启停（同时运行上限 3 个）、flag 验证（预设/任意）、关卡解锁（当前关卡 easy 全解）。

数据模型

@dataclass
class Challenge:
    code: str               # 赛题唯一标识
    title: str
    difficulty: str         # easy / medium / hard
    description: str
    level: int              # 关卡编号
    total_score: int        # 基础分值
    total_got_score: int    # 已获得分数
    flag_count: int         # 预期 flag 数量
    flag_got_count: int     # 已提交的 flag 数量
    hint_viewed: bool       # 是否已查看提示
    instance_status: str    # stopped / pending / running
    entrypoint: list[str]   # 入口地址列表

@dataclass
class SubmitResult:
    correct: bool
    message: str
    flag_count: int         # 题目总 flag 数
    flag_got_count: int     # 当前已提交数
    level_unlocked: bool    # 是否触发了新关卡解锁

3.2 Strategy（目标选择策略）

comp/strategy.py

难度配置

难度	首次超时	重试超时	最大尝试	Hint 阈值
easy	30min	25min	3	第 2 次失败后
medium	35min	25min	3	第 2 次失败后
hard	50min	40min	3	第 1 次失败后

目标选择

pick_next_batch(challenges, max_workers, exclude_codes) 选择一批目标（最多 N 个并行）：

排序规则：

低关卡优先（level 1 → level N）
低难度优先（easy → medium → hard）
尝试次数少的优先
已 solved / skip / 达到 max_attempts / 正在执行中的跳过

时间分配

allocate_time() 根据难度和尝试次数决定超时：

首次尝试 → 首次超时
重试 → 重试超时（稍短，已有侦察信息）

Hint 策略

should_use_hint() — 失败 N 次后自动获取平台提示：

easy/medium: 2 次失败后请求
hard: 1 次失败后请求
每道题只请求一次，hint 内容注入到后续 extra_prompt

攻击方向推断

infer_attack_prompt() 从 title + description 关键词推断攻击方向（SQL 注入、弱口令、文件上传、RCE、域渗透等），自动注入到 extra_prompt。

历史注入

get_extra_prompt() 生成附加提示词：

攻击方向推断
平台 hint 内容
第 N 次尝试、之前发现的漏洞和 flag
尚未成功提交的 flag
之前的错误信息
"请尝试不同的攻击方向和方法"

3.3 WorkerPool（并发调度器）

comp/worker_pool.py（v2 新增）

流式并发 Worker 池，替代 v1 的顺序执行。

核心机制

启动 N 个 worker 协程（N = 1-3，--workers 参数控制）
每个 worker 循环：领取 → 执行 → 结果处理 → 领取下一个
领取时刷新赛题列表（检测关卡解锁、新题可用）
无可用目标时等待其他 worker 完成（最多 5 分钟，完成可能触发解锁）
使用 asyncio.Lock 保护领取操作，避免同一题被多个 worker 抢占
全部完成 → _stop_event → 所有 worker 优雅退出

Dry-run 模式

--dry-run 只显示调度计划（选哪些题、超时多少、第几次尝试），不实际执行。

3.4 Runner（单目标执行器）

comp/runner.py

完整生命周期

start_challenge → 等待就绪 → 攻击 → 实时提交 → 最终提交 → stop_challenge → 记录

启动实例: platform.start_challenge(code) 获取 entrypoint
构建命令: python3 main.py <target> --max-rounds N --round-timeout T --total-timeout T ...
启动子进程: asyncio.create_subprocess_exec，独立进程组 (start_new_session=True)
并行任务:
- 输出流 (_stream_output): 实时打印 + 捕获 run_dir 路径
- flag 监控 (_monitor_flags): 每 5 秒扫描 evidence 目录，发现 flag 立即提交
- solved 检测 (solved_event): 平台确认所有 flag 已提交 → 提前终止子进程（省时）
硬超时: rounds × round_timeout × 0.6 + 10 分钟，超时 SIGTERM
最终扫描 (_final_flag_scan): 再扫一次 evidence
最终提交 (_submit_all_flags): 提交残余 flag
停止实例: platform.stop_challenge(code) 释放资源
记录: state_mgr.record_attempt()

Flag 检测

正则: re.compile(r'flag\{[a-zA-Z0-9_\-]+\}', re.IGNORECASE)

扫描范围（仅当前目标的 run_dir，不跨目录）：

output/<run_dir>/evidence/*.md（实时监控 + 最终扫描）
report.json 中的 flag_value 字段

多 Flag 支持

多 flag 题的 solved 判定：

实时提交路径: 每次 submit_flag 后检查 flag_got_count >= flag_count，满足时调用 mark_solved 并触发 solved_event（提前终止子进程）
最终提交路径: 同样检查 SubmitResult 的 flag_got_count >= flag_count，仅达标时调用 mark_solved
部分提交（如 3 个 flag 只找到 1 个）不会被错误标记为 solved

子进程管理

start_new_session=True 创建独立进程组
os.killpg() 终止整个进程组（tchkiller + 子进程）
fallback 到 proc.kill() 如果进程组不存在

入口引导

自动注入 已知入口为 <target>，优先从此端口开始攻击。 到 prompt，避免浪费时间做全端口扫描。

3.5 State（持久化状态）

comp/state.py

状态文件 comp_state.json，每次 flag 提交或攻击完成后自动保存。

数据结构

CompetitionState
├── session_id              # 会话 ID
├── start_time              # 开始时间
├── current_level           # 当前关卡
├── daily_windows_used      # 今日已用挑战时段数
├── total_cost_usd          # 总花费
├── total_flags             # 总 flag 数
├── total_challenges_solved # 已解决赛题数
├── total_score             # 总得分
├── log                     # 操作日志 (最近 500 条)
└── challenges              # 赛题状态字典 (code → ChallengeState)
    └── ChallengeState
        ├── code, level, difficulty, target, name, description
        ├── total_score, flag_count, flag_got_count
        ├── hint_viewed, hint_content
        ├── attempts[]              # 攻击记录列表
        │   └── AttemptRecord
        │       ├── timestamp, duration_s, mode
        │       ├── flags_found, flags_submitted
        │       ├── vulns, cost_usd, run_dir
        │       └── success, error
        ├── all_flags_found         # 历次发现的所有 flag
        ├── flags_submitted         # 已成功提交的 flag
        ├── solved                  # 是否已解决
        └── skip                    # 手动标记跳过

平台同步

sync_from_platform() 每次 worker 领取任务前调用，从 API 拉取最新状态：

更新 entrypoint、分数、flag 计数、hint 状态
平台标记 solved 的题目同步到本地
不会覆盖本地已有的 attempts 历史

断点恢复

--resume 加载已有的 comp_state.json，跳过已 solved 的赛题，从未完成的目标继续。已尝试过的目标会获得历史上下文注入（通过 strategy.get_extra_prompt()）。

4. 与 tchkiller 的集成

不改动 tchkiller

Competition Runner 通过子进程调用 main.py，不修改 tchkiller 本体的任何代码。集成点：

交互方式	说明
命令行参数	`python3 main.py <target> --max-rounds N --round-timeout T --total-timeout T --orchestrator --team --prompt <extra>`
透传参数	`--model`, `--provider`, `--browser`, `--no-team`, `--no-orchestrator`
文件读取	`output/<run_dir>/report.json` — 结果（vulns、cost）
文件读取	`output/<run_dir>/evidence/*.md` — flag 检测
stdout	捕获 `输出:` 行获取 run_dir 路径

tchkiller 命令构建逻辑

rounds = 4 if difficulty == "hard" else 3
cmd = [
    "python3", "main.py", target,
    "--max-rounds", str(rounds),
    "--round-timeout", str(timeout_min),
    "--total-timeout", str(int(rounds * timeout_min * 0.6) + 5),
    "--orchestrator", "--team",
    "--prompt", extra_prompt,
]

flag 格式

所有组件统一使用 case-insensitive 正则: flag\{[^}]+\} 或 flag\{[a-zA-Z0-9_\-]+\}

5. CLI 用法

# 正式比赛
python3 competition.py --server <HOST:PORT> --token <TOKEN>
python3 competition.py --server <HOST:PORT> --token <TOKEN> --workers 3

# Mock 模式 (本地测试)
python3 competition.py --mock
python3 competition.py --mock --targets targets.txt
python3 competition.py --mock --challenges challenges.json

# 恢复中断
python3 competition.py --server <HOST:PORT> --token <TOKEN> --resume

# 紧急停止所有实例
python3 competition.py --server <HOST:PORT> --token <TOKEN> --stop-all

# 查看当前状态
python3 competition.py --status

# 试运行
python3 competition.py --mock --dry-run

# 清除本地状态重新开始
python3 competition.py --mock --reset

# 透传 tchkiller 参数
python3 competition.py --server <HOST:PORT> --token <TOKEN> --model claude-sonnet --provider openrouter

6. 文件清单

tchkiller/
├── competition.py              # 主入口 (CLI + 编排主循环)
├── comp/
│   ├── __init__.py
│   ├── platform.py             # 平台接口 + TCHPlatform (HTTP) + MockPlatform
│   ├── state.py                # 持久化状态管理
│   ├── strategy.py             # 目标选择 + 时间分配 + Hint 策略
│   ├── runner.py               # 单目标执行器 (子进程 + flag 监控 + 实例生命周期)
│   └── worker_pool.py          # 流式并发 Worker 池
├── comp_state.json             # 运行时生成: 持久化状态
├── comp_report.json            # 运行时生成: 最终比赛报告
└── api_debug.log               # 运行时生成: API 请求/响应日志

7. 后续迭代

近期（比赛前）

~~TCHPlatform 实现~~ — HTTP API 客户端已完成
~~并行攻击~~ — WorkerPool 流式并发已实现
MockPlatform 解锁逻辑对齐 — 当前用 "easy 全解"，应模拟 flag 阈值解锁

中期（比赛中迭代）

动态时间调整 — 根据历史攻击成功率动态调整超时
成本预警 — 接近预算上限时告警

长期

赛后复盘自动化 — 自动生成比赛报告（按关卡、按时间线）
A/B 测试 — 不同策略配置的效果对比

Uh oh!

FilesExpand file tree

competition-runner-design.md

Latest commit

History