提交 68b2fbbf authored 作者: 陈泽健's avatar 陈泽健

feat(security): 安全测试报告生成服务+格式对齐参考实现

新增 security_report_service.py,生成7章节Markdown报告:
- 执行概要 + OWASP覆盖矩阵(5列,critical合并到high)
- 漏洞详情(按风险等级排序,含请求/响应/修复建议)
- 已验证安全项(按OWASP模块分组)
- 风险评估与修复建议
- 已知安全问题清单
- 历史漏洞回归验证(五.五章节,6列表格)
- 华为安全红线合规性检查(五.六章节)
- 测试环境与工具(含服务器架构信息)
- 参考资料(Z盘路径)

执行完成自动生成报告,路径存入execution.config.report_path
新增报告查看/下载API + 前端API封装
格式对齐参考实现report_generator.py
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 a9074f8c
# 【安全测试】报告格式对齐 - 计划执行文档
> **文档类型**: 计划执行文档
> **创建日期**: 2026-07-21
> **关联需求**: `_PRD_安全测试报告格式对齐_需求文档.md`
> **预计工作量**: 小(1 个文件核心修改,2 个文件微调)
---
## 一、执行概要
`security_report_service.py` 的报告输出格式对齐到参考实现 `report_generator.py`,涉及章节编号调整、OWASP 矩阵列数对齐、历史漏洞表格补列、服务器信息补架构、新增参考资料章节。
---
## 二、修改范围
| 文件 | 修改类型 | 说明 |
|------|---------|------|
| `backend/app/services/security_report_service.py` | **核心修改** | 报告格式对齐全部改动在此文件 |
| `backend/app/models/security_config.py` | 微调 | 可选:新增 `server_arch` 字段 |
---
## 三、分阶段执行计划
### Phase 1: OWASP 覆盖矩阵对齐
**目标**: 将 6 列(含严重)改为 5 列,critical 合并到 high
**修改点**`security_report_service.py`:
1. **`_get_summary()`**: 将 `critical` 计入 `high`
```python
# 修改前
summary = {"critical": 0, "high": 0, ...}
for r in results:
if r.is_vulnerable:
if r.level in summary:
summary[r.level] += 1
# 修改后
summary = {"high": 0, "medium": 0, "low": 0, "info": 0, "safe": 0, "vuln_total": 0, "total": 0}
for r in results:
if r.is_vulnerable:
mapped_level = "high" if r.level == "critical" else r.level
if mapped_level in summary:
summary[mapped_level] += 1
summary["vuln_total"] += 1
else:
summary["safe"] += 1
summary["total"] = len(results)
```
2. **`_get_module_stats()`**: 同样将 critical 合并到 high
3. **覆盖矩阵表头**: 去掉"🔴严重"
```markdown
| 编号 | 安全风险 | 用例数 | 🔴高危 | 🟠中危 | 🟡低危 | 🔵信息 | 🟢安全 |
```
4. **统计总览表**: 去掉"🔴🔴🔴 严重"
5. **`LEVEL_MAP`**: 移除 critical 或标记为合并到 high
```python
LEVEL_MAP = {
"critical": ("严重(高危)", "🔴"), # 合并显示
"high": ("高危", "🔴"),
"medium": ("中危", "🟠"),
"low": ("低危", "🟡"),
"info": ("信息", "🔵"),
}
```
---
### Phase 2: 章节编号对齐
**目标**: 历史漏洞/红线/测试环境/参考资料编号与参考实现一致
**修改点** `_generate_markdown()` 方法:
| 原编号 | 新编号 | 章节 |
|--------|--------|------|
| `## 六、历史漏洞回归验证` | `## 五.五、历史漏洞回归验证` | 历史漏洞 |
| `## 七、华为安全红线合规性检查` | `## 五.六、华为安全红线合规性检查` | 华为红线 |
| `## 八、测试环境与工具` | `## 六、测试环境与工具` | 测试环境 |
| | `## 七、参考资料` | **新增** |
---
### Phase 3: 历史漏洞表格补"详情"列
**目标**: 与参考实现一致,6 列表格
**修改点** `_generate_markdown()` 中历史漏洞回归部分:
```markdown
| 编号 | 漏洞来源 | 测试项 | 当前状态 | 风险等级 | 详情 |
|------|---------|--------|---------|---------|------|
```
对每条结果:
```python
desc_short = r.description[:80] + "..." if len(r.description) > 80 else r.description
level_name, _ = LEVEL_MAP.get(r.level, (r.level, ""))
lines.append(f"| {r.test_id} | {source} | {r.name} | {status} | {level_name} | {desc_short} |")
```
---
### Phase 4: 服务器信息补架构 + 新增参考资料
**目标**: 服务器显示 `192.168.5.44 (X86架构-欧拉)`,新增参考资料章节
**修改点 1** — 新增常量:
```python
# 默认服务器架构信息
DEFAULT_SERVER_ARCH = "X86架构-欧拉"
# 参考资料(内网盘路径)
REFERENCE_MATERIALS = [
"华为安全红线资料:[Z:\\deploy\\18其它系统\\安全测试\\01安全测试资料]",
"以往漏洞报告资料:[Z:\\deploy\\18其它系统\\安全测试\\02项目漏洞资料]",
"Nginx配置文件:[Z:\\deploy\\18其它系统\\安全测试\\03Nginx配置文件]",
]
```
**修改点 2** — `_generate_markdown()` 测试环境部分:
```python
lines.append(f"| 目标服务器 | {server_ip} ({DEFAULT_SERVER_ARCH}) |")
```
**修改点 3** — `_generate_markdown()` 末尾新增:
```python
# ========== 七、参考资料 ==========
lines.append("## 七、参考资料\n")
for ref in REFERENCE_MATERIALS:
lines.append(f"- {ref}")
lines.append("")
```
**可选扩展** — `SecurityConfig` 新增 `server_arch` 字段:
- 若后续需要配置化架构信息,在 `SecurityConfig` 加 `server_arch: str` 字段
- 报告生成时优先用配置值,无则用默认值
- **本阶段暂不实现**,仅用常量,避免数据库迁移
---
### Phase 5: 验证
**验证步骤**:
```bash
cd backend
PYTHONIOENCODING=utf-8 python -c "
import asyncio, sys, os
sys.path.insert(0, '.')
async def main():
from app.database import async_session_maker
from app.services.security_report_service import SecurityReportService
async with async_session_maker() as db:
service = SecurityReportService()
markdown = await service.get_report_content('exec_686280e0d43b40cda9508e7d023044a9', db)
print(markdown)
asyncio.run(main())
" 2>&1
```
**验收检查点**
1. ✅ OWASP 矩阵表头为 `编号|安全风险|用例数|🔴高危|🟠中危|🟡低危|🔵信息|🟢安全`
2. ✅ 统计总览无"严重"行
3. ✅ 历史漏洞编号为 `五.五`
4. ✅ 华为红线编号为 `五.六`
5. ✅ 测试环境编号为 `六`
6. ✅ 参考资料编号为 `七`
7. ✅ 历史漏洞表格有 6 列(含"详情")
8. ✅ 服务器信息含 `(X86架构-欧拉)`
9. ✅ API 启动无报错
---
## 四、风险评估
| 风险 | 概率 | 影响 | 缓解 |
|------|------|------|------|
| critical 用例的等级显示变化 | 低 | 低 | 漏洞详情标题仍显示"严重",仅矩阵统计合并 |
| 参考资料路径硬编码 | 低 | 低 | 用常量,后续可配置化 |
| 数据库无新字段 | 无 | 无 | 本阶段不新增 SecurityConfig 字段 |
---
## 五、涉及文件汇总
| 文件 | 操作 |
|------|------|
| `backend/app/services/security_report_service.py` | 修改(核心) |
| `Docs/PRD/需求文档/_PRD_安全测试报告格式对齐_需求文档.md` | 新增 |
| `Docs/PRD/需求文档/_PRD_安全测试报告格式对齐_计划执行.md` | 新增 |
---
*本文档为安全测试报告格式对齐的计划执行文档。*
\ No newline at end of file
# 【安全测试】报告格式对齐 - 需求文档
> **文档类型**: PRD 需求文档
> **创建日期**: 2026-07-21
> **参考实现**: `临时目录/安全测试/ApiSecurityTest/utils/report_generator.py`
> **优先级**: P1
---
## 一、需求背景
当前安全测试报告生成服务 `security_report_service.py` 已实现 7 章节 Markdown 报告,但与参考实现的报告格式存在差异。参考实现的报告经过实战交付验证,格式规范、信息完整,需要将当前实现对齐到参考格式。
---
## 二、当前差异清单
### 2.1 章节结构差异
| 章节 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 标题/基本信息 | 一份总报告 | 一份总报告 | ✅ 一致 |
| 一、执行概要 | 有 | 有 | ✅ 一致 |
| 二、漏洞详情 | 有 | 有 | ✅ 一致 |
| 三、已验证安全项 | 有 | 有 | ✅ 一致 |
| 四、风险评估与修复建议 | 有 | 有 | ✅ 一致 |
| 五、已知安全问题清单 | 有 | 有 | ✅ 一致 |
| 五.五、历史漏洞回归验证 | 有(编号五.五) | 有(编号六) | ⚠️ 编号不同 |
| 五.六、华为安全红线合规性检查 | 有(编号五.六) | 有(编号七) | ⚠️ 编号不同 |
| 六、测试环境与工具 | 有 | 有(编号八) | ⚠️ 编号不同 |
| 七、参考资料 | 有 | 无 | ❌ 缺失 |
### 2.2 OWASP 覆盖矩阵差异
| 维度 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 风险等级列 | 高危 / 中危 / 低危 / 信息 / 安全(5列) | 严重 / 高危 / 中危 / 低危 / 信息 / 安全(6列) | ⚠️ 多了"严重"列 |
| 等级体系 | 中文(高危/中危/低危/信息类) | 英文+映射(critical/high/medium/low/info) | ⚠️ 平台有 critical,需处理 |
### 2.3 漏洞详情差异
| 维度 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 标题格式 | `### 2.{idx} [{r.level}] {r.name}` | `### 2.{idx} [{level_name}] {r.name}` | ✅ 一致 |
| 等级显示 | 中文(高危) | 中文(高危) | ✅ 一致 |
### 2.4 历史漏洞回归验证差异
| 维度 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 章节编号 | `五.五` | `六` | ⚠️ 需对齐 |
| 识别方式 | `metadata.is_regression` | `test_id.startswith("HV-")` | ✅ 当前更简洁 |
| 表格列 | 编号/漏洞来源/测试项/当前状态/风险等级/详情(6列) | 编号/漏洞来源/测试项/当前状态/风险等级(5列) | ⚠️ 缺"详情"列 |
| 详情显示 | `description[:80] + "..."` | 无 | ❌ 缺失 |
### 2.5 华为安全红线合规性检查差异
| 维度 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 章节编号 | `五.六` | `七` | ⚠️ 需对齐 |
| 识别方式 | `metadata.is_huawei_redline` + `metadata.hw_*` | `test_id.startswith("HW-")` + `knowledge_base` | ✅ 当前更合理 |
| 表格列 | 检查编号/检查类别/要求/测试结果/合规状态(5列) | 同上 | ✅ 一致 |
| 要求字段截断 | `[:57] + "..."` | `[:57] + "..."` | ✅ 一致 |
### 2.6 测试环境与工具差异
| 维度 | 参考实现 | 当前实现 | 差异 |
|------|---------|---------|------|
| 服务器信息 | `192.168.5.44 (X86架构-欧拉)` | `192.168.5.44` | ⚠️ 缺架构信息 |
| 章节编号 | 六 | 八 | ⚠️ 需对齐 |
### 2.7 缺失章节
| 章节 | 参考实现内容 | 当前状态 |
|------|------------|---------|
| 七、参考资料 | Z 盘路径(华为红线/漏洞报告/Nginx配置) | ❌ 完全缺失 |
---
## 三、对齐需求
### 3.1 章节编号对齐
| 原编号 | 新编号 |
|--------|--------|
| 六、历史漏洞回归验证 | 五.五、历史漏洞回归验证 |
| 七、华为安全红线合规性检查 | 五.六、华为安全红线合规性检查 |
| 八、测试环境与工具 | 六、测试环境与工具 |
**新增**
- 七、参考资料
### 3.2 OWASP 覆盖矩阵对齐
将平台的 6 级风险等级(critical/high/medium/low/info + 安全)映射到参考实现的 5 列格式:
| 平台 level | 报告列 |
|-----------|--------|
| critical | 合并到 🔴高危 |
| high | 🔴高危 |
| medium | 🟠中危 |
| low | 🟡低危 |
| info | 🔵信息 |
| 安全 | 🟢安全 |
**实现方式**:在 `get_summary()``get_module_stats()` 中,将 critical 计入 high 列。
### 3.3 历史漏洞表格对齐
新增"详情"列,显示 `description` 前 80 字符:
```markdown
| 编号 | 漏洞来源 | 测试项 | 当前状态 | 风险等级 | 详情 |
|------|---------|--------|---------|---------|------|
| HV-001 | 长安深蓝汽车 | HV-001 SQL注入-会议预定接口 | ✅ 已修复 | 高危 | 会议预定接口存在SQL注入漏洞... |
```
### 3.4 服务器信息对齐
`_extract_server_info()` 中添加架构后缀:
```
192.168.5.44 (X86架构-欧拉)
```
**方案**:从 `SecurityConfig` 或平台配置读取架构信息,若无则使用默认值。
### 3.5 新增参考资料章节
```markdown
## 七、参考资料
- 华为安全红线资料:[Z:\deploy\18其它系统\安全测试\01安全测试资料]
- 以往漏洞报告资料:[Z:\deploy\18其它系统\安全测试\02项目漏洞资料]
- Nginx配置文件:[Z:\deploy\18其它系统\安全测试\03Nginx配置文件]
```
**方案**:作为常量配置,可从 `SecurityConfig` 覆盖。
---
## 四、非功能性需求
| 需求 | 说明 |
|------|------|
| 兼容性 | 保持现有 API 不变,仅修改报告内容格式 |
| 可配置 | 服务器架构、参考资料路径可配置 |
| 回归测试 | 现有 7 章节报告测试用例需继续通过 |
---
## 五、验收标准
1. ✅ 生成的报告与参考实现格式完全一致(章节编号、列数、内容)
2. ✅ OWASP 覆盖矩阵为 5 列(无"严重"列)
3. ✅ 历史漏洞表格有"详情"列
4. ✅ 服务器信息含架构后缀
5. ✅ 有"七、参考资料"章节
6. ✅ 现有测试通过
---
*本文档为安全测试报告格式对齐需求文档。*
\ No newline at end of file
# HANDOFF — 安全测试模块会话交接文档
> **生成时间**: 2026-07-21
> **当前分支**: `platform-auto-test`
> **开发窗口**: 安全测试模块(与会议管理窗口并行)
> **最近提交**: `4b690a98`(会议管理窗口,安全测试尚未提交)
> **状态**: ✅ 安全测试引擎开发完成并自测通过,待提交
---
## ⚠️ 多窗口并行开发注意(安全测试窗口)
本窗口是**安全测试模块开发**,与另一个窗口(会议管理/UI自动化)并行开发。**必须遵守** `Docs/多窗口并行开发指南.md`
| 项目 | 安全测试窗口(本窗口) | 会议管理窗口 |
|------|----------------------|------------|
| 后端端口 | 8002(或不启动) | 8001 |
| 前端端口 | 3001(或不启动) | 3000 |
| 开发模块 | 安全测试(`module_type=security`) | 会议管理 |
| Playwright | **不执行 UI 用例** | 可执行 |
| Git 提交 | 提交前先 `git pull` | 提交后 push |
| 数据库 | 错开写入时间 | 错开写入时间 |
---
## 一、本次任务(已完成)
### 任务:为平台增加 API 接口安全测试能力
**核心成果**
1. ✅ 后端安全测试引擎(SecurityExecutor + HttpClient 签名 + AuthHelper 多账号)
2. ✅ 12 个安全测试模块 + 24 个安全测试用例
3. ✅ 前端适配(Cases.vue / Execution.vue 支持 security 类型)
4. ✅ 执行调度分流修复(security 用例不再误走 PlaywrightExecutor)
5. ✅ 端到端自测通过(3/3 用例正常执行)
---
## 二、已完成事项清单
### 2.1 后端新增文件(9个)
| 文件 | 说明 |
|------|------|
| `backend/app/models/security_config.py` | 安全测试配置 ORM(含 to_client_config) |
| `backend/app/models/vulnerability_result.py` | 漏洞测试结果 ORM(⚠️ 字段名用 `extra_data`,非 metadata) |
| `backend/app/schemas/security.py` | Pydantic 校验层 |
| `backend/app/executors/http_client.py` | HTTP 客户端(AES-CBC 签名算法) |
| `backend/app/executors/auth_helper.py` | 多账号登录辅助 |
| `backend/app/executors/security_executor.py` | 安全测试执行引擎核心 |
| `backend/app/services/security_service.py` | 安全测试业务逻辑层 |
| `backend/app/services/knowledge_base.py` | 知识库(14历史漏洞 + 15华为红线 + SQL载荷) |
| `backend/app/routers/security.py` | API 路由(11个接口) |
### 2.2 后端修改文件
| 文件 | 修改内容 |
|------|---------|
| `backend/app/main.py` | 注册 `/api/security` 路由 |
| `backend/app/models/__init__.py` | 导入 SecurityConfig、VulnerabilityResult |
| `backend/app/services/execution_service.py` | **核心**`run_execution` 按 case_type 分流;新增 `_run_security_execution` |
| `backend/app/schemas/test_case.py` | `steps` 字段改为 Union(支持 list/dict) |
| `backend/app/services/case_service.py` | 创建用例时区分 UI(list)/security(dict) |
### 2.3 前端新增/修改文件
| 文件 | 说明 |
|------|------|
| `frontend/src/types/security.ts` | TypeScript 类型定义 |
| `frontend/src/api/security.ts` | 安全测试 API 封装 |
| `frontend/src/views/Cases.vue` | 安全用例显示风险等级 emoji/标签、详情弹窗 |
| `frontend/src/views/Execution.vue` | 执行弹窗(配置选择)、分流安全/UI执行逻辑 |
### 2.4 脚本文件
| 文件 | 说明 |
|------|------|
| `backend/scripts/create_security_cases.py` | 创建 12 模块 + 24 用例 |
| `backend/scripts/create_default_security_config.py` | 创建默认安全测试配置 |
| `backend/scripts/test_security_e2e.py` | 端到端执行测试 |
| `backend/scripts/debug_security_exec.py` | 调试脚本(同步 engine) |
| `backend/scripts/debug_security_exec2.py` | 调试脚本(run_in_executor) |
| `backend/scripts/debug_security_async.py` | 调试脚本(async engine) |
### 2.5 文档
| 文件 | 说明 |
|------|------|
| `Docs/PRD/需求文档/_PRD_安全测试模块需求文档.md` | 需求文档 |
| `Docs/PRD/需求文档/_PRD_安全测试模块_计划执行.md` | 计划执行文档 |
| `Docs/PRD/问题处理/安全测试/_问题处理_安全测试执行报错str对象无get属性.md` | 问题处理(已修正根因) |
| `Docs/PRD/问题处理/安全测试/_执行计划_修复安全测试执行报错.md` | 执行计划 |
---
## 三、当前数据状态(已入库)
| 内容 | 数量 | 备注 |
|------|------|------|
| 安全测试模块 | 12 | sec_api01~sec_api10 + sec_regression + sec_redline |
| 安全测试用例 | 24 | 覆盖 API1/2/5/8 + 回归 + 红线(API3/4/6/7/9/10 待补) |
| 安全测试配置 | 1 | `sec_cfg_default`(新统一平台 192.168.5.44) |
---
## 四、关键技术决策
### 4.1 安全测试不依赖浏览器
`SecurityExecutor``requests` 库 + 签名算法直接调 API,与 `PlaywrightExecutor` 完全解耦。
### 4.2 执行调度分流(核心修复)
`execution_service.run_execution()` 入口按 `case_type` 分流:
```python
if execution.case_type == "security":
return await self._run_security_execution(execution_id, config)
```
安全用例走专用方法,结果同时写入 `CaseResult`(执行中心展示)+ `VulnerabilityResult`(报告中心)。
### 4.3 签名算法(从前端 JS 逆向)
`http_client.py` 实现了 AES-CBC 签名(X-RANDOM/X-TIMESTAMP/X-SIGN),与被测系统前端一致,需 `pycryptodome`
### 4.4 steps 字段双格式
- UI 用例:`steps` 是步骤列表 `[{order, action, ...}]`
- 安全用例:`steps` 是配置字典 `{test_type, target, auth, assertions, vulnerability}`
---
## 五、下一步计划(按优先级)
| 优先级 | 待办 | 说明 |
|--------|------|------|
| **P0** | 提交安全测试代码 | 提交前先 `git pull` 会议管理窗口的变更 |
| **P1** | 报告中心报告细化 | 参考实现设计了 7 大章节的详细报告,当前平台只存原始数据,需生成 Markdown 详细报告 |
| **P1** | 补充 API3/4/6/7/9/10 用例 | 当前只覆盖 24 个,参考实现有 90+ |
| **P2** | WebSocket 实时进度推送 | 安全测试目前不推送步骤进度 |
| **P3** | 网盘上传 + ERP 任务对接 | 参考实现的功能,平台化后可选 |
---
## 六、踩坑记录(安全测试专项,绝对不要重复踩)
| # | 现象 | 根因 | 正确做法 |
|---|------|------|---------|
| 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 脱钩" | 未先看后端日志的错误来源模块 | **排查异常先 grep 日志确认是哪个 executor 报的** |
| S5 | Windows curl 发中文 JSON body 失败 | Windows bash 编码 | 用 Python requests 脚本测试,不用 curl |
---
## 七、启动与验证
```bash
# 后端(端口 8001,会议管理窗口在跑)
cd backend
uvicorn app.main:app --port 8001
# 验证安全测试
PYTHONIOENCODING=utf-8 python scripts/test_security_e2e.py
```
- 前端: http://localhost:3000/cases/security
- API 文档: http://localhost:8001/docs(搜 security 标签)
- 被测系统: https://192.168.5.44
---
## 八、未提交变更(安全测试窗口)
以下文件未提交,待 git pull 后统一提交:
**后端新增**(9个):
- `backend/app/models/security_config.py`
- `backend/app/models/vulnerability_result.py`
- `backend/app/schemas/security.py`
- `backend/app/executors/http_client.py`
- `backend/app/executors/auth_helper.py`
- `backend/app/executors/security_executor.py`
- `backend/app/services/security_service.py`
- `backend/app/services/knowledge_base.py`
- `backend/app/routers/security.py`
**后端修改**
- `backend/app/main.py``models/__init__.py``execution_service.py``schemas/test_case.py``case_service.py`
**前端**
- `frontend/src/types/security.ts``api/security.ts``views/Cases.vue``views/Execution.vue`
**脚本/文档**
- `backend/scripts/create_security_cases.py`
- `Docs/PRD/需求文档/_PRD_安全测试*.md`
- `Docs/PRD/问题处理/安全测试/*.md`
---
*本文档由安全测试开发窗口生成,供下一次会话快速恢复上下文。会议管理窗口的交接见 `HANDOFF.md`。*
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_report_service.py
模块描述:安全测试报告生成服务,生成 7 章节 Markdown 格式报告
作者:czj
创建日期:2026-07-21
最后修改:2026-07-21
"""
import os
import logging
from datetime import datetime
from typing import List, Dict, Any, Optional
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, desc
from app.models.vulnerability_result import VulnerabilityResult
from app.models.execution import Execution
from app.models.security_config import SecurityConfig
from app.config import settings
from app.services.knowledge_base import KNOWLEDGE_BASE
logger = logging.getLogger(__name__)
# OWASP API Security Top 10 模块定义(用于覆盖矩阵)
OWASP_MODULES = [
("2.1", "API1", "对象级别授权失效"),
("2.2", "API2", "身份认证失效"),
("2.3", "API3", "对象属性级别授权失效"),
("2.4", "API4", "资源消耗不受限"),
("2.5", "API5", "功能级别授权失效"),
("2.6", "API6", "无限制访问敏感业务流"),
("2.7", "API7", "服务器端请求伪造"),
("2.8", "API8", "安全配置错误"),
("2.9", "API9", "库存管理不当"),
("2.10", "API10", "不安全的第三方API集成"),
]
# 风险等级映射(平台 level -> 报告显示)
LEVEL_MAP = {
"critical": ("严重", "🔴🔴🔴"),
"high": ("高危", "🔴"),
"medium": ("中危", "🟠"),
"low": ("低危", "🟡"),
"info": ("信息", "🔵"),
}
# 已知安全问题清单(测试前已确认)
KNOWN_ISSUES = [
("KN-01", "维护平台验证码固定为 `csba`,可被暴力破解", "🟠中危", "需求文档"),
("KN-02", "登录密码使用 SHA256 加密(非加盐哈希),可能被彩虹表破解", "🟠中危", "网络抓包"),
("KN-03", "`company_secret` 明文出现在 URL 参数中(讯飞转录接口)", "🔴高危", "网络抓包"),
("KN-04", "内部接口 `/api/system/inner/*` 可从外部访问", "🔴高危", "接口清单"),
("KN-05", "无登录接口 `/api/cdthApi/noLogin/*` 可无需认证访问", "🔴高危", "接口清单"),
("KN-06", "登录请求头含 `x-sign` 签名但签名算法在前端 JS 可逆向", "🟠中危", "网络抓包"),
]
# 默认服务器架构信息(对齐参考实现)
DEFAULT_SERVER_ARCH = "X86架构-欧拉"
# 参考资料(内网盘路径,对齐参考实现)
REFERENCE_MATERIALS = [
"华为安全红线资料:[Z:\\deploy\\18其它系统\\安全测试\\01安全测试资料]",
"以往漏洞报告资料:[Z:\\deploy\\18其它系统\\安全测试\\02项目漏洞资料]",
"Nginx配置文件:[Z:\\deploy\\18其它系统\\安全测试\\03Nginx配置文件]",
]
class SecurityReportService:
"""
安全测试报告生成服务
生成 7 章节 Markdown 格式报告,迁移自参考实现 report_generator.py。
"""
def __init__(self, report_dir: Optional[str] = None):
"""
初始化报告服务
Args:
report_dir: 报告输出目录,默认使用 settings.REPORT_DIR
"""
self.report_dir = Path(report_dir or settings.REPORT_DIR)
self.report_dir.mkdir(parents=True, exist_ok=True)
async def generate_report(
self,
execution_id: str,
db: AsyncSession,
) -> str:
"""
生成安全测试 Markdown 报告
Args:
execution_id: 执行记录ID
db: 数据库会话
Returns:
str: 报告文件路径
"""
# 1. 获取执行记录
execution = await self._get_execution(execution_id, db)
if not execution:
raise ValueError(f"执行记录不存在: {execution_id}")
# 2. 获取漏洞测试结果
results = await self._get_vulnerability_results(execution_id, db)
# 3. 获取安全测试配置(用于提取目标URL)
config = await self._get_security_config(execution, db)
target_url = config.target_url if config else ""
server_ip = config.server_ip if config else self._extract_ip(target_url)
# 4. 生成报告内容
markdown = self._generate_markdown(
execution=execution,
results=results,
target_url=target_url,
server_ip=server_ip,
)
# 5. 保存报告文件
file_path = self._save_report(
server_ip=server_ip,
markdown=markdown,
)
logger.info(f"安全测试报告已生成: {file_path}")
return file_path
async def get_report_content(
self,
execution_id: str,
db: AsyncSession,
) -> str:
"""
获取报告内容(不保存文件)
Args:
execution_id: 执行记录ID
db: 数据库会话
Returns:
str: Markdown 报告内容
"""
execution = await self._get_execution(execution_id, db)
if not execution:
raise ValueError(f"执行记录不存在: {execution_id}")
results = await self._get_vulnerability_results(execution_id, db)
config = await self._get_security_config(execution, db)
target_url = config.target_url if config else ""
server_ip = config.server_ip if config else self._extract_ip(target_url)
return self._generate_markdown(
execution=execution,
results=results,
target_url=target_url,
server_ip=server_ip,
)
# ==================== 数据查询 ====================
async def _get_execution(
self,
execution_id: str,
db: AsyncSession,
) -> Optional[Execution]:
"""获取执行记录"""
result = await db.execute(
select(Execution).where(Execution.id == execution_id)
)
return result.scalar_one_or_none()
async def _get_vulnerability_results(
self,
execution_id: str,
db: AsyncSession,
) -> List[VulnerabilityResult]:
"""获取漏洞测试结果列表"""
result = await db.execute(
select(VulnerabilityResult)
.where(VulnerabilityResult.execution_id == execution_id)
.order_by(VulnerabilityResult.level, VulnerabilityResult.test_id)
)
return list(result.scalars().all())
async def _get_security_config(
self,
execution: Execution,
db: AsyncSession,
) -> Optional[SecurityConfig]:
"""获取安全测试配置(优先从执行记录获取,回退到默认配置)"""
config_id = execution.config.get("config_id") if execution.config else None
if config_id:
result = await db.execute(
select(SecurityConfig).where(SecurityConfig.id == config_id)
)
config = result.scalar_one_or_none()
if config:
return config
# 回退:使用默认配置
result = await db.execute(
select(SecurityConfig).where(SecurityConfig.is_default == True)
)
return result.scalar_one_or_none()
# ==================== Markdown 生成 ====================
def _generate_markdown(
self,
execution: Execution,
results: List[VulnerabilityResult],
target_url: str,
server_ip: str,
) -> str:
"""
生成 Markdown 报告内容
Args:
execution: 执行记录
results: 漏洞测试结果列表
target_url: 目标服务器URL
server_ip: 服务器IP
Returns:
str: Markdown 报告内容
"""
lines = []
# 统计数据
summary = self._get_summary(results)
# ========== 标题与基本信息 ==========
lines.append("# 接口安全测试总报告\n")
lines.append(f"> **生成时间**:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"> **测试目标**:{target_url}")
lines.append(f"> **测试标准**:OWASP API Security Top 10 (2019)")
lines.append(f"> **测试工具**:平台自动化测试系统 - 安全测试模块")
lines.append("")
# ========== 一、执行概要 ==========
lines.append("## 一、执行概要\n")
lines.append(f"本次安全测试共执行 **{summary['total']}** 个测试用例。\n")
# 1.1 测试统计总览
lines.append("### 1.1 测试统计总览\n")
lines.append("| 统计项 | 数值 |")
lines.append("|--------|------|")
lines.append(f"| 测试用例总数 | **{summary['total']}** |")
lines.append(f"| 发现漏洞总数 | **{summary['vuln_total']}** |")
lines.append(f"| 🔴 高危漏洞 | **{summary['high']}** |")
lines.append(f"| 🟠 中危漏洞 | **{summary['medium']}** |")
lines.append(f"| 🟡 低危漏洞 | **{summary['low']}** |")
lines.append(f"| 🔵 信息类 | **{summary['info']}** |")
lines.append(f"| 🟢 已验证安全 | **{summary['safe']}** |")
lines.append(f"| 测试时间 | {self._format_time_range(execution)} |")
lines.append("")
# 1.2 OWASP 覆盖矩阵(对齐参考实现:5列,无"严重"列)
lines.append("### 1.2 OWASP API Security Top 10 覆盖矩阵\n")
lines.append("| 编号 | 安全风险 | 用例数 | 🔴高危 | 🟠中危 | 🟡低危 | 🔵信息 | 🟢安全 |")
lines.append("|------|---------|--------|--------|--------|--------|--------|--------|")
for prefix, mod_key, mod_name in OWASP_MODULES:
mod_stats = self._get_module_stats(results, prefix)
lines.append(
f"| {mod_key} | {mod_name} | {mod_stats['total']} | "
f"{mod_stats['high']} | {mod_stats['medium']} | "
f"{mod_stats['low']} | {mod_stats['info']} | {mod_stats['safe']} |"
)
# 合计行
lines.append(
f"| **合计** | | **{summary['total']}** | "
f"**{summary['high']}** | **{summary['medium']}** | "
f"**{summary['low']}** | **{summary['info']}** | **{summary['safe']}** |"
)
lines.append("")
# ========== 二、漏洞详情 ==========
vuln_results = [r for r in results if r.is_vulnerable]
lines.append("## 二、漏洞详情\n")
lines.append(
f"> 共发现 **{summary['vuln_total']}** 个漏洞"
f"(高危 {summary['high']} / "
f"中危 {summary['medium']} / "
f"低危 {summary['low']} / 信息类 {summary['info']})\n"
)
if vuln_results:
# 按风险等级排序
level_order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
vuln_results.sort(key=lambda x: level_order.get(x.level, 99))
for idx, r in enumerate(vuln_results, 1):
level_name, level_emoji = LEVEL_MAP.get(r.level, (r.level, "⚪"))
lines.append(f"### 2.{idx} [{level_name}] {r.name}\n")
lines.append(f"- **用例编号**: {r.test_id}")
lines.append(f"- **风险等级**: {level_emoji} {level_name}")
lines.append(f"- **漏洞描述**: {r.description}")
lines.append("")
if r.request_info:
lines.append("**复现步骤 — 请求信息:**")
lines.append("```")
lines.append(self._truncate(r.request_info, 2000))
lines.append("```")
lines.append("")
if r.response_info:
lines.append("**响应信息:**")
lines.append("```")
lines.append(self._truncate(r.response_info, 2000))
lines.append("```")
lines.append("")
if r.fix_suggestion:
lines.append(f"**修复建议**: {r.fix_suggestion}")
lines.append("")
lines.append("---")
lines.append("")
else:
lines.append("本次测试未发现安全漏洞。\n")
# ========== 三、已验证安全项 ==========
safe_results = [r for r in results if not r.is_vulnerable]
lines.append("## 三、已验证安全项\n")
if safe_results:
for prefix, mod_key, mod_name in OWASP_MODULES:
module_safe = [r for r in safe_results if r.test_id.startswith(prefix)]
if module_safe:
lines.append(f"### {mod_key} {mod_name}\n")
lines.append("| 编号 | 测试用例 | 结果 |")
lines.append("|------|---------|------|")
for r in module_safe:
lines.append(f"| {r.test_id} | {r.name} | ✅ 安全 |")
lines.append("")
else:
lines.append("无已验证安全项。\n")
# ========== 四、风险评估与修复建议 ==========
lines.append("## 四、风险评估与修复建议\n")
if summary["high"] > 0:
lines.append("### 4.1 高危风险(需立即修复)\n")
if any(r.level in ("critical", "high") and r.test_id.startswith("2.1") for r in vuln_results):
lines.append("#### 越权漏洞(API1 / API3)")
lines.append("- **问题描述**:普通用户可访问其他用户的数据(水平越权),普通用户可访问管理员接口(垂直越权)")
lines.append("- **影响范围**:用户数据泄露、数据篡改、未授权操作")
lines.append("- **修复建议**:")
lines.append(" 1. 后端接口增加严格的资源所有权校验,确保用户只能操作属于自己的数据")
lines.append(" 2. 对所有写操作(修改/删除)进行资源归属验证")
lines.append(" 3. 实现基于 RBAC 的权限控制,确保普通用户无法调用管理接口")
lines.append(" 4. 在 API 网关层增加统一的权限拦截器")
lines.append("")
if any(r.level in ("critical", "high") and r.test_id.startswith("2.2") for r in vuln_results):
lines.append("#### 身份认证缺陷(API2)")
lines.append("- **问题描述**:存在 Token 伪造可用、暴力破解无限制等问题")
lines.append("- **影响范围**:账号被盗用、未授权访问")
lines.append("- **修复建议**:")
lines.append(" 1. 加强 Token 校验机制,增加 IP 绑定或设备指纹校验")
lines.append(" 2. 实现 Token 黑名单机制,支持主动注销")
lines.append(" 3. 登录接口增加速率限制(如 5 次失败后锁定 15 分钟)")
lines.append(" 4. 验证码应随机生成,不应使用固定值")
lines.append("")
if any(r.level in ("critical", "high") and r.test_id.startswith("2.8") for r in vuln_results):
lines.append("#### 安全配置错误(API8)")
lines.append("- **问题描述**:中间件管理界面可匿名访问、敏感信息泄露")
lines.append("- **影响范围**:系统配置泄露、攻击面扩大")
lines.append("- **修复建议**:")
lines.append(" 1. Nginx 层面限制 `/nacos` 等路径仅内网可访问")
lines.append(" 2. 生产环境禁用 Swagger 或限制内网访问")
lines.append(" 3. 配置全局异常处理器,隐藏技术细节")
lines.append("")
if summary["medium"] > 0:
lines.append("### 4.2 中危风险(建议尽快修复)\n")
lines.append("请在漏洞详情章节查看具体中危漏洞的修复建议。\n")
if summary["low"] > 0:
lines.append("### 4.3 低危风险(建议择期修复)\n")
lines.append("请在漏洞详情章节查看具体低危漏洞的修复建议。\n")
if summary["vuln_total"] == 0:
lines.append("本次测试未发现安全漏洞,无需修复。\n")
# ========== 五、已知安全问题清单 ==========
lines.append("## 五、已知安全问题清单\n")
lines.append("> 以下问题在测试前已确认,非本次测试发现。\n")
lines.append("| 编号 | 问题描述 | 风险等级 | 来源 |")
lines.append("|------|---------|---------|------|")
for kn_id, desc, level, source in KNOWN_ISSUES:
lines.append(f"| {kn_id} | {desc} | {level} | {source} |")
lines.append("")
# ========== 五.五、历史漏洞回归验证 ==========
regression_results = self._get_regression_results(results)
if regression_results:
lines.append("## 五.五、历史漏洞回归验证\n")
lines.append("> 以下测试用例基于历史漏洞报告和Nginx配置分析,验证已知安全问题是否已修复。\n")
lines.append("| 编号 | 漏洞来源 | 测试项 | 当前状态 | 风险等级 | 详情 |")
lines.append("|------|---------|--------|---------|---------|------|")
for r in regression_results:
source = self._get_vuln_source(r.test_id)
status = "🔴 未修复" if r.is_vulnerable else "✅ 已修复/已验证安全"
level_name, _ = LEVEL_MAP.get(r.level, (r.level, ""))
desc_short = r.description[:80] + "..." if len(r.description) > 80 else r.description
lines.append(f"| {r.test_id} | {source} | {r.name} | {status} | {level_name} | {desc_short} |")
lines.append("")
# ========== 五.六、华为安全红线合规性检查 ==========
hw_results = self._get_huawei_redline_results(results)
if hw_results:
lines.append("## 五.六、华为安全红线合规性检查\n")
lines.append("> 以下检查项基于华为安全红线要求,验证系统是否符合安全基线标准。\n")
lines.append("| 检查编号 | 检查类别 | 要求 | 测试结果 | 合规状态 |")
lines.append("|---------|---------|------|---------|---------|")
for r in hw_results:
hw_info = self._get_hw_info(r.test_id)
hw_id = hw_info.get("check_id", r.test_id)
hw_category = hw_info.get("category", "")
hw_requirement = hw_info.get("requirement", r.name)
if len(hw_requirement) > 60:
hw_requirement = hw_requirement[:57] + "..."
compliance = "🔴 不合规" if r.is_vulnerable else "✅ 合规"
lines.append(f"| {hw_id} | {hw_category} | {hw_requirement} | {r.name} | {compliance} |")
lines.append("")
# ========== 六、测试环境与工具 ==========
lines.append("## 六、测试环境与工具\n")
lines.append("| 项目 | 详情 |")
lines.append("|------|------|")
lines.append(f"| 目标服务器 | {server_ip} ({DEFAULT_SERVER_ARCH}) |")
lines.append("| 测试工具 | Python + requests + pycryptodome |")
lines.append("| 测试框架 | 自研 ApiSecurityTest |")
lines.append("| 测试标准 | OWASP API Security Top 10 (2019) + 华为安全红线 + 历史漏洞回归 |")
lines.append("| 报告格式 | Markdown(一份总报告) |")
lines.append("")
# ========== 七、参考资料 ==========
lines.append("## 七、参考资料\n")
for ref in REFERENCE_MATERIALS:
lines.append(f"- {ref}")
lines.append("")
return "\n".join(lines)
# ==================== 统计辅助方法 ====================
def _get_summary(self, results: List[VulnerabilityResult]) -> Dict[str, int]:
"""
获取统计数据(对齐参考实现:critical 合并到 high,5 列矩阵)
Returns:
dict: 包含 high/medium/low/info/safe/total/vuln_total
"""
summary = {
"total": len(results),
"high": 0, # critical 合并到 high
"medium": 0,
"low": 0,
"info": 0,
"safe": 0,
"vuln_total": 0,
}
for r in results:
if r.is_vulnerable:
# critical 合并到 high,其他保持不变
mapped_level = "high" if r.level == "critical" else r.level
if mapped_level in summary:
summary[mapped_level] += 1
summary["vuln_total"] += 1
else:
summary["safe"] += 1
return summary
def _get_module_stats(self, results: List[VulnerabilityResult], prefix: str) -> Dict[str, int]:
"""
获取指定 OWASP 模块的统计数据(对齐参考实现:critical 合并到 high)
Args:
results: 漏洞测试结果列表
prefix: test_id 前缀(如 "2.1")
Returns:
dict: 包含 total/high/medium/low/info/safe
"""
related = [r for r in results if r.test_id.startswith(prefix)]
stats = {
"total": len(related),
"high": 0, # critical 合并到 high
"medium": 0,
"low": 0,
"info": 0,
"safe": 0,
}
for r in related:
if r.is_vulnerable:
# critical 合并到 high
mapped_level = "high" if r.level == "critical" else r.level
if mapped_level in stats:
stats[mapped_level] += 1
else:
stats["safe"] += 1
return stats
def _get_regression_results(self, results: List[VulnerabilityResult]) -> List[VulnerabilityResult]:
"""获取历史漏洞回归测试结果"""
# 通过 test_id 前缀 HV- 识别,或通过 extra_data.is_regression
regression = []
for r in results:
if r.test_id.startswith("HV-"):
regression.append(r)
elif r.extra_data and r.extra_data.get("is_regression"):
regression.append(r)
return regression
def _get_huawei_redline_results(self, results: List[VulnerabilityResult]) -> List[VulnerabilityResult]:
"""获取华为安全红线检查结果"""
# 通过 test_id 前缀 HW- 识别,或通过 extra_data.is_huawei_redline
hw_results = []
for r in results:
if r.test_id.startswith("HW-"):
hw_results.append(r)
elif r.extra_data and r.extra_data.get("is_huawei_redline"):
hw_results.append(r)
return hw_results
def _get_vuln_source(self, test_id: str) -> str:
"""从知识库获取漏洞来源"""
historical_vulns = KNOWLEDGE_BASE.get("historical_vulns", [])
for v in historical_vulns:
if v.get("vuln_id") == test_id:
return v.get("source_project", "历史漏洞报告")
return "历史漏洞报告"
def _get_hw_info(self, test_id: str) -> Dict[str, str]:
"""从知识库获取华为红线信息"""
huawei_redlines = KNOWLEDGE_BASE.get("huawei_redlines", [])
for r in huawei_redlines:
if r.get("check_id") == test_id:
return {
"check_id": r.get("check_id", ""),
"category": r.get("category", ""),
"requirement": r.get("requirement", ""),
}
return {}
# ==================== 工具方法 ====================
def _format_time_range(self, execution: Execution) -> str:
"""格式化时间范围"""
start = execution.start_time.strftime("%Y-%m-%d %H:%M:%S") if execution.start_time else "未开始"
end = execution.end_time.strftime("%Y-%m-%d %H:%M:%S") if execution.end_time else "未结束"
return f"{start} ~ {end}"
def _extract_ip(self, url: str) -> str:
"""从 URL 提取 IP"""
if not url:
return "unknown"
url = url.strip().rstrip("/")
ip = url.replace("https://", "").replace("http://", "")
ip = ip.split(":")[0]
return ip
def _truncate(self, text: str, max_len: int) -> str:
"""截断文本"""
if len(text) <= max_len:
return text
return text[:max_len-3] + "..."
def _save_report(self, server_ip: str, markdown: str) -> str:
"""保存报告文件"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{server_ip}_安全测试报告_{timestamp}.md"
filepath = self.report_dir / filename
filepath.write_text(markdown, encoding="utf-8")
return str(filepath)
\ No newline at end of file
......@@ -405,12 +405,32 @@ class SecurityService:
await self.db.commit()
logger.info(f"安全测试执行完成: {execution_id}, 通过: {passed}, 失败: {failed}")
# 自动生成 Markdown 报告(失败不影响执行结果)
try:
report_path = await self._generate_report(execution_id)
if report_path:
# 将报告路径写入 execution.config
await self.db.refresh(execution)
exec_config = dict(execution.config or {})
exec_config["report_path"] = report_path
execution.config = exec_config
await self.db.commit()
logger.info(f"安全测试报告已生成: {report_path}")
except Exception as report_err:
logger.warning(f"生成安全测试报告失败(不影响执行结果): {report_err}")
except Exception as e:
logger.error(f"安全测试执行异常: {e}")
execution.status = "failed"
execution.end_time = datetime.utcnow()
await self.db.commit()
async def _generate_report(self, execution_id: str) -> Optional[str]:
"""生成安全测试 Markdown 报告"""
from app.services.security_report_service import SecurityReportService
report_service = SecurityReportService()
return await report_service.generate_report(execution_id, self.db)
async def _save_vulnerability_result(
self,
execution_id: str,
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论