tsec-benchmark 0.1.2

Python SDK for the TSec Benchmark Platform player-facing Challenges API.

tsec-benchmark

TSec Benchmark 平台 面向选手的题目挑战 API 的 Python SDK。它可以让基于 Python 的受测 Agent 驱动跑分流程——列出题目、启动容器、获取提示、提交 flag、关闭容器——而无需手写 HTTP 样板代码。

pip install tsec-benchmark

• Python ≥ 3.9
• 依赖：httpx
• 以异步为主，并附带适用于脚本的同步封装

前置条件（所有接入方式通用）：使用本 SDK 之前，你必须已在平台控制台/管理端获取以下内容：

• 对应你跑分任务的 BENCHMARK_TOKEN，
• 平台 base URL（例如 https://benchmark.example.com），
• 已连通的 VPN（访问题目容器地址需要通过 VPN）。

本 SDK 只覆盖面向选手的流程——不会创建跑分任务，也不会管理 VPN。

VPN 连通性预检（pre-flight check）

题目容器地址 container_addr 只能从跑分 VPN 网络内直连，VPN 未连通时一切流程都会失败。因此 SDK 在一切流程开始之前，会做 VPN 联通检测：

• 自动检测（默认）：用上下文管理器（async with / with）进入客户端时，SDK 会自动发起一次预检；若不通，立即抛出 VpnCheckError 并中断，不会发起任何平台请求。错误信息为：

  VPN检测未通过,请检查靶场VPN网络配置
  

• 手动检测：也可直接调用 client.check_vpn()，返回 VpnCheckResult（含 status/client_ip/time/ok），失败时同样抛 VpnCheckError。

• 关闭自动检测：构造时传 auto_check_vpn=False（如单测/无 VPN 的离线场景），自行决定是否调用 check_vpn()。

from tsec_benchmark import TSecBenchmark, VpnCheckError

try:
    with TSecBenchmark(base_url="...", token="...") as client:   # 进 with 时自动预检 VPN
        ...                                                        # 预检通过才会执行到这里
except VpnCheckError as e:
    print(e)   # VPN检测未通过,请检查靶场VPN网络配置

快速开始

同步用法（最简单）

from tsec_benchmark import TSecBenchmark, DuplicateSubmit, InvalidState, VpnCheckError


def call_your_agent(ch, started):
    """← 这里写你自己的解题逻辑。

    题目容器已启动，靶场地址是 started.container_addr（`IP:端口` 地址数组，可能有多个，需经 VPN 直连）。
    在这里调用你的渗透/解题 Agent（读源码、发请求、跑 exploit 等）去拿 flag。
    一道题可能有多个 flag（ch.flag_count），全部拿到后返回 flag 列表。
    """
    # TODO: 在这里实现你的解题逻辑，例如连接 started.container_addr 进行渗透
    flags = ["flag{...}"]  # 替换为你的 Agent 实际拿到的 flag
    return flags


# 进入 with 时 SDK 自动做 VPN 联通预检；不通则抛 VpnCheckError 中断，不会发起任何平台请求。
with TSecBenchmark(base_url="https://benchmark.example.com", token="<BENCHMARK_TOKEN>") as client:
    # 1) 列出题目：拿到本轮所有题目及每题作答进度（已通关的跳过）
    challenges = client.list_challenges()
    for ch in challenges:
        if ch.is_completed:
            continue

        # 2) 启动题目容器：拿到靶场直连地址 started.container_addr
        started = client.start_challenge(ch.unique_code)
        print(f"{ch.unique_code} -> {started.container_addr}")

        try:
            # 3) 解题（你自己的逻辑）：连接靶场地址、渗透、拿到 flag
            flags = call_your_agent(ch, started)

            # 4) 提交 flag：每拿到一个就提交一次，推进 correct_flag_count
            for flag in flags:
                try:
                    res = client.submit_flag(ch.unique_code, flag)
                    print(f"  correct={res.correct} awarded={res.awarded} "
                          f"progress={res.correct_flag_count}/{res.total_flag_count}")
                except DuplicateSubmit:
                    # 同一个 flag 已正确提交过（幂等），跳过即可
                    pass
        finally:
            # 5) 关闭题目容器：释放靶场资源与活跃名额，无论解题是否完成
            client.close_challenge(ch.unique_code)

