tsec-benchmark 0.1.0

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

tsec-benchmark

Python SDK for the TSec Benchmark Platform player-facing Challenges API. Lets a Python-based evaluated Agent drive the benchmark flow — list challenges, start containers, get hints, submit flags, close containers — without hand-writing HTTP boilerplate.

pip install tsec-benchmark

• Python ≥ 3.9
• Dependency: httpx
• Async-first, with a synchronous wrapper for scripts

Prerequisites (same for all integration paths): before using this SDK you must already have, from the platform console/admin:

• a BENCHMARK_TOKEN for your benchmark task,
• the platform base URL (e.g. https://benchmark.example.com),
• a connected VPN (required to reach challenge container addresses).

This SDK covers the player-facing flow only — it does not create benchmark tasks or manage VPN.

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

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

GET http://healthcheck.tsecbench.com
→ {"client_ip":"10.0.0.6","status":"ok","time":"2026-06-24 16:27:33"}

该地址只在 VPN 网络内可达，能成功返回 status:"ok" 即证明 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网络配置

Quick start

Synchronous (simplest)

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)

Asynchronous (httpx-based)

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 reference

Five methods, mapping 1:1 to the platform endpoints (see ../api/CHALLENGES_API.md §5 for authoritative field specs). Each returns a frozen dataclass. The 步骤 column shows where each call sits in the solve loop. A pre-flight check_vpn() (步骤 0) runs automatically on context-manager entry.

步骤	Method	Endpoint	这一步在干什么	Returns
0	check_vpn()	GET http://healthcheck.tsecbench.com	一切流程开始前的 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

Dataclasses

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 都已提交正确）

StartResult

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

字段	类型	说明
unique_code	str	题目唯一标识码
container_addr	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）

Errors

Every platform error code raises a dedicated exception (all inherit TSecError, carrying .code, .message, .detail, .status_code):

Platform code	Exception	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
(transport)	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                        # idempotent: flag already counted
except InvalidState as e:
    if "max active" in e.message:
        client.close_challenge(some_other)  # free a slot, then retry
    else:
        raise                  # task ended (timeout) — stop
except ResourceUnavailable:
    ...                        # retry start later, or skip
except TSecError:
    raise                      # any other platform error

CLI smoke test

A tiny CLI is installed with the package for a quick connectivity check:

tsec-run --base-url https://benchmark.example.com --token <BENCHMARK_TOKEN>
# or via env: TSEC_BASE_URL / TSEC_TOKEN

It lists the task's challenges and per-challenge flag progress — useful to confirm your token and base URL work before wiring the SDK into your Agent.

Field mapping (SDK ↔ API doc)

Authoritative definitions live in ../api/CHALLENGES_API.md. The SDK dataclass fields match the platform JSON field names exactly (snake_case on both sides).

Development

cd integrations/python-sdk
pip install -e ".[dev]"
pytest            # respx-stubbed unit tests for all 5 endpoints + error codes

Publishing to PyPI

The package name tsec-benchmark may already be taken on PyPI — check https://pypi.org/project/tsec-benchmark/ before first publish, and rename if needed (updating name in pyproject.toml and the import package accordingly).

Steps (run from integrations/python-sdk/):

python -m build          # produces dist/tsec_benchmark-0.1.0-py3-none-any.whl + sdist
twine upload dist/*      # requires a PyPI API token in ~/.pypirc or TWINE_PASSWORD env

Bump version in pyproject.toml and tsec_benchmark/_version.py together for each release.