提交 001a68a9 authored 作者: 陈泽健's avatar 陈泽健

feat(security): 新增API接口安全测试模块

后端安全测试引擎(与PlaywrightExecutor解耦,基于requests+AES-CBC签名):
- 新增 SecurityConfig/VulnerabilityResult ORM 与 Pydantic schema
- 新增 HttpClient(AES-CBC签名算法,X-RANDOM/X-TIMESTAMP/X-SIGN)
- 新增 AuthHelper 多账号登录辅助
- 新增 SecurityExecutor 安全测试执行引擎
- 新增 security_service 业务逻辑层与 knowledge_base 知识库
- 新增 /api/security 路由(11个接口)

执行调度分流修复:
- execution_service.run_execution 按 case_type 分流,security 走专用 _run_security_execution
- 安全用例不再误交给 PlaywrightExecutor 解析 dict steps
- test_case schema steps 字段支持 Union(list/dict)

前端适配:
- 新增 types/security.ts 与 api/security.ts
- Cases.vue 支持安全用例风险等级标签与详情弹窗
- Execution.vue 执行弹窗分流 security/UI 执行逻辑

配套脚本与文档:
- create_security_cases / create_default_security_config / test_security_e2e
- PRD 需求文档、计划执行文档、报告设计文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 d3587c55
# 【安全测试】执行计划 - 修复安全测试执行报错
> **文档类型**: 计划执行文档(安全测试模块)
> **创建日期**: 2026-07-21
> **关联问题**: `_问题处理_安全测试执行报错str对象无get属性.md`
> **执行策略**: 直接修复,无需分阶段
---
## 一、执行目标
修复安全测试模块执行时的 3 个问题,使安全测试用例能正常执行并保存结果。
| 修复点 | 优先级 | 目标 |
|--------|--------|------|
| 1. ORM 对象在 run_in_executor 中脱钩 | P0 | 安全测试用例正常执行 |
| 2. metadata 字段名冲突 | P0 | 漏洞结果正常保存 |
| 3. 前端 metadata 引用同步 | P0 | 接口返回正常 |
---
## 二、任务清单
### 2.1 后端修复
| # | 任务 | 文件 | 说明 |
|---|------|------|------|
| 1.1 | 重构 run_execution 执行逻辑 | `backend/app/services/security_service.py` | 在 async 上下文内预先转纯字典 |
| 1.2 | 修复字段名 metadata → extra_data | `backend/app/services/security_service.py` | `_save_vulnerability_result` |
| 1.3 | 修复 schema 字段名 | `backend/app/schemas/security.py` | `VulnerabilityResultResponse` |
### 2.2 前端修复
| # | 任务 | 文件 | 说明 |
|---|------|------|------|
| 2.1 | 同步类型定义 | `frontend/src/types/security.ts` | `VulnerabilityResult` 字段 |
### 2.3 验证
| # | 任务 | 说明 |
|---|------|------|
| 3.1 | 重启后端服务 | 加载新代码 |
| 3.2 | 执行安全测试 | 全部用例不再 error |
| 3.3 | 查看执行结果 | 漏洞结果正常保存 |
---
## 三、详细改动
### 3.1 后端:security_service.py(已实施)
#### 改动 A:run_execution 中预先转纯字典
**位置**: `run_execution()` 方法
**改动前**
```python
def sync_run():
executor.start()
return executor.execute_cases(
[{"id": c.id, "name": c.name, "steps": c.steps} for c in cases],
...
)
results = await loop.run_in_executor(None, sync_run)
```
**改动后**
```python
import json as _json
def _to_plain_steps(steps_val):
"""把 steps 字段统一转成 dict(兼容 str/dict/None)"""
if isinstance(steps_val, str):
try:
parsed = _json.loads(steps_val)
return parsed if isinstance(parsed, dict) else {}
except Exception:
return {}
if isinstance(steps_val, dict):
return steps_val
return {}
# 在 async 上下文内(run_in_executor 之前)就完成转换
plain_cases = [
{"id": c.id, "name": c.name, "steps": _to_plain_steps(c.steps)}
for c in cases
]
client_config = config.to_client_config()
def sync_run():
executor = SecurityExecutor(client_config)
executor.start()
return executor.execute_cases(plain_cases, ...)
results = await loop.run_in_executor(None, sync_run)
```
**原理**:线程池闭包只捕获纯字典数据,不再持有 ORM 对象引用,避免脱钩。
#### 改动 B:_save_vulnerability_result 字段名
**改动前**
```python
vuln_result = VulnerabilityResult(
...
metadata=result.metadata,
)
```
**改动后**
```python
vuln_result = VulnerabilityResult(
...
extra_data=result.metadata,
)
```
### 3.2 后端:schemas/security.py(已实施)
**改动**`VulnerabilityResultResponse``metadata: Dict[str, Any]``extra_data: Dict[str, Any]`
### 3.3 前端:types/security.ts(待实施)
**改动**`VulnerabilityResult` 接口中 `metadata: Record<string, any>``extraData: Record<string, any>`
---
## 四、验证检查清单
### 4.1 后端验证
- [ ] `security_service.py``run_execution` 已用 `_to_plain_steps` 预转换
- [ ] `security_service.py``_save_vulnerability_result` 已用 `extra_data`
- [ ] `schemas/security.py``VulnerabilityResultResponse` 已用 `extra_data`
### 4.2 执行验证
- [ ] 重启 8001 后端服务
- [ ] 前端 `/execution/security` 执行用例
- [ ] 全部用例状态为 passed/failed(不再 error)
- [ ] 执行记录可展开查看漏洞结果
### 4.3 时间显示(记录备案,不修复)
- [ ] 全项目时间显示问题单独评估(UTC → 本地时区转换)
---
## 五、风险评估
| 风险 | 影响 | 缓解 |
|------|------|------|
| 预转换逻辑遗漏嵌套字段 | 用例执行失败 | `_to_plain_steps` 兼容 str/dict/None 三种情况 |
| 前端未同步 extraData | 详情显示异常 | 前端 types 同步修改 |
| 现有执行记录数据 | 无 | 仅影响新执行,旧数据不受影响 |
---
*本文档为安全测试执行报错修复的计划执行文档。*
# 【安全测试】问题处理 - 安全测试执行报错 'str' object has no attribute 'get'
> **文档类型**: 问题处理文档(安全测试模块)
> **创建日期**: 2026-07-21
> **最后更新**: 2026-07-21(修正根因)
> **问题等级**: P0(阻塞,安全测试全部执行失败)
> **关联模块**: 安全测试(security)
> **状态**: ✅ 已修复并验证
---
## 一、问题现象
在执行中心执行安全测试用例时,**全部用例执行失败**,前端报错:
```
执行异常: 'str' object has no attribute 'get'
```
后端日志关键行:
```
app.executors.playwright_executor - ERROR - 用例执行异常: 水平越权访问其他用户会议详情, 'str' object has no attribute 'get'
```
同时观察到执行记录的**时间显示为 UTC 时间**(如本地 17 点执行,显示为 9 点)。
---
## 二、根因分析(已修正)
### 2.1 ⚠️ 真正根因:安全测试用例被误交给 PlaywrightExecutor 执行
**最初误判**:以为是 `run_in_executor` 线程池导致 ORM 对象脱钩、JSON 字段退化为字符串。
**实际真相**(通过后端日志确认):
错误来源是 **`playwright_executor`**,而不是我新写的 `security_executor`。这说明安全测试用例**根本没有走安全测试执行器**,而是走了 UI 用例的通用执行链路。
**执行链路**
```
前端 Cases.vue 点"执行选中"
└─ executionApi.create() # 通用执行接口
└─ executionApi.run() # 通用执行触发
└─ ExecutionService.run_execution()# 后端统一入口
└─ PlaywrightExecutor # ⚠️ 无脑用 UI 执行器
└─ 解析 steps dict # 安全用例 steps 是 dict,不是步骤列表
└─ dict.get('action') # ❌ str 没 .get() → 报错
```
**根因**
1. 前端从「用例管理页 Cases.vue」点"执行选中",调用的是**通用执行 API**`/api/executions`),不是安全测试专用 API(`/api/security/executions`
2. 后端 `ExecutionService.run_execution()` **没有按 case_type 分流**,所有用例都交给 `PlaywrightExecutor`
3. 安全测试用例的 `steps` 是 dict(`{"test_type":"api_security", ...}`),而 `PlaywrightExecutor` 期望的是步骤列表 `[{order, action, ...}]`
4. Playwright 解析 dict 时 `dict[xxx].get(...)``str.get(...)` 触发 `'str' object has no attribute 'get'`
**验证**
```bash
# 端到端测试脚本 test_security_e2e.py
# 修复前:日志显示 playwright_executor 报错
# 修复后:日志显示 security_executor 执行,3/3 用例正常完成
```
### 2.2 次要问题:模型字段名 `metadata` 与 SQLAlchemy 保留字冲突
用户/linter 已将 `VulnerabilityResult.metadata` 改名为 `extra_data``metadata` 是 SQLAlchemy 保留字段名),但以下代码未同步:
- `security_service.py``_save_vulnerability_result()` 仍用 `metadata=result.metadata`
- `schemas/security.py``VulnerabilityResultResponse` 仍用 `metadata`
这会导致**漏洞结果保存失败****接口返回报错**
### 2.3 防御性增强:ORM 对象在 run_in_executor 中脱钩
虽然真正根因不是脱钩,但作为**防御性增强**,在丢入线程池前把 ORM 对象转纯字典仍然是好实践,避免后续 session 行为变化引入隐患。
### 2.4 时间显示问题(已有行为,非安全测试独有)
全项目统一使用 `datetime.utcnow()` 存储 UTC 时间,前端直接显示未做 +8 时区转换。
- 数据库存 UTC 时间(如 09:00 UTC = 17:00 本地)
- 前端按字符串直接显示,显示为 09:00
这是**全项目既有行为**(会议管理、数据分析等所有模块都一样),不属于本次安全测试引入的问题,记录在案但**不在本次修复范围**(修复需统一改全项目时间显示逻辑,影响面大)。
---
## 三、修复方案
### 3.1 修复点 1(核心):execution_service 按 case_type 分流(P0)
**文件**: `backend/app/services/execution_service.py``run_execution()`
**改动**:在 `run_execution()` 入口处判断 `case_type`,安全测试分流到专用方法:
```python
async def run_execution(self, execution_id, config=None):
execution = await self.get_execution(execution_id)
if not execution:
raise ValueError(f"执行记录不存在: {execution_id}")
# ===== 按用例类型分流:安全测试走专用执行器 =====
if execution.case_type == "security":
return await self._run_security_execution(execution_id, config)
# 以下为原有 UI 用例执行逻辑(PlaywrightExecutor)...
```
**新增方法** `_run_security_execution()`
1. 从执行记录取关联用例(按模块分组排序)
2. 加载默认安全测试配置(`SecurityConfig`
3. 在 async 上下文内预先把 ORM 对象转纯字典(防御脱钩)
4. `run_in_executor` 调用 `SecurityExecutor` 执行
5. 结果写回 `CaseResult`(供执行中心展示)+ `VulnerabilityResult`(供报告中心)
6. 广播 WebSocket 执行完成
### 3.2 修复点 2:模型字段名同步 `metadata` → `extra_data`(P0)
| 文件 | 改动 |
|------|------|
| `services/security_service.py` | `_save_vulnerability_result()``metadata=result.metadata``extra_data=result.metadata` |
| `services/execution_service.py` | `_run_security_execution()``metadata=r.metadata``extra_data=r.metadata` |
| `schemas/security.py` | `VulnerabilityResultResponse``metadata``extra_data` |
### 3.3 修复点 3:前端 `metadata` 引用同步(P0)
| 文件 | 改动 |
|------|------|
| `frontend/src/types/security.ts` | `VulnerabilityResult.metadata``extraData` |
### 3.4 时间显示问题(不在本次修复范围,记录备案)
全项目使用 UTC 时间存储,若需修复需统一在前端 `formatTime` 工具函数中做时区转换,影响所有页面,单独评估。
---
## 四、验证结果(已自测通过)
端到端测试脚本:`backend/scripts/test_security_e2e.py`
```
=== 1. 创建执行 ===
创建状态码: 201
执行ID: exec_xxx
case_type: security
=== 2. 触发执行 ===
触发状态码: 200
{"message":"执行完成","status":"completed"}
=== 3. 等待执行完成 ===
轮询 1: 3/3 完成
=== 4. 执行结果 ===
Nacos未授权访问: failed | Nacos控制台可匿名访问
Swagger文档暴露: passed |
SQL注入-会议查询接口: passed |
```
**验证结论**
- ✅ 安全用例正确走 SecurityExecutor(不再走 PlaywrightExecutor)
- ✅ 3/3 用例完成,无 `'str' object has no attribute 'get'` 错误
- ✅ 漏洞判定正确(Nacos 未授权=漏洞,Swagger/SQL注入=已防护)
- ✅ 结果同时写入 CaseResult 和 VulnerabilityResult
---
## 五、踩坑记录(修正后)
| # | 现象 | 根因 | 正确做法 |
|---|------|------|---------|
| S1 | 安全测试执行报 `'str' object has no attribute 'get'` | `execution_service` 未按 case_type 分流,安全用例被交给 PlaywrightExecutor 解析 dict steps | **在 `run_execution` 入口按 `case_type` 分流**,security 走专用 `_run_security_execution` |
| S2 | `metadata` 字段保存/查询报错 | `metadata` 是 SQLAlchemy 保留字段名 | 模型用 `extra_data`,service/schema 同步 |
| S3 | 执行时间显示比实际少 8 小时 | 全项目用 `utcnow()` 存 UTC 时间,前端未转时区 | 全项目既有行为,需统一在前端做时区转换(单独评估) |
| S4 | 误判为"ORM 脱钩" | 调试时用同步 engine 复现不出,未直接看生产日志 | **排查异常时先看后端日志的错误来源模块**,不要只看错误消息臆测 |
---
## 六、教训
1. **排查异常先看日志来源模块**:日志明确写了 `app.executors.playwright_executor`,我一开始却以为是 `security_executor` 的脱钩问题,浪费了时间。**应该先 grep 错误消息确认是哪个模块报的**
2. **多入口要统一分流**:安全测试有两个执行入口(Cases.vue 的通用执行 + Execution.vue 的专用执行),后端 `run_execution` 必须在入口统一分流,不能假设前端会调用专用 API。
3. **防御性编程**:即便不是真正根因,在 `run_in_executor` 前把 ORM 转纯字典也是好习惯,避免 session 行为变化引入隐患。
---
*本文档为安全测试模块问题处理记录,供后续维护参考。*
# 【安全测试】报告中心 - 报告设计文档(分析,不写代码)
> **文档类型**: 设计文档(安全测试报告)
> **创建日期**: 2026-07-21
> **参考实现**: `临时目录/安全测试/ApiSecurityTest/utils/report_generator.py`
> **状态**: 设计阶段,未实现
---
## 一、现状分析
### 1.1 当前平台报告中心的问题
平台报告中心(`/reports/security`)目前**只存原始执行数据**`CaseResult` + `VulnerabilityResult`),展示方式与 UI 测试报告一致:
- 展示执行记录列表 + 用例结果表格
- 每条结果只显示:用例名、状态、耗时、错误信息
- **缺少**:风险等级统计、OWASP 覆盖矩阵、漏洞详情(请求/响应/复现步骤/修复建议)、历史回归验证、华为红线合规检查
### 1.2 参考实现的报告优势
参考实现 `report_generator.py` 生成一份**结构化 Markdown 总报告**,包含 7 大章节,直接可用于交付:
- 执行概要 + OWASP 覆盖矩阵
- 漏洞详情(按风险等级排序,含复现步骤)
- 已验证安全项
- 风险评估与修复建议
- 已知安全问题清单
- 历史漏洞回归验证
- 华为安全红线合规性检查
---
## 二、参考实现的报告结构(详细分析)
### 2.1 报告命名规则
```
{服务器IP}_安全测试报告_{时间戳}.md
例:192.168.5.44_安全测试报告_20260721_143025.md
```
### 2.2 报告章节结构
#### 标题与基本信息
```
# 接口安全测试总报告
> 生成时间:2026-07-21 14:30:25
> 测试目标:https://192.168.5.44
> 测试标准:OWASP API Security Top 10 (2019)
> 测试工具:ApiSecurityTest
```
#### 一、执行概要
**1.1 测试统计总览**(表格):
| 统计项 | 数值 |
|--------|------|
| 测试用例总数 | 90 |
| 发现漏洞总数 | 20 |
| 🔴 高危漏洞 | 8 |
| 🟠 中危漏洞 | 12 |
| 🟡 低危漏洞 | 0 |
| 🔵 信息类 | 0 |
| 🟢 已验证安全 | 69 |
| 测试时间 | 14:00:00 ~ 14:30:25 |
**1.2 OWASP API Security Top 10 覆盖矩阵**(表格):
| 编号 | 安全风险 | 用例数 | 🔴高危 | 🟠中危 | 🟡低危 | 🔵信息 | 🟢安全 |
|------|---------|--------|--------|--------|--------|--------|--------|
| API1 | 对象级别授权失效 | 8 | 2 | 1 | 0 | 0 | 5 |
| API2 | 身份认证失效 | 13 | 3 | 4 | 0 | 0 | 6 |
| ... | | | | | | | |
| **合计** | | **90** | **8** | **12** | **0** | **0** | **69** |
> 按模块统计每个 OWASP 类别的漏洞分布,一眼看出哪个类别风险最高。
#### 二、漏洞详情
按风险等级排序(critical > high > medium > low > info),每个漏洞包含:
```markdown
### 2.1 [高危] 水平越权访问其他用户会议详情
- **用例编号**: 2.1.1
- **风险等级**: 🔴 高危
- **漏洞描述**: user 使用 Token 成功访问了 superadmin 的会议详情,存在水平越权漏洞
**复现步骤 — 请求信息:**
请求方法: GET
请求路径: /api/message/getMessageById
查询参数: {"id": "12345"}
使用 Token: user (越权访问 superadmin 的会议 ID: 12345)
**响应信息:**
状态码: 200
响应体: {"code":200,"data":{"id":12345,"title":"管理员会议"...}}
**修复建议**: 在接口中增加用户身份校验,确保用户只能访问自己创建的会议数据
---
```
> 含完整的请求/响应信息,可直接复现漏洞。
#### 三、已验证安全项
按 OWASP 模块分组,展示通过的测试用例:
```markdown
### API1 对象级别授权失效
| 编号 | 测试用例 | 结果 |
|------|---------|------|
| 2.1.5 | 水平越权访问运维集控设备接口 | ✅ 安全 |
| 2.1.6 | 水平越权跨公司数据访问 | ✅ 安全 |
```
#### 四、风险评估与修复建议
按风险等级给出通用修复方案:
```markdown
### 4.1 高危风险(需立即修复)
#### 越权漏洞(API1 / API3)
- 问题描述:普通用户可访问其他用户的数据
- 影响范围:用户数据泄露、数据篡改
- 修复建议:
1. 后端接口增加严格的资源所有权校验
2. 实现基于 RBAC 的权限控制
3. 在 API 网关层增加统一的权限拦截器
#### 身份认证缺陷(API2)
- 问题描述:Token 伪造可用、暴力破解无限制
- 修复建议:
1. 加强 Token 校验机制
2. 实现 Token 黑名单机制
3. 登录接口增加速率限制
4. 验证码应随机生成
```
#### 五、已知安全问题清单
测试前已确认的问题(非本次测试发现,来自需求文档/抓包):
| 编号 | 问题描述 | 风险等级 | 来源 |
|------|---------|---------|------|
| KN-01 | 维护平台验证码固定为 csba | 🟠中危 | 需求文档 |
| KN-02 | 登录密码 SHA256 无加盐 | 🟠中危 | 网络抓包 |
| KN-03 | company_secret 明文出现在 URL | 🔴高危 | 网络抓包 |
#### 六、历史漏洞回归验证
基于 35 个历史漏洞报告的回归测试结果:
| 编号 | 漏洞来源 | 测试项 | 当前状态 | 风险等级 |
|------|---------|--------|---------|---------|
| HV-001 | 长安深蓝汽车 | SQL注入-会议预定接口 | ✅ 已修复 | 高危 |
| HV-005 | 南山区委 | Nacos未授权访问 | 🔴 未修复 | 高危 |
| HV-026 | 新统一平台 | NoSQL注入-登录接口 | 🔴 未修复 | 严重 |
#### 七、华为安全红线合规性检查
22 项华为安全红线的合规状态:
| 检查编号 | 检查类别 | 要求 | 测试结果 | 合规状态 |
|---------|---------|------|---------|---------|
| HW-01 | 加密规范 | 密码存储加盐哈希 | SHA256无盐 | 🔴 不合规 |
| HW-06 | 鉴权机制 | 注销后Token立即失效 | Token仍可用 | 🔴 不合规 |
| HW-10 | 配置管理 | Swagger不暴露 | 已拦截 | ✅ 合规 |
---
## 三、平台报告设计方案(待实现)
### 3.1 数据来源
平台已有数据可直接复用,无需新增采集:
| 报告章节 | 数据来源 |
|---------|---------|
| 执行概要 | `VulnerabilityResult` 按 level 统计 |
| OWASP 覆盖矩阵 | `VulnerabilityResult.test_id` 前缀分组 |
| 漏洞详情 | `VulnerabilityResult`(request_info/response_info/fix_suggestion) |
| 已验证安全项 | `VulnerabilityResult` where is_vulnerable=False |
| 历史漏洞回归 | `VulnerabilityResult.extra_data` 标记 is_regression |
| 华为红线检查 | `VulnerabilityResult.extra_data` 标记 is_huawei_redline |
| 已知问题清单 | 知识库 `KNOWN_ISSUES`(静态数据) |
### 3.2 生成时机
安全测试执行完成(`_run_security_execution` 末尾)自动生成 Markdown 报告:
- 路径:`backend/data/reports/{server_ip}_安全测试报告_{timestamp}.md`
- 同时在 `Execution` 记录存储报告路径(`config.report_path`
### 3.3 后端实现要点
**新增** `backend/app/services/security_report_service.py`(迁移参考实现 `report_generator.py`):
| 参考实现 | 平台适配 |
|---------|---------|
| `VulnResult` 数据类 | 直接用 `VulnerabilityResult` ORM |
| `config.yaml` 读 target | 从 `SecurityConfig` 读 target_url |
| `KNOWLEDGE_BASE` 常量 | 从 `services/knowledge_base.py` 导入 |
| 独立运行 | 在 `_run_security_execution` 末尾调用 |
| 网盘上传 | 可选,先不做 |
**新增 API**`routers/security.py`):
- `GET /api/security/executions/{id}/report` — 返回报告内容(Markdown)
- `GET /api/security/executions/{id}/report/download` — 下载 .md 文件
### 3.4 前端实现要点
**修改** `frontend/src/views/Reports.vue`(或 `views/security/ReportTab.vue`):
| 区域 | 实现 |
|------|------|
| 报告列表 | 展示安全测试执行记录,点击进入报告详情 |
| 报告预览 | 用 `markdown-it` 渲染 Markdown(需加依赖) |
| 下载按钮 | 调用 download API 下载 .md |
### 3.5 风险等级体系对齐
参考实现用中文等级(高危/中危/低危/信息类),平台用英文(critical/high/medium/low/info)。报告生成时需映射:
| 平台 level | 报告显示 | emoji |
|-----------|---------|-------|
| critical | 严重 | 🔴🔴🔴 |
| high | 高危 | 🔴 |
| medium | 中危 | 🟠 |
| low | 低危 | 🟡 |
| info | 信息 | 🔵 |
---
## 四、实现优先级
| 优先级 | 任务 | 工作量 |
|--------|------|--------|
| P0 | 后端报告生成服务(7章节 Markdown) | 中 |
| P0 | 后端报告查看/下载 API | 小 |
| P1 | 前端报告列表 + Markdown 预览 | 中 |
| P2 | 网盘上传(参考实现功能) | 小 |
| P2 | ERP 任务创建(参考实现功能) | 中 |
---
## 五、与参考实现的差异
| 对比项 | 参考实现 | 平台方案 |
|--------|---------|---------|
| 运行方式 | 独立 Python 脚本 | 集成到平台,执行后自动生成 |
| 报告存储 | 本地 reports/ 目录 | 平台 data/reports/ + DB 记录路径 |
| 数据来源 | 内存 VulnResult 列表 | DB VulnerabilityResult 表 |
| 查看方式 | 打开 .md 文件 | 前端在线预览 + 下载 |
| 网盘/ERP | 有 | 暂不做(可选) |
---
*本文档为安全测试报告的设计分析,待实现时参考。*
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:auth_helper.py
模块描述:安全测试认证辅助模块,提供多账号登录、Token 管理等功能
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
import logging
from typing import Dict, Optional
logger = logging.getLogger(__name__)
class AuthHelper:
"""
认证辅助类,管理多账号登录和 Token 获取
支持三个测试账号(superadmin/admin/user)的登录和 Token 管理,
用于越权测试场景中获取不同角色的 Token。
"""
# 支持的账号键名
ACCOUNT_KEYS = ["superadmin", "admin", "user"]
def __init__(self, http_client):
"""
初始化认证辅助
Args:
http_client: HttpClient 实例
"""
self.client = http_client
def login_all_accounts(self) -> Dict[str, Optional[str]]:
"""
使用全部配置账号登录
Returns:
dict: {账号键名: Token} 的字典,失败的账号值为 None
"""
results = {}
for account_key in self.ACCOUNT_KEYS:
logger.info(f"尝试登录账号: {account_key}")
token = self.client.login(account_key)
results[account_key] = token
if token:
logger.info(f"账号 [{account_key}] 登录成功")
else:
logger.warning(f"账号 [{account_key}] 登录失败")
return results
def get_superadmin_token(self) -> Optional[str]:
"""获取超管 Token"""
return self.client.get_token("superadmin")
def get_admin_token(self) -> Optional[str]:
"""获取管理员 Token"""
return self.client.get_token("admin")
def get_user_token(self) -> Optional[str]:
"""获取普通用户 Token"""
return self.client.get_token("user")
def get_token(self, account_key: str) -> Optional[str]:
"""
获取指定账号的 Token
Args:
account_key: 账号键名
Returns:
str: Token 字符串,失败返回 None
"""
return self.client.get_token(account_key)
def refresh_all_tokens(self) -> Dict[str, Optional[str]]:
"""
刷新所有 Token(清除缓存后重新登录)
Returns:
dict: {账号键名: 新Token}
"""
self.client.clear_token()
return self.login_all_accounts()
def get_tokens_dict(self) -> Dict[str, str]:
"""
获取所有已缓存的 Token
Returns:
dict: {账号键名: Token}
"""
return dict(self.client._tokens)
def test_account(self, account_key: str) -> Dict[str, object]:
"""
测试指定账号是否可以登录
Args:
account_key: 账号键名
Returns:
dict: {"success": bool, "token_preview": str, "error": str}
"""
# 清除旧 Token,强制重新登录
self.client.clear_token(account_key)
token = self.client.login(account_key)
if token:
return {
"success": True,
"token_preview": f"{token[:20]}...{token[-10:]}",
"error": ""
}
else:
return {
"success": False,
"token_preview": "",
"error": f"账号 {account_key} 登录失败"
}
此差异已折叠。
此差异已折叠。
......@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings
from app.database import init_db
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security
# 配置日志
logging.basicConfig(
......@@ -140,6 +140,12 @@ app.include_router(
tags=["依赖管理"]
)
app.include_router(
security.router,
prefix="/api/security",
tags=["安全测试"]
)
# ==================== 根路径 ====================
......
......@@ -13,6 +13,8 @@ from app.models.test_case import TestCase
from app.models.execution import Execution
from app.models.case_result import CaseResult
from app.models.case_dependency import CaseDependency
from app.models.security_config import SecurityConfig
from app.models.vulnerability_result import VulnerabilityResult
__all__ = [
"Module",
......@@ -20,4 +22,6 @@ __all__ = [
"Execution",
"CaseResult",
"CaseDependency",
"SecurityConfig",
"VulnerabilityResult",
]
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_config.py
模块描述:安全测试配置数据库模型定义
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Text, JSON, DateTime, Boolean, Integer
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
class SecurityConfig(Base):
"""
安全测试配置数据库模型
存储安全测试的目标服务器、测试账号、认证配置等信息。
Attributes:
id (str): 配置唯一标识
name (str): 配置名称
target_url (str): 目标服务器地址
verify_ssl (bool): 是否验证SSL证书
timeout (int): 请求超时时间(秒)
accounts (dict): 测试账号配置(JSON)
auth_config (dict): 认证配置(JSON)
rate_limits (dict): 限流配置(JSON)
is_default (bool): 是否为默认配置
created_at (datetime): 创建时间
updated_at (datetime): 更新时间
"""
__tablename__ = "security_configs"
id: Mapped[str] = mapped_column(String(32), primary_key=True, comment="配置ID")
name: Mapped[str] = mapped_column(String(100), nullable=False, comment="配置名称")
target_url: Mapped[str] = mapped_column(String(500), nullable=False, comment="目标服务器地址")
server_ip: Mapped[Optional[str]] = mapped_column(String(50), default="", comment="服务器IP")
verify_ssl: Mapped[bool] = mapped_column(Boolean, default=False, comment="是否验证SSL证书")
timeout: Mapped[int] = mapped_column(Integer, default=30, comment="请求超时时间(秒)")
# 测试账号配置(JSON)
# 格式: {"superadmin": {"username": "...", "password": "..."}, "admin": {...}, "user": {...}, "captcha": "csba"}
accounts: Mapped[dict] = mapped_column(JSON, default=dict, comment="测试账号配置")
# 认证配置(JSON)
# 格式: {"token_type": "accessToken", "mechanism": "JWT", "login_path": "/platform/api/auth/login", ...}
auth_config: Mapped[dict] = mapped_column(JSON, default=dict, comment="认证配置")
# 限流配置(JSON)
# 格式: {"brute_force_max": 20, "rate_limit_max": 100, "request_interval": 0.5}
rate_limits: Mapped[dict] = mapped_column(JSON, default=dict, comment="限流配置")
# API路径前缀(JSON)
# 格式: {"meeting": "/api/", "monitor": "/monitor/api2/api/", ...}
api_prefixes: Mapped[dict] = mapped_column(JSON, default=dict, comment="API路径前缀")
is_default: Mapped[bool] = mapped_column(Boolean, default=False, comment="是否为默认配置")
description: Mapped[Optional[str]] = mapped_column(Text, default="", comment="配置描述")
created_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="创建时间"
)
updated_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, comment="更新时间"
)
def __repr__(self) -> str:
"""字符串表示"""
return f"<SecurityConfig(id={self.id}, name={self.name})>"
def to_dict(self) -> dict:
"""
转换为字典
Returns:
dict: 配置数据字典
"""
return {
"id": self.id,
"name": self.name,
"target_url": self.target_url,
"server_ip": self.server_ip,
"verify_ssl": self.verify_ssl,
"timeout": self.timeout,
"accounts": self.accounts or {},
"auth_config": self.auth_config or {},
"rate_limits": self.rate_limits or {},
"api_prefixes": self.api_prefixes or {},
"is_default": self.is_default,
"description": self.description or "",
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
def to_client_config(self) -> dict:
"""
转换为 HttpClient 可用的配置格式(兼容参考实现 config.yaml 结构)
Returns:
dict: HttpClient 配置字典
"""
return {
"target": {
"base_url": self.target_url,
"server_ip": self.server_ip or self.target_url.replace("https://", "").replace("http://", "").split(":")[0],
"verify_ssl": self.verify_ssl,
"timeout": self.timeout,
},
"accounts": self.accounts or {},
"auth": self.auth_config or {},
"limits": self.rate_limits or {},
"api_prefixes": self.api_prefixes or {},
}
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:vulnerability_result.py
模块描述:漏洞测试结果数据库模型定义
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Text, JSON, DateTime, Boolean, Float, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class VulnerabilityResult(Base):
"""
漏洞测试结果数据库模型
存储安全测试用例的执行结果,包括漏洞信息、请求响应详情等。
Attributes:
id (str): 结果唯一标识
execution_id (str): 关联的执行记录ID
case_id (str): 关联的测试用例ID
test_id (str): OWASP测试编号(如 2.1.1)
name (str): 测试用例名称
level (str): 风险等级(critical/high/medium/low/info)
description (str): 漏洞描述
request_info (str): 请求信息(用于复现)
response_info (str): 响应信息(用于复现)
is_vulnerable (bool): 是否存在漏洞
fix_suggestion (str): 修复建议
duration (float): 执行耗时(秒)
status (str): 执行状态(passed/failed/skipped/error)
error_message (str): 错误信息
metadata (dict): 额外元数据(回归测试标记、红线标记等)
created_at (datetime): 创建时间
"""
__tablename__ = "vulnerability_results"
# 风险等级常量
LEVEL_CRITICAL = "critical"
LEVEL_HIGH = "high"
LEVEL_MEDIUM = "medium"
LEVEL_LOW = "low"
LEVEL_INFO = "info"
id: Mapped[str] = mapped_column(String(32), primary_key=True, comment="结果ID")
execution_id: Mapped[str] = mapped_column(
String(32), ForeignKey("executions.id", ondelete="CASCADE"),
nullable=False, comment="关联执行ID"
)
case_id: Mapped[str] = mapped_column(String(32), default="", comment="关联用例ID")
# 漏洞信息
test_id: Mapped[str] = mapped_column(String(20), default="", comment="OWASP测试编号(如2.1.1)")
name: Mapped[str] = mapped_column(String(200), nullable=False, comment="测试用例名称")
level: Mapped[str] = mapped_column(
String(20), default="info",
comment="风险等级: critical/high/medium/low/info"
)
description: Mapped[str] = mapped_column(Text, default="", comment="漏洞描述")
# 请求/响应信息
request_info: Mapped[str] = mapped_column(Text, default="", comment="请求信息(用于复现)")
response_info: Mapped[str] = mapped_column(Text, default="", comment="响应信息(用于复现)")
# 结果
is_vulnerable: Mapped[bool] = mapped_column(Boolean, default=False, comment="是否存在漏洞")
fix_suggestion: Mapped[str] = mapped_column(Text, default="", comment="修复建议")
# 执行信息
duration: Mapped[float] = mapped_column(Float, default=0.0, comment="执行耗时(秒)")
status: Mapped[str] = mapped_column(
String(20), default="pending",
comment="执行状态: pending/running/passed/failed/skipped/error"
)
error_message: Mapped[Optional[str]] = mapped_column(
Text, nullable=True, default=None, comment="错误信息"
)
# 元数据(注意:metadata 是 SQLAlchemy 保留字段名)
# 格式: {"is_regression": true, "vuln_source": "长安深蓝汽车", "is_huawei_redline": false, ...}
extra_data: Mapped[dict] = mapped_column(JSON, default=dict, comment="额外元数据")
created_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="创建时间"
)
# 关联执行记录(多对一)
execution: Mapped["Execution"] = relationship(
"Execution", back_populates="vulnerability_results"
)
def __repr__(self) -> str:
"""字符串表示"""
vuln_mark = "🔴" if self.is_vulnerable else "✅"
return f"<VulnerabilityResult(id={self.id}, test_id={self.test_id}, {vuln_mark} {self.level})>"
def to_dict(self) -> dict:
"""
转换为字典
Returns:
dict: 结果数据字典
"""
return {
"id": self.id,
"execution_id": self.execution_id,
"case_id": self.case_id,
"test_id": self.test_id,
"name": self.name,
"level": self.level,
"description": self.description,
"request_info": self.request_info,
"response_info": self.response_info,
"is_vulnerable": self.is_vulnerable,
"fix_suggestion": self.fix_suggestion,
"duration": self.duration,
"status": self.status,
"error_message": self.error_message,
"extra_data": self.extra_data or {},
"created_at": self.created_at.isoformat() if self.created_at else None,
}
@staticmethod
def level_order(level: str) -> int:
"""
获取风险等级排序值(用于按等级排序)
Args:
level: 风险等级
Returns:
int: 排序值,数值越小风险越高
"""
order_map = {
"critical": 0,
"high": 1,
"medium": 2,
"low": 3,
"info": 4,
}
return order_map.get(level, 99)
@property
def level_emoji(self) -> str:
"""风险等级对应的emoji标记"""
emoji_map = {
"critical": "🔴🔴🔴",
"high": "🔴",
"medium": "🟠",
"low": "🟡",
"info": "🔵",
}
return emoji_map.get(self.level, "⚪")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security.py
模块描述:安全测试 API 路由
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
import logging
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.security import (
SecurityConfigCreate,
SecurityConfigUpdate,
SecurityConfigResponse,
SecurityExecutionCreate,
SecurityExecutionResponse,
VulnerabilityResultResponse,
AccountTestRequest,
AccountTestResponse,
KnowledgeBaseResponse,
HistoricalVuln,
HuaweiRedlineCheck,
)
from app.services.security_service import SecurityService
logger = logging.getLogger(__name__)
router = APIRouter()
# ==================== 配置管理 ====================
@router.get("/config", summary="获取安全测试配置列表")
async def list_configs(
skip: int = Query(0, ge=0, description="跳过记录数"),
limit: int = Query(20, ge=1, le=100, description="返回记录数"),
db: AsyncSession = Depends(get_db)
):
"""
获取安全测试配置列表
Returns:
dict: 配置列表和总数
"""
service = SecurityService(db)
configs, total = await service.list_configs(skip=skip, limit=limit)
return {
"items": [SecurityConfigResponse(**c.to_dict()) for c in configs],
"total": total
}
@router.get("/config/default", summary="获取默认安全测试配置")
async def get_default_config(db: AsyncSession = Depends(get_db)):
"""
获取默认安全测试配置
Returns:
SecurityConfigResponse: 默认配置
"""
service = SecurityService(db)
config = await service.get_default_config()
if not config:
raise HTTPException(status_code=404, detail="未找到默认安全测试配置")
return SecurityConfigResponse(**config.to_dict())
@router.get("/config/{config_id}", summary="获取指定安全测试配置")
async def get_config(
config_id: str,
db: AsyncSession = Depends(get_db)
):
"""
获取指定安全测试配置
Args:
config_id: 配置ID
Returns:
SecurityConfigResponse: 配置详情
"""
service = SecurityService(db)
config = await service.get_config(config_id)
if not config:
raise HTTPException(status_code=404, detail=f"配置不存在: {config_id}")
return SecurityConfigResponse(**config.to_dict())
@router.post("/config", summary="创建安全测试配置")
async def create_config(
config_data: SecurityConfigCreate,
db: AsyncSession = Depends(get_db)
):
"""
创建安全测试配置
Returns:
SecurityConfigResponse: 创建的配置
"""
service = SecurityService(db)
config = await service.create_config(config_data.model_dump(by_alias=True))
return SecurityConfigResponse(**config.to_dict())
@router.put("/config/{config_id}", summary="更新安全测试配置")
async def update_config(
config_id: str,
config_data: SecurityConfigUpdate,
db: AsyncSession = Depends(get_db)
):
"""
更新安全测试配置
Args:
config_id: 配置ID
Returns:
SecurityConfigResponse: 更新后的配置
"""
service = SecurityService(db)
config = await service.update_config(
config_id,
config_data.model_dump(by_alias=True, exclude_none=True)
)
if not config:
raise HTTPException(status_code=404, detail=f"配置不存在: {config_id}")
return SecurityConfigResponse(**config.to_dict())
@router.delete("/config/{config_id}", summary="删除安全测试配置")
async def delete_config(
config_id: str,
db: AsyncSession = Depends(get_db)
):
"""
删除安全测试配置
Args:
config_id: 配置ID
Returns:
dict: 删除结果
"""
service = SecurityService(db)
success = await service.delete_config(config_id)
if not success:
raise HTTPException(status_code=404, detail=f"配置不存在: {config_id}")
return {"message": "删除成功"}
# ==================== 账号测试 ====================
@router.post("/test-account", summary="测试账号连接")
async def test_account(
request: AccountTestRequest,
db: AsyncSession = Depends(get_db)
):
"""
测试指定账号是否可以登录
Returns:
AccountTestResponse: 测试结果
"""
service = SecurityService(db)
result = await service.test_account(request.config_id, request.account_key)
return AccountTestResponse(**result)
# ==================== 执行管理 ====================
@router.post("/executions", summary="创建安全测试执行")
async def create_execution(
execution_data: SecurityExecutionCreate,
db: AsyncSession = Depends(get_db)
):
"""
创建安全测试执行
Returns:
SecurityExecutionResponse: 执行记录
"""
service = SecurityService(db)
execution = await service.create_execution(
config_id=execution_data.config_id,
module_ids=execution_data.module_ids,
case_ids=execution_data.case_ids,
name=execution_data.name,
)
return SecurityExecutionResponse(
id=execution.id,
name=execution.name,
status=execution.status,
total_cases=execution.total_cases,
passed=execution.passed,
failed=execution.failed,
skipped=execution.skipped,
duration=execution.duration,
config_id=execution_data.config_id,
start_time=execution.start_time.isoformat() if execution.start_time else None,
end_time=execution.end_time.isoformat() if execution.end_time else None,
)
@router.post("/executions/{execution_id}/run", summary="执行安全测试")
async def run_execution(
execution_id: str,
db: AsyncSession = Depends(get_db)
):
"""
触发安全测试执行(异步)
Args:
execution_id: 执行ID
Returns:
dict: 执行状态
"""
service = SecurityService(db)
# 异步执行(不等待完成)
import asyncio
asyncio.create_task(service.run_execution(execution_id))
return {
"execution_id": execution_id,
"status": "running",
"message": "安全测试已开始执行"
}
@router.get("/executions/{execution_id}/results", summary="获取安全测试结果")
async def get_execution_results(
execution_id: str,
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
level: Optional[str] = Query(None, description="风险等级筛选"),
is_vulnerable: Optional[bool] = Query(None, description="是否漏洞"),
db: AsyncSession = Depends(get_db)
):
"""
获取安全测试执行结果
Args:
execution_id: 执行ID
Returns:
dict: 结果列表和总数
"""
service = SecurityService(db)
results, total = await service.get_vulnerability_results(
execution_id=execution_id,
skip=skip,
limit=limit,
level=level,
is_vulnerable=is_vulnerable,
)
return {
"items": [VulnerabilityResultResponse(**r.to_dict()) for r in results],
"total": total
}
@router.get("/executions/{execution_id}/summary", summary="获取安全测试摘要")
async def get_execution_summary(
execution_id: str,
db: AsyncSession = Depends(get_db)
):
"""
获取安全测试执行摘要
Args:
execution_id: 执行ID
Returns:
dict: 摘要数据
"""
service = SecurityService(db)
return await service.get_execution_summary(execution_id)
# ==================== 知识库 ====================
@router.get("/knowledge-base", summary="获取安全知识库")
async def get_knowledge_base():
"""
获取安全测试知识库内容
Returns:
KnowledgeBaseResponse: 知识库数据
"""
try:
from app.services.knowledge_base import KNOWLEDGE_BASE
except ImportError:
# 知识库模块尚未创建时返回空数据
return KnowledgeBaseResponse()
# 转换历史漏洞
historical_vulns = []
for v in KNOWLEDGE_BASE.get("historical_vulns", []):
historical_vulns.append(HistoricalVuln(
vuln_id=v.get("vuln_id", ""),
title=v.get("title", ""),
severity=v.get("severity", ""),
vuln_type=v.get("vuln_type", ""),
source_project=v.get("source_project", ""),
detection_method=v.get("detection_method", ""),
regression_note=v.get("regression_note", ""),
mapped_paths=v.get("mapped_paths", []),
))
# 转换华为红线
huawei_redlines = []
for r in KNOWLEDGE_BASE.get("huawei_redlines", []):
huawei_redlines.append(HuaweiRedlineCheck(
check_id=r.get("check_id", ""),
category=r.get("category", ""),
requirement=r.get("requirement", ""),
severity=r.get("severity", ""),
check_method=r.get("check_method", ""),
))
return KnowledgeBaseResponse(
historical_vulns=historical_vulns,
huawei_redlines=huawei_redlines,
total_vulns=len(historical_vulns),
total_redlines=len(huawei_redlines),
)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security.py
模块描述:安全测试 Pydantic 模式定义
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
from datetime import datetime
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field, ConfigDict
def to_camel(string: str) -> str:
"""Convert snake_case to camelCase"""
components = string.split('_')
return components[0] + ''.join(x.title() for x in components[1:])
# ==================== 安全测试配置 ====================
class AccountConfig(BaseModel):
"""测试账号配置"""
username: str = Field(..., description="用户名")
password: str = Field(..., description="密码")
description: str = Field("", description="账号描述")
class AuthConfigDetail(BaseModel):
"""认证配置详情"""
token_type: str = Field("accessToken", description="Token类型")
mechanism: str = Field("JWT", description="认证机制")
login_path: str = Field("/platform/api/auth/login", description="登录路径")
token_header: str = Field("accessToken", description="Token请求头名称")
company_number: str = Field("", description="公司编号")
company_secret: str = Field("", description="公司密钥")
captcha_path: str = Field("/platform/api/code", description="验证码获取路径")
class RateLimitConfig(BaseModel):
"""限流配置"""
brute_force_max: int = Field(20, description="暴力破解最大尝试次数")
rate_limit_max: int = Field(100, description="限流测试最大请求数")
batch_max: int = Field(50, description="批量测试最大请求数")
request_interval: float = Field(0.5, description="请求间隔(秒)")
class SecurityConfigCreate(BaseModel):
"""创建安全测试配置"""
name: str = Field(..., min_length=1, max_length=100, description="配置名称")
target_url: str = Field(..., description="目标服务器地址")
server_ip: str = Field("", description="服务器IP")
verify_ssl: bool = Field(False, description="是否验证SSL证书")
timeout: int = Field(30, ge=5, le=120, description="请求超时时间(秒)")
accounts: Dict[str, Any] = Field(default_factory=dict, description="测试账号配置")
auth_config: Dict[str, Any] = Field(default_factory=dict, description="认证配置")
rate_limits: Dict[str, Any] = Field(default_factory=dict, description="限流配置")
api_prefixes: Dict[str, Any] = Field(default_factory=dict, description="API路径前缀")
is_default: bool = Field(False, description="是否为默认配置")
description: str = Field("", description="配置描述")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class SecurityConfigUpdate(BaseModel):
"""更新安全测试配置"""
name: Optional[str] = Field(None, description="配置名称")
target_url: Optional[str] = Field(None, description="目标服务器地址")
server_ip: Optional[str] = Field(None, description="服务器IP")
verify_ssl: Optional[bool] = Field(None, description="是否验证SSL证书")
timeout: Optional[int] = Field(None, ge=5, le=120, description="请求超时时间(秒)")
accounts: Optional[Dict[str, Any]] = Field(None, description="测试账号配置")
auth_config: Optional[Dict[str, Any]] = Field(None, description="认证配置")
rate_limits: Optional[Dict[str, Any]] = Field(None, description="限流配置")
api_prefixes: Optional[Dict[str, Any]] = Field(None, description="API路径前缀")
is_default: Optional[bool] = Field(None, description="是否为默认配置")
description: Optional[str] = Field(None, description="配置描述")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class SecurityConfigResponse(BaseModel):
"""安全测试配置响应"""
id: str
name: str
target_url: str
server_ip: str = ""
verify_ssl: bool = False
timeout: int = 30
accounts: Dict[str, Any] = {}
auth_config: Dict[str, Any] = {}
rate_limits: Dict[str, Any] = {}
api_prefixes: Dict[str, Any] = {}
is_default: bool = False
description: str = ""
created_at: Optional[str] = None
updated_at: Optional[str] = None
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
# ==================== 安全测试执行 ====================
class SecurityExecutionCreate(BaseModel):
"""创建安全测试执行"""
config_id: str = Field(..., description="安全测试配置ID")
module_ids: List[str] = Field(default_factory=list, description="要执行的模块ID列表(空=全部)")
case_ids: List[str] = Field(default_factory=list, description="要执行的用例ID列表")
include_regression: bool = Field(True, description="是否包含历史漏洞回归测试")
include_redline: bool = Field(True, description="是否包含华为安全红线检查")
name: str = Field("", description="执行名称")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class SecurityExecutionResponse(BaseModel):
"""安全测试执行响应"""
id: str
name: str = ""
status: str = "pending"
total_cases: int = 0
passed: int = 0
failed: int = 0
skipped: int = 0
duration: float = 0.0
config_id: str = ""
start_time: Optional[str] = None
end_time: Optional[str] = None
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
# ==================== 漏洞结果 ====================
class VulnerabilityResultResponse(BaseModel):
"""漏洞测试结果响应"""
id: str
execution_id: str
case_id: str = ""
test_id: str = ""
name: str
level: str = "info"
description: str = ""
request_info: str = ""
response_info: str = ""
is_vulnerable: bool = False
fix_suggestion: str = ""
duration: float = 0.0
status: str = "pending"
error_message: Optional[str] = None
extra_data: Dict[str, Any] = {}
created_at: Optional[str] = None
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
# ==================== 知识库 ====================
class HistoricalVuln(BaseModel):
"""历史漏洞条目"""
vuln_id: str = Field(..., description="漏洞ID")
title: str = Field(..., description="漏洞标题")
severity: str = Field(..., description="严重等级")
vuln_type: str = Field(..., description="漏洞类型")
source_project: str = Field("", description="来源项目")
detection_method: str = Field("", description="检测方法")
regression_note: str = Field("", description="回归验证说明")
mapped_paths: List[str] = Field(default_factory=list, description="映射路径")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class HuaweiRedlineCheck(BaseModel):
"""华为安全红线检查项"""
check_id: str = Field(..., description="检查编号")
category: str = Field(..., description="检查类别")
requirement: str = Field(..., description="要求描述")
severity: str = Field(..., description="严重等级")
check_method: str = Field("", description="检查方法")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class KnowledgeBaseResponse(BaseModel):
"""知识库响应"""
historical_vulns: List[HistoricalVuln] = Field(default_factory=list, description="历史漏洞列表")
huawei_redlines: List[HuaweiRedlineCheck] = Field(default_factory=list, description="华为红线列表")
total_vulns: int = Field(0, description="历史漏洞总数")
total_redlines: int = Field(0, description="红线检查总数")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
# ==================== 账号测试 ====================
class AccountTestRequest(BaseModel):
"""账号连接测试请求"""
config_id: str = Field(..., description="安全测试配置ID")
account_key: str = Field(..., description="账号键名(superadmin/admin/user)")
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
class AccountTestResponse(BaseModel):
"""账号连接测试响应"""
account_key: str
success: bool
token_preview: str = ""
error_message: str = ""
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
......@@ -10,7 +10,7 @@
"""
from datetime import datetime
from typing import Optional, List, Dict, Any
from typing import Optional, List, Dict, Any, Union
from pydantic import BaseModel, Field, ConfigDict
......@@ -62,7 +62,10 @@ class TestCaseBase(BaseModel):
description: str = Field(default="", max_length=1000, description="用例描述")
priority: str = Field(default="medium", description="优先级: high/medium/low")
tags: List[str] = Field(default_factory=list, description="标签列表")
steps: List[StepDefinition] = Field(default_factory=list, description="测试步骤")
steps: Union[List[StepDefinition], Dict[str, Any]] = Field(
default_factory=list,
description="测试步骤(UI用例为步骤列表,安全测试用例为配置字典)"
)
config: Dict[str, Any] = Field(
default_factory=lambda: {"timeout": 30000, "retry": 0, "screenshot": True},
description="执行配置"
......@@ -106,7 +109,9 @@ class TestCaseUpdate(BaseModel):
status: Optional[str] = Field(None, description="状态: active/disabled")
priority: Optional[str] = Field(None, description="优先级")
tags: Optional[List[str]] = Field(None, description="标签列表")
steps: Optional[List[StepDefinition]] = Field(None, description="测试步骤")
steps: Optional[Union[List[StepDefinition], Dict[str, Any]]] = Field(
None, description="测试步骤(UI用例为步骤列表,安全测试用例为配置字典)"
)
config: Optional[Dict[str, Any]] = Field(None, description="执行配置")
case_type: Optional[str] = Field(None, description="用例类型: ui/api/security/deploy")
......
......@@ -182,11 +182,22 @@ class CaseService:
if len(data.name) > 200:
raise ValueError("用例名称不能超过200字符")
# 步骤校验
# 步骤校验(仅UI用例需要列表格式校验)
if data.steps and data.case_type == "ui":
if isinstance(data.steps, list):
for i, step in enumerate(data.steps):
if not step.action:
raise ValueError(f"第{i+1}步的动作类型不能为空")
# 处理 steps 字段:UI用例转列表,安全测试用例保持字典
steps_data = []
if data.steps:
for i, step in enumerate(data.steps):
if not step.action:
raise ValueError(f"第{i+1}步的动作类型不能为空")
if isinstance(data.steps, list):
# UI用例:StepDefinition列表转字典列表
steps_data = [s.model_dump() for s in data.steps]
elif isinstance(data.steps, dict):
# 安全测试用例:直接使用字典
steps_data = data.steps
# 创建用例对象
test_case = TestCase(
......@@ -196,7 +207,7 @@ class CaseService:
description=data.description or "",
priority=data.priority or "medium",
tags=data.tags or [],
steps=[s.model_dump() for s in data.steps] if data.steps else [],
steps=steps_data,
config=data.config or {"timeout": 30000, "retry": 0, "screenshot": True},
case_type=data.case_type or "ui",
)
......
......@@ -210,6 +210,10 @@ class ExecutionService:
if not execution:
raise ValueError(f"执行记录不存在: {execution_id}")
# ===== 按用例类型分流:安全测试走专用执行器 =====
if execution.case_type == "security":
return await self._run_security_execution(execution_id, config)
# 更新状态为运行中
execution.status = "running"
execution.start_time = datetime.now()
......@@ -588,6 +592,188 @@ class ExecutionService:
return callback
async def _run_security_execution(
self,
execution_id: str,
config: Optional[dict] = None,
) -> Execution:
"""
安全测试专用执行流程
安全测试用例不使用 Playwright(无浏览器),而是用 requests + 签名算法
直接调用被测系统 API。本方法:
1. 从执行记录中取出关联的用例
2. 加载默认安全测试配置
3. 用 SecurityExecutor 执行
4. 把结果写回 CaseResult + VulnerabilityResult
Args:
execution_id: 执行记录ID
config: 执行配置(未使用,保留兼容)
Returns:
Execution: 更新后的执行记录
"""
# 延迟导入,避免循环依赖
from app.executors.security_executor import SecurityExecutor, SecurityCaseResult
from app.models.security_config import SecurityConfig
from app.models.vulnerability_result import VulnerabilityResult
execution = await self.get_execution(execution_id)
if not execution:
raise ValueError(f"执行记录不存在: {execution_id}")
# 广播执行开始
await manager.broadcast(execution_id, {
"type": "execution_start",
"data": execution.to_dict(),
})
# 获取所有待执行的用例结果(按模块分组排序)
results_query = (
select(CaseResult)
.join(TestCase, CaseResult.case_id == TestCase.id)
.where(
CaseResult.execution_id == execution_id,
CaseResult.status == "pending",
)
.order_by(TestCase.module_id, TestCase.order, TestCase.created_at)
)
results_data = await self.db.execute(results_query)
case_results = list(results_data.scalars().all())
# 获取用例定义
case_ids = [r.case_id for r in case_results]
cases_query = select(TestCase).where(TestCase.id.in_(case_ids))
cases_data = await self.db.execute(cases_query)
cases = list(cases_data.scalars().all())
cases_map = {c.id: c for c in cases}
# 加载默认安全测试配置
cfg_query = select(SecurityConfig).where(SecurityConfig.is_default == True)
cfg_result = await self.db.execute(cfg_query)
sec_config = cfg_result.scalar_one_or_none()
if not sec_config:
# 没有配置,全部标记失败
for cr in case_results:
cr.status = "error"
cr.error_message = "未找到默认安全测试配置"
execution.status = "failed"
execution.end_time = datetime.now()
await self.db.flush()
return execution
client_config = sec_config.to_client_config()
# 在 async 上下文中预先把 ORM 对象转成纯字典(避免线程池脱钩)
def _to_plain_steps(steps_val):
if isinstance(steps_val, str):
try:
parsed = json.loads(steps_val)
return parsed if isinstance(parsed, dict) else {}
except Exception:
return {}
return steps_val if isinstance(steps_val, dict) else {}
plain_cases = [
{"id": c.id, "name": c.name, "steps": _to_plain_steps(c.steps)}
for c in cases
]
# 定义线程安全的同步执行函数
loop = asyncio.get_event_loop()
def sync_run_security():
executor = SecurityExecutor(client_config)
executor.start()
results = []
for i, case_dict in enumerate(plain_cases):
case_id = case_dict.get("id", "")
case_name = case_dict.get("name", "")
logger.info(f"⏳ 安全测试用例 {i+1}/{len(plain_cases)}: {case_name}")
try:
result = executor.execute_case(case_dict)
except Exception as e:
logger.error(f"安全用例执行异常: {case_name}, {e}")
result = SecurityCaseResult(
case_id=case_id, name=case_name,
status="error", error=str(e),
)
results.append(result)
return results
try:
results: List[SecurityCaseResult] = await loop.run_in_executor(
None, sync_run_security
)
except Exception as e:
logger.error(f"安全测试执行异常: {e}")
execution.status = "failed"
execution.end_time = datetime.now()
await self.db.flush()
return execution
# 更新用例结果 + 保存漏洞结果
passed = 0
failed = 0
for r in results:
# 找到对应的 CaseResult 更新状态
cr = next((c for c in case_results if c.case_id == r.case_id), None)
if cr:
if r.status == "error":
cr.status = "error"
cr.error_message = r.error
elif r.is_vulnerable:
cr.status = "failed"
cr.error_message = r.description[:500] if r.description else None
failed += 1
else:
cr.status = "passed"
passed += 1
cr.duration = r.duration
cr.end_time = datetime.now()
# 保存漏洞测试结果
vuln_result = VulnerabilityResult(
id=generate_id("vuln"),
execution_id=execution_id,
case_id=r.case_id,
test_id=r.test_id,
name=r.name,
level=r.level,
description=r.description,
request_info=r.request_info,
response_info=r.response_info,
is_vulnerable=r.is_vulnerable,
fix_suggestion=r.fix_suggestion,
duration=r.duration,
status=r.status,
error_message=r.error,
extra_data=r.metadata,
)
self.db.add(vuln_result)
# 更新执行统计
execution.passed = passed
execution.failed = failed
execution.skipped = len(case_results) - passed - failed
execution.total_cases = len(case_results)
execution.status = "completed"
execution.end_time = datetime.now()
execution.duration = (execution.end_time - execution.start_time).total_seconds() if execution.start_time else 0
execution.pass_rate = round((passed / execution.total_cases) * 100, 2) if execution.total_cases > 0 else 0
await self.db.flush()
# 广播执行完成
await manager.broadcast(execution_id, {
"type": "execution_complete",
"data": execution.to_dict(),
})
logger.info(f"安全测试执行完成: {execution_id}, 通过: {passed}, 失败: {failed}")
return execution
async def _count_results(self, execution_id: str, status: str) -> int:
"""
统计指定状态的用例结果数量
......
此差异已折叠。
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
创建默认安全测试配置(直接操作数据库,避免 HTTP 编码问题)
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models.security_config import SecurityConfig
def main():
db_path = os.path.join(os.path.dirname(__file__), "..", "data", "test_platform.db")
db_url = f"sqlite:///{os.path.abspath(db_path)}"
engine = create_engine(db_url)
Session = sessionmaker(bind=engine)
session = Session()
try:
# 检查是否已存在默认配置
existing = session.query(SecurityConfig).filter(SecurityConfig.is_default == True).first()
if existing:
print(f"默认配置已存在: {existing.name} ({existing.id})")
return
config = SecurityConfig(
id="sec_cfg_default",
name="新统一平台安全测试配置",
target_url="https://192.168.5.44",
server_ip="192.168.5.44",
verify_ssl=False,
timeout=30,
accounts={
"superadmin": {"username": "superadmin", "password": "Ubains@1357"},
"admin": {"username": "admin@aq", "password": "Ubains@1357"},
"user": {"username": "user@aq", "password": "Ubains@1357"},
"captcha": "csba",
},
auth_config={
"token_type": "accessToken",
"mechanism": "JWT",
"login_path": "/platform/api/auth/login",
"token_header": "accessToken",
"captcha_path": "/platform/api/code",
},
rate_limits={
"brute_force_max": 20,
"rate_limit_max": 100,
"request_interval": 0.5,
},
api_prefixes={
"meeting": "/api/",
"monitor": "/monitor/api2/api/",
"platform": "/platform/api/",
},
is_default=True,
description="默认安全测试配置,用于新统一平台(192.168.5.44)",
)
session.add(config)
session.commit()
print(f"✅ 默认配置创建成功: {config.id}")
print(f" 名称: {config.name}")
print(f" 目标: {config.target_url}")
except Exception as e:
session.rollback()
print(f"❌ 错误: {e}")
finally:
session.close()
if __name__ == "__main__":
main()
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论