异步用法（基于 httpx）

import asyncio
from tsec_benchmark import TSecBenchmarkAsync


async def call_your_agent(ch, started):
    """← 这里写你自己的解题逻辑（异步）。靶场地址是 started.container_addr，经 VPN 直连。
    调用你的解题 Agent 拿到 flag 列表后返回。"""
    # TODO: 在这里实现你的解题逻辑
    return ["flag{...}"]


async def main():
    # 进入 async with 时 SDK 自动做 VPN 联通预检；不通则抛 VpnCheckError 中断。
    async with TSecBenchmarkAsync(base_url="https://benchmark.example.com", token="<BENCHMARK_TOKEN>") as client:
        challenges = await client.list_challenges()              # 1) 列出题目及进度
        for ch in challenges:
            if ch.is_completed:
                continue
            started = await client.start_challenge(ch.unique_code)   # 2) 启动题目容器
            try:
                flags = await call_your_agent(ch, started)            # 3) 解题（你的逻辑）
                for flag in flags:
                    await client.submit_flag(ch.unique_code, flag)    # 4) 提交 flag
            finally:
                await client.close_challenge(ch.unique_code)         # 5) 关闭容器、释放资源

asyncio.run(main())

API 参考

五个方法，与平台API接口一一对应。每个方法返回一个不可变的 dataclass。步骤列展示了该调用在解题循环中所处的位置。进入上下文管理器时会自动运行预检 check_vpn()（步骤 0）。

步骤	方法	端点	这一步在干什么	返回值
0	check_vpn()	GET http://{healthcheck_url}	一切流程开始前的 VPN 联通预检；不通则抛 VpnCheckError 中断。默认进上下文管理器时自动执行	VpnCheckResult
1	list_challenges()	GET /api/v1/challenges	拉取题目列表与每题作答进度，决定下一道要做的题	list[Challenge]
2	start_challenge(unique_code)	POST /api/v1/challenges/start?unique_code=	启动该题靶场容器，拿到经 VPN 直连的地址 container_addr	StartResult
3	call_your_agent(...)	—	你自己的解题逻辑：连接 container_addr 渗透、拿到 flag	list[str]
4	submit_flag(unique_code, flag)	POST /api/v1/challenges/submit (JSON body)	提交 flag 判定，推进 correct_flag_count（多 flag 需多次提交）	SubmitResult
5	close_challenge(unique_code)	POST /api/v1/challenges/close?unique_code=	关闭容器、释放靶场资源与活跃名额	CloseResult
可选	get_hint(unique_code)	GET /api/v1/challenges/hint?unique_code=	步骤 3 前可调用获取提示；查看后该题 flag 得分按比例扣减，权衡后再用	HintResult

数据类

SDK 的每个 API 方法都会返回一个不可变（frozen）的数据类，字段与平台 JSON 字段名完全一致（均为 snake_case）。

Challenge

题目信息数据类，由 list_challenges() 返回，包含题目基本信息和作答进度。

字段	类型	说明
unique_code	str	题目唯一标识码，用于启动/提交/关闭等操作
description	str	题目描述信息
difficulty	str	题目难度级别（如：easy/medium/hard）
level	str	题目等级
total_score	int	题目总分数
flag_count	int	题目包含的 flag 总数（一道题可能有多个 flag）
correct_flag_count	int	已正确提交的 flag 数量
is_completed	bool	题目是否已全部完成（所有 flag 都已提交正确）
container_status	str	该题靶场容器当前状态：pending/available/stop_pending/stopped；尚未启动或已关闭为 stopped
container_addr	list[str]	靶场容器直连地址（IP:端口）数组；仅当 container_status == "available" 时才有值，其余状态为空列表 []

StartResult

启动题目容器后返回的结果数据类，由 start_challenge() 返回。

字段	类型	说明
unique_code	str	题目唯一标识码
container_addr	list[str]	题目容器的可访问地址数组（格式：IP:端口），一个题目可能有多个地址，需要通过 VPN 直连

HintResult

获取提示后返回的结果数据类，由 get_hint() 返回。

字段	类型	说明
unique_code	str	题目唯一标识码
hint	str | None	提示内容；如果题目没有提示则返回 None；注意：查看提示后该题 flag 得分会按比例扣减

