deepcyber-ai/airt-harness
GitHub: deepcyber-ai/airt-harness
一个通用的AI红队测试框架，通过协议转换和统一接口连接各类红队工具与目标AI系统。
Stars: 0 | Forks: 0
# DeepCyber AI 红队测试框架
一个用于 AI 红队测试的通用测试框架。它位于您的红队工具和目标 AI 系统之间，提供统一的规范 API、协议转换、模拟仿真、情报收集和回归重放。
```
____ ______ __
/ __ \___ ___ ____ / ____/_ __/ /_ ___ _____
/ / / / _ \/ _ \/ __ \/ / / / / / __ \/ _ \/ ___/
/ /_/ / __/ __/ /_/ / /___/ /_/ / /_/ / __/ /
/_____/\___/\___/ .___/\____/\__, /_.___/\___/_/
/_/ /____/
AI Red Teaming Harness
```
```
pip install airt-harness
```
**一套框架，任意目标，任意工具。** 只需为目标的有线格式编写一次映射器，然后每一个红队工具——PyRIT、Promptfoo、Garak、Spikee、curl 或您自己的脚本——都可以与同一个规范 API 通信。在无需修改工具的情况下，即可在真实目标和本地模拟器之间切换。在修复后重放记录的漏洞会话，以进行回归测试。
### 核心能力
- **统一 API** —— 在任意目标前提供规范的 OpenAI 兼容 `/v1/chat/completions` 端点，无论其原生有线格式如何
- **协议转换** —— 双向消息映射器按目标处理身份验证、标头、会话管理和响应解析
- **模拟仿真** —— 本地模拟服务器，具有可插拔的 LLM 后端（Ollama、OpenAI、Anthropic、Gemini、DeepSeek、echo），使用目标的确切有线格式
- **情报收集** —— 将每个请求/响应记录为 JSONL，以便分析和重放
- **回归重放** —— 从情报日志或精选的 Metabase CSV 重放记录的攻击会话，比较响应，并使用 LLM 评判器进行评分
- **配置隔离** —— 每次演练的配置文件独立保存配置、日志和情报
## 架构
```
Red Team Tool / curl / Promptfoo / Replay
|
v
HARNESS (server.py)
- Canonical /v1/chat/completions endpoint (OpenAI-compatible)
- Simplified /chat endpoint
- Auth, retry, token refresh, intel collection
|
v
MESSAGE MAPPER (mappers/example.py, your_target.py, ...)
- Canonical <-> target wire format translation
- Bidirectional: client side + mock server side
|
v
TARGET (real API or mock server)
- Real: mTLS, bearer, API key -- whatever the target needs
- Mock: local LLM (ollama, openai, anthropic, gemini, deepseek, echo)
```
## 快速开始
该框架附带一个默认配置文件（**Deep Vault Capital**，一家虚构的英国金融公司），因此您可以立即开始，而无需任何配置、API 密钥或证书。
```
# 安装
pip install airt-harness
# 启动 harness 容器（echo 后端，无需 API 密钥）
MOCK_BACKEND=echo airt-launch
# 测试
curl http://localhost:8000/health
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"input": "What is an ISA?"}'
# 重放录制的会话以进行回归测试
airt-replay profiles/default/intel/ --list-sessions
airt-replay profiles/default/intel/ --session
# 停止
airt-stop
```
### 从源码（不使用 Docker）
```
pip install -r requirements.txt
# 使用默认配置文件启动 mock 服务器
python -m harness.mock --backend echo --port 8089
# 启动指向 mock 的 harness
BACKEND=mock uvicorn harness.server:app --port 8000
# 测试
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"input": "What is an ISA?"}'
```
## Docker
```
docker build -t deepcyber/airt-harness .
docker run -p 7860:7860 -p 8000:8000 -p 8089:8089 deepcyber/airt-harness
# 使用从镜像外部挂载的自定义配置文件：
docker run -p 8000:8000 \
-v ./profiles/myproject:/app/profiles/myproject \
-e PROFILE=profiles/myproject/profile.yaml \
deepcyber/airt-harness
```
## 端点
| Endpoint | Method | Purpose |
|---|---|---|
| `/v1/chat/completions` | POST | 规范的 OpenAI 兼容（与任何 OpenAI 兼容工具一起使用） |
| `/chat` | POST | 简化版：`{"input": "message"}` + 可选的 `x-session-id` 标头 |
| `/auth` | POST | 验证/刷新身份验证（mTLS 证书检查或令牌刷新） |
| `/firewall` | POST | 开启/关闭 HB Firewall（默认关闭） |
| `/init` | POST | 初始化会话（对于需要显式初始化的目标） |
| `/backend` | POST | 在运行时在真实后端和模拟后端之间切换 |
| `/token` | POST | 无需重启即可热交换 bearer token |
| `/health` | GET | 当前配置和状态 |
| `/intel/summary` | GET | 记录的 API 调用摘要 |
| `/v1/models` | GET | OpenAI 兼容的模型列表 |
## 切换配置文件
### 启动时（环境变量）
```
# 默认（Deep Vault Capital -- 无需配置）
uvicorn harness.server:app --port 8000
# 你的 engagement 配置文件
PROFILE=profiles/myproject/profile.yaml uvicorn harness.server:app --port 8000
```
模拟服务器使用相同的 `--profile` 标志：
```
python -m harness.mock --profile profiles/myproject/profile.yaml --backend echo
```
### 运行时（后端切换）
在无需重启的情况下在真实目标和模拟器之间切换：
```
# 切换到 mock
curl -X POST http://localhost:8000/backend -d '{"backend": "mock"}'
# 切换到真实目标
curl -X POST http://localhost:8000/backend -d '{"backend": "real"}'
# 切换（无 body）
curl -X POST http://localhost:8000/backend
```
后端状态在重启后持久保存（保存到配置文件目录中的 `.harness_state.json`）。
## 日志隔离
每个配置文件的数据都存储在其自己的目录中——不同的配置文件绝不会相互污染：
```
profiles/
myproject/
harness.log # Harness server log (this profile only)
mock_server.log # Mock server log (this profile only)
mock-audit.jsonl # Mock audit trail (this profile only)
intel/
responses.jsonl # Intel store (this profile only)
.harness_state.json # Backend state (this profile only)
```
如果需要，您可以在 `profile.yaml` 中覆盖日志和情报路径：
```
harness:
intel_dir: /custom/path/intel
log_file: /custom/path/harness.log
```
但建议使用默认值（配置文件目录）以保持隔离。
## 项目配置文件
每次演练都会获得一个配置文件目录：
```
profiles/
myproject/
profile.yaml # Target config (API, auth, TLS, session, mock)
mock/
system_prompt.txt # Optional custom system prompt for the mock server
certs/ # TLS client certificates (gitignored)
```
### 配置文件 YAML
```
target: example # mapper name (built-in or custom)
display_name: "My Target" # shown in GUI and health endpoint
# 对于位于此包之外的 engagement 专用 mapper：
# target_module: profiles.myproject.mapper
api:
url: "https://api.example.com"
path: "/v1/chat"
auth:
mode: none # none | bearer | api_key
bearer:
env_var: MY_BEARER_TOKEN
prefix: "Bearer "
token_refresh:
order: ["cli"] # ["cli"] | ["endpoint"] | ["cli", "endpoint"]
cli:
command: "gcloud auth print-access-token"
endpoint:
url: "" # remote token server URL
secret_env: TOKEN_SECRET
tls:
cert: certs/tls.crt
key: certs/tls.key
ca_bundle: null # null = no verify; path = custom CA
verify: false # true = system CA store
session:
header: x-session-id # header for session ID
prefix: "replay-"
mock:
url: "http://localhost:8089"
harness:
port: 8000
# intel_dir and log_file default to the profile directory
```
## 消息映射器
映射器在框架的规范格式和目标的有线协议之间进行转换。每个映射器都是双向的：
| Method | Side | Purpose |
|---|---|---|
| `build_request(message, session_id)` | Client | 规范 -> 目标请求 |
| `parse_response(data)` | Client | 目标响应 -> 规范格式 |
| `parse_incoming_request(body, headers)` | Server | 目标请求 -> 提取消息（用于模拟） |
| `build_mock_response(answer, session_id)` | Server | LLM 应答 -> 目标响应格式（用于模拟） |
该框架附带一个内置映射器作为起点：
| Mapper | Wire format | Use case |
|---|---|---|
| `example` | 扁平化 JSON `{"input": "..."}` 请求，`{"output": "...", "session_id": "..."}` 响应，会话 ID 在 `x-session-id` 标头中 | 通用基线；复制并改编 |
### 添加新的映射器
两种方式：
**1. 作为内置映射器（适合您想要发布的映射器）：**
1. 将 `harness/mappers/example.py` 复制到 `harness/mappers/mytarget.py`
2. 继承 `BaseMapper`，实现四个方法，设置 `name = "mytarget"`
3. 在 `mappers/__init__.py` 的 `_BUILTIN_MAPPERS` 中添加 `"mytarget": "harness.mappers.mytarget"`
4. 在您的配置文件中：`target: mytarget`
**2. 作为演练专用的私有映射器（保留在此仓库之外）：**
1. 创建一个带有 `create_mapper(config)` 函数的 `profiles/myproject/mapper.py`
2. 在您的配置文件 YAML 中，将 `target_module:` 指向可导入模块或文件路径：
target: myproject # 显示/日志记录名称
target_module: profiles/myproject/mapper.py # 文件路径（相对于 CWD）
# 或者：
# target_module: my_package.my_target_mapper # 可导入模块
3. 从文件或模块可访问的目录运行框架
这使您可以将特定于客户的映射器保留在私有仓库中，同时共享来自此仓库的框架脚手架。
### 通用映射器（仅限 YAML，无需 Python）
对于具有标准 REST API 的目标，请使用内置的 `generic` 映射器，而不是编写自定义映射器文件。完全在 profile.yaml 中配置请求正文模板和响应提取路径：
```
target: generic
api:
url: "https://api.example.com"
path: "/v1/chat"
# 请求体 —— {{PROMPT}} 替换为用户消息
request_template:
model: "gpt-4"
messages:
- role: "user"
content: "{{PROMPT}}"
# 用于从 JSON 响应中提取答案的点记法路径
response_path: "choices.0.message.content"
```
点表示法示例：
| Target response shape | `response_path` |
|---|---|
| `{"answer": "..."}` | `answer` |
| `{"responses": [{"value": "..."}]}` | `responses.0.value` |
| `{"choices": [{"message": {"content": "..."}}]}` | `choices.0.message.content` |
对于具有复杂有线格式（SSE、双重编码、自定义会话初始化）的目标，请改用自定义映射器。
## PyRIT 集成
该框架提供兼容 PyRIT 的目标，用于自动化的红队测试编排。
```
pip install airt-harness[pyrit]
```
### ProxyTarget —— 通过框架路由
```
from harness.pyrit import ProxyTarget
target = ProxyTarget(
harness_url="http://localhost:8000",
session_id="RedTeam-001",
)
# 与任何 PyRIT orchestrator 配合使用
from pyrit.orchestrator import PromptSendingOrchestrator
orchestrator = PromptSendingOrchestrator(objective_target=target)
```
`ProxyTarget` 将提示发送到框架 `/chat` 端点。该框架处理协议转换、身份验证、会话管理和情报收集——PyRIT 只看到一个简单的文本输入/文本输出目标。
### BedrockTarget —— 直接连接 AWS Bedrock
```
from harness.pyrit import BedrockTarget
target = BedrockTarget(
model_id="anthropic.claude-sonnet-4-6",
region="eu-west-2",
)
```
需要 `pip install airt-harness[bedrock]` 并配置 AWS 凭证。
## HB Firewall
使用 HumanBound Firewall 的可选输入筛选层。**默认关闭。** 启用后，每条消息在到达目标之前都会被评估——被阻止的消息将收到标准的拒绝响应。
### 运行时启用
```
# 启用
curl -X POST http://localhost:8000/firewall -d '{"enabled": true}'
# 禁用
curl -X POST http://localhost:8000/firewall -d '{"enabled": false}'
# 切换
curl -X POST http://localhost:8000/firewall
# 检查状态
curl http://localhost:8000/health | jq .firewall_enabled
```
### 配置文件配置（可选）
```
firewall:
enabled: false # default off
agent_config: "hb-firewall/agent.yaml" # HB security policy
model_path: null # optional Tier 2 model
```
### A/B 测试工作流
在关闭和开启防火墙的情况下运行相同的攻击会话，以衡量其有效性：
```
# 1. 在不开启 firewall 的情况下运行测试
airt-replay evidence/metabase.csv --session abc123 -o results/no-firewall.md
# 2. 启用 firewall
curl -X POST http://localhost:8000/firewall -d '{"enabled": true}'
# 3. 运行相同的测试
airt-replay evidence/metabase.csv --session abc123 -o results/with-firewall.md
# 4. 比较报告
```
需要 `pip install hb-firewall` —— 防火墙仅在首次启用时延迟加载。
## Token 刷新
两种独立的方法，按 `auth.token_refresh.order` 中配置的顺序尝试：
| Method | Config Key | Default | Use Case |
|---|---|---|---|
| `cli` | `auth.token_refresh.cli.command` | `gcloud auth print-access-token` | 本地开发、GCP、Azure、任何 CLI |
| `endpoint` | `auth.token_refresh.endpoint.url` | (无) | 通过 Cloudflare 隧道的远程令牌服务器 |
默认顺序为 `["cli"]`。要添加远程端点：
```
auth:
token_refresh:
order: ["cli", "endpoint"]
endpoint:
url: "https://token.example.com/token"
secret_env: TOKEN_SECRET
```
Token 刷新在聊天期间的 401/403 错误时自动触发，或通过 `POST /auth` 手动触发。
## 模拟服务器
模拟服务器使用与框架相同的映射器，因此它模拟目标的确切有线格式。任何对真实目标有效的工具对模拟器也同样有效。
```
# Echo 模式（快速，无 LLM）
python -m harness.mock --backend echo
# 本地 Ollama
python -m harness.mock --backend ollama --model llama3.2
# OpenAI
python -m harness.mock --backend openai --model gpt-4o-mini
# Anthropic
python -m harness.mock --backend anthropic
# Gemini
python -m harness.mock --backend gemini
# DeepSeek
python -m harness.mock --backend deepseek
```
API 密钥从环境变量中读取：`OPENAI_API_KEY`、`ANTHROPIC_API_KEY`、`GOOGLE_API_KEY` / `GEMINI_API_KEY`、`DEEPSEEK_API_KEY`。
## 聊天 GUI
一个基于 Gradio 的交互式聊天客户端，用于手动测试和探索。GUI 连接到框架，并自动从配置文件的 `display_name` 字段检测目标名称。
```
# 首先启动 harness
PROFILE=profiles/default/profile.yaml BACKEND=mock uvicorn harness.server:app --port 8000
# 启动 GUI（自动检测目标名称）
python -m harness.gui
# 自定义端口或公共链接
python -m harness.gui --port 7860 --share
```
在配置文件 YAML 中设置 `display_name` 以自定义 GUI 标题：
```
target: example
display_name: "My Target" # shown in GUI title and health endpoint
```
功能：
- 带有对话历史记录和会话管理的 **聊天面板**
- 显示完整目标响应的 **原始响应选项卡**（如果映射器将其保留在 `raw` 中，代理思维或护栏事件等特定于演练的字段会出现在此处）
- 用于验证/刷新身份验证的 **身份验证选项卡**
- 运行时在真实和模拟之间切换的 **后端切换**
- 显示累积 API 调用统计信息的 **情报选项卡**
- 显示框架配置的 **运行状况选项卡**
## 情报收集
通过框架的每个 API 调用都记录到 `/intel/responses.jsonl`：
```
{
"timestamp": "2026-04-07T14:30:00Z",
"session_id": "harness-abc123",
"backend": "real",
"target": "example",
"prompt": "What is an ISA?",
"answer": "An Individual Savings Account is...",
"raw": { "...": "full target response (engagement-specific fields live here)" },
"error": null
}
```
`raw` 字段保存目标返回的任何内容——包括任何自定义元数据、代理思维、护栏事件或其他特定于演练的数据。框架本身保持中立，不解析这些字段。
## 重放
针对正在运行的框架重放记录的会话，以进行回归测试。发送先前测试中的确切攻击者提示，并将新响应与原始响应进行比较。可以选择使用 LLM 评判器对每个轮次进行评分。
```
# 首先启动 harness（mock 或真实）
BACKEND=mock uvicorn harness.server:app --port 8000
```
### 两种源类型（自动检测）
| Source | Path | 工作原理 |
|---|---|---|
| **情报日志** | 目录或 `.jsonl` 文件 | 读取 `responses.jsonl`，按 session_id 分组，按时间戳排序 |
| **Metabase CSV** | `.csv` 文件 | 读取包含 `session_id`、`turn`、`request`/`prompt`、`answer`/`response` 列的 CSV |
情报日志是框架自动记录的内容（请参阅 [情报收集](#intel-collection)）。Metabase CSV 是精选的子集，通常在演练期间生成，包含用于回归测试的选定会话。
### 列出会话
```
# 来自 intel 日志
airt-replay profiles/default/intel/ --list-sessions
# 来自 metabase CSV
airt-replay evidence/metabase.csv --list-sessions
```
### 重放会话
```
airt-replay evidence/metabase.csv --session abc123
```
支持部分会话 ID 匹配。重放使用新的会话 ID 将每个提示发送到框架 `/chat` 端点，并将新响应与原始响应进行比较。
### 评判评估
使用 LLM 评判器对每个重放的轮次进行通过/失败（PASS/FAIL）评分：
```
airt-replay evidence/metabase.csv --session abc123 \
--evaluate \
--judge-config replay/judge_config.yaml \
--judge-prompts replay/judge_prompts.yaml \
-o results/regression-report.md
```
评判配置和评判提示是特定于演练的文件——框架提供了多提供商框架。支持的评判提供商：
| Provider | Config key | Notes |
|---|---|---|
| AWS Bedrock | `bedrock` | 零数据保留；需要 `boto3` |
| Ollama (local) | `ollama` | 完全物理隔离；没有数据离开您的机器 |
| Google Gemini | `gemini` | 直接 API |
| Anthropic Claude | `claude` | 直接 API |
| OpenAI-compatible | (默认) | vLLM、RunPod、LiteLLM 或任何 chat/completions 端点 |
### Metabase 格式兼容性
重放适用于任何具有这四列的 CSV：
| Column | Required | 映射到 |
|---|---|---|
| `session_id` | 是 | 将轮次分组为会话 |
| `turn` | 是 | 在会话内对轮次进行排序 |
| `request` 或 `prompt` | 是 | 攻击者提示（逐字重放） |
| `answer` 或 `response` | 是 | 原始响应（用于比较） |
其他列（finding_id、scenario_id、护栏评分等）会被通用重放引擎忽略。特定于演练的包装器可以使用它们进行导航和报告。
### CLI 参考
| Flag | Purpose |
|---|---|
| `source` (positional) | 情报目录、`.jsonl` 或 `.csv` 的路径（默认：`.`） |
| `--list-sessions` | 列出中的所有会话 |
| `--session ID` | 重放会话（支持部分匹配） |
| `--harness-url URL` | 框架 URL（默认：`$HARNESS_URL` 或 `http://localhost:8000`） |
| `--delay SECONDS` | 轮次之间的延迟（默认：1） |
| `--evaluate` | 对重放的响应运行 LLM 评判器 |
| `--judge-config FILE` | 评判配置 YAML 的路径 |
| `--judge-prompts FILE` | 评判提示 YAML 的路径 |
| `--judge-criteria KEY` | 将单个条件键应用于所有轮次 |
| `-o FILE` | 将比较报告保存到文件 |
## 许可证
版权所有 (c) 2026 Deep Cyber Ltd。
根据 Apache License 2.0 版获得许可。有关详细信息，请参阅 [LICENSE](LICENSE)。
标签：AI红队, API网关, CISA项目, Garak, LLM评估, Mock模拟, Ollama, OpenAI兼容, Promptfoo, PyRIT, 中间件, 会话回放, 协议转换, 多智能体系统, 多语言支持, 大模型安全, 安全测试, 安全测试框架, 对话日志, 情报收集, 攻击性安全, 攻击模拟, 数据泄露, 时序数据库, 漏洞回归, 漏洞研究, 请求拦截, 逆向工具, 驱动签名利用