Metadata-Version: 2.4
Name: tsec-benchmark
Version: 0.1.2
Summary: Python SDK for the TSec Benchmark Platform player-facing Challenges API.
Author: TSec Benchmark
License-Expression: MIT
Keywords: agent,benchmark,ctf,security,tsec
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Requires-Python: >=3.9
Requires-Dist: httpx>=0.27
Provides-Extra: cli
Requires-Dist: httpx>=0.27; extra == 'cli'
Provides-Extra: dev
Requires-Dist: build>=1.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Description-Content-Type: text/markdown

# tsec-benchmark

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

```bash
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()`。

```python
from tsec_benchmark import TSecBenchmark, VpnCheckError

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

## 快速开始

### 同步用法（最简单）

```python
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）

```python
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`）。

```python
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                      # 其它任何平台错误
```

## 命令行冒烟测试

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

```bash
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）。

## 开发

```bash
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/` 目录下执行）：

```bash
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`。