SubmitResult

提交 flag 后返回的结果数据类，由 submit_flag() 返回。

字段	类型	说明
correct	bool	提交的 flag 是否正确
awarded	int	本次提交获得的分数（错误则为 0）
cumulative_score	int	累计已获得的总分数
correct_flag_count	int	已正确提交的 flag 数量
total_flag_count	int	题目包含的 flag 总数
matched_flag_index	int | None	匹配的 flag 索引（从 0 开始）；提交错误时为 None

CloseResult

关闭题目容器后返回的结果数据类，由 close_challenge() 返回。

字段	类型	说明
unique_code	str	题目唯一标识码
closed	bool	容器是否成功关闭

VpnCheckResult

VPN 连通性检测结果数据类，由 check_vpn() 返回，或在上下文管理器入口自动检测时内部使用。

字段	类型	说明
status	str	检测状态；正常时为 "ok"
client_ip	str	当前客户端 IP 地址（VPN 分配的地址）
time	str	检测时间（格式：YYYY-MM-DD HH:MM:SS）
ok	bool	检测是否通过（status == "ok" 时为 True）

异常

每个平台错误码都会抛出对应的专用异常（全部继承自 TSecError，携带 .code、.message、.detail、.status_code）：

平台 code	异常类	HTTP
（VPN 预检失败）	VpnCheckError	—
task_not_found	TaskNotFound	404
challenge_not_found	ChallengeNotFound	404
invalid_state	InvalidState	409
duplicate	DuplicateSubmit	409
resource_unavailable	ResourceUnavailable	503
internal_error	InternalError	500
（FastAPI 422）	ValidationError	422
（网络传输）	TSecConnectionError	—

VpnCheckError 的 message 固定为 VPN检测未通过,请检查靶场VPN网络配置，detail.reason 为失败原因（network_error / bad_status / bad_body / status_not_ok）。

from tsec_benchmark import (
    TSecBenchmark, TSecError, VpnCheckError, TaskNotFound, InvalidState,
    DuplicateSubmit, ResourceUnavailable, TSecConnectionError,
)

try:
    with TSecBenchmark(base_url="...", token="...") as client:  # 自动预检 VPN
        res = client.submit_flag(code, flag)
except VpnCheckError as e:
    print(e)                    # VPN检测未通过,请检查靶场VPN网络配置 —— 先连 VPN 再重试
except DuplicateSubmit:
    pass                        # 幂等：该 flag 之前已计入
except InvalidState as e:
    if "max active" in e.message:
        client.close_challenge(some_other)  # 释放一个名额，然后重试
    else:
        raise                  # 任务已结束（超时）—— 停止
except ResourceUnavailable:
    ...                        # 稍后重试启动，或跳过该题
except TSecError:
    raise                      # 其它任何平台错误

命令行冒烟测试

随包安装了一个小命令行工具，用于快速连通性检查：

tsec-run --base-url https://benchmark.example.com --token <BENCHMARK_TOKEN>
# 或通过环境变量：TSEC_BASE_URL / TSEC_TOKEN

它会列出跑分任务的所有题目及每题的 flag 进度——便于在把 SDK 接入你的 Agent 之前确认 token 和 base URL 可用。

字段映射（SDK ↔ API 文档）

SDK 数据类字段与平台 JSON 字段名完全一致（两边都是 snake_case）。

开发

cd integrations/python-sdk
pip install -e ".[dev]"
pytest            # 基于 respx 打桩的单测，覆盖全部 5 个端点 + 错误码

发布到 PyPI

包名 tsec-benchmark 在 PyPI 上可能已被占用——首次发布前请检查 https://pypi.org/project/tsec-benchmark/，如需更名请同步修改 pyproject.toml 中的 name 及对应的导入包名。

步骤（在 integrations/python-sdk/ 目录下执行）：

python -m build          # 生成 dist/tsec_benchmark-0.1.2-py3-none-any.whl + sdist
twine upload dist/*      # 需要在 ~/.pypirc 配置 PyPI API token，或设置 TWINE_PASSWORD 环境变量

每次发版时，请同时更新 pyproject.toml 和 tsec_benchmark/_version.py 中的 version。