# CodeGuard RAG MCP Server

> 基于 RAG + MCP 的 Python 代码缺陷与漏洞诊断平台

CodeGuard 接收 Python 报错、traceback 或代码片段，经过静态特征提取和
Dense + BM25 混合检索，返回问题分类、漏洞类型、CWE、风险等级、判断证据、
根因、修复建议、安全代码及验证方法。用户代码只会被解析，不会被执行。

当前版本是个人项目 M1：保留原项目的模块化 RAG、ChromaDB、BM25、RRF、
可选 Rerank、MCP Server、Streamlit Dashboard 和可观测性技术栈，将核心场景
收敛为代码缺陷与安全漏洞诊断。

## 项目定位

该项目解决两类输入：

- 运行时报错：如 `TypeError`、`KeyError`、`ImportError`，输出缺陷根因和修复步骤。
- 危险代码：如 `shell=True`、`eval()`、不安全反序列化，输出漏洞类型、CWE 和安全写法。

M1 仅支持 Python。它是辅助诊断工具，不替代人工代码审计，也不声称已经集成
Bandit、Semgrep 或能够发现所有漏洞。

## 核心能力

- 静态输入解析：提取异常类型、traceback 文件与行号、危险 API 和关键符号。
- 结构化安全知识库：内置 30 条经过 Schema 校验的 Python 缺陷、漏洞、配置和依赖案例。
- 混合检索：Dense Embedding 负责语义匹配，BM25 负责异常名、API、CWE 等精确匹配。
- 确定性诊断：以检索证据生成结构化报告；没有直接代码证据时降低安全结论置信度。
- MCP 接入：通过 `diagnose_code_issue` 向 MCP Client 暴露统一诊断能力。
- 双格式输出：同时返回便于阅读的中文 Markdown 和便于程序消费的 JSON。
- 离线回归：核心测试使用固定 Embedding 和固定检索结果，不依赖外部模型 API。

## 系统架构

```text
报错 / traceback / Python 代码
              │
              ▼
    SecurityInputParser
   异常、位置、危险模式、符号
              │
              ▼
    SecurityQueryBuilder
 精确词 + 安全语义扩展 + CWE
              │
       ┌──────┴──────┐
       ▼             ▼
Dense Retrieval   BM25 Retrieval
ChromaDB/cosine   关键词精确召回
       └──────┬──────┘
              ▼
         RRF Fusion
              │
        Optional Rerank
              │
              ▼
      DiagnosticService
  分类、证据、置信度、修复方案
              │
              ▼
 diagnose_code_issue (MCP)
      Markdown + JSON 报告
```

主要代码位置：

- `src/security/analysis/`：输入解析与检索查询构造。
- `src/security/loaders/`：JSON/JSONL 安全案例加载与校验。
- `src/security/ingestion/`：ChromaDB 与 BM25 双索引写入。
- `src/security/services/`：诊断编排、分类和降级策略。
- `src/mcp_server/tools/diagnose_code_issue.py`：MCP 工具及报告格式。
- `knowledge/security_cases.json`：M1 安全知识库。

## 安全案例数据模型

每条案例包含 `case_id`、`issue_kind`、`error_type`、
`vulnerability_type`、`cwe`、`severity`、症状、危险模式、根因、
脆弱代码、修复方案、安全代码、验证方法和参考来源。

知识库支持 JSON 数组和 JSONL。导入时一条案例生成一个稳定 Chunk，
`case_id` 同时用于 ChromaDB 和 BM25 文档标识，避免两路结果错位。

M1 的 30 条案例构成为：

- 8 条普通代码缺陷
- 17 条安全漏洞
- 3 条配置风险
- 2 条依赖风险

## PDF 与 JSON 处理方式

CodeGuard 主知识库优先使用 JSON/JSONL，因为 CWE、风险等级和修复建议需要稳定的
结构化字段。原有 PDF 摄取链路仍然保留，适合后续导入安全规范、漏洞报告或内部文档：

1. 使用 SHA256 判断文件是否已经处理。
2. 使用 MarkItDown 将 PDF 文本转换为 Markdown。
3. 使用 PyMuPDF 抽取图片，保存到 `data/images/`，并写入 `[IMAGE: id]` 占位符。
4. 可选使用 Vision LLM 为图片生成描述；失败时降级为纯文本处理。
5. 对文档分块并补充 Metadata。
6. 同时写入 Dense 向量库与 BM25 索引。

PDF 是通用文档检索入口；`knowledge/security_cases.json` 是当前诊断结果的主要可信依据。

## Dense + BM25 + RRF + Rerank

这里的 Dense Retrieval 不是一个具体算法名，而是一类语义向量检索：

- `EmbeddingFactory` 按 `config/settings.yaml` 选择 DashScope、OpenAI、Azure OpenAI 或 Ollama Embedding。
- 文本向量写入 ChromaDB HNSW 集合，距离空间为 cosine。
- 查询向量与案例向量按 cosine 相似度召回。

另一路使用 BM25 对 `TypeError`、`subprocess.run`、`shell=True`、`CWE-78`
等关键词进行稀疏检索。RRF（Reciprocal Rank Fusion）合并的正是
Dense 语义检索排名和 BM25 关键词检索排名，默认 `rrf_k=60`。
融合后可按配置启用 Cross-Encoder 或 LLM Rerank；M1 默认关闭 Rerank，
便于低成本本地运行。

## 快速开始

以下命令面向 Windows PowerShell，要求 Python 3.11+。

```powershell
cd <project-directory>
py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -U pip
python -m pip install -e ".[dev]"
```

项目默认使用 DashScope 的 OpenAI 兼容接口：LLM 为 `qwen3.7-plus`，Embedding
为 `qwen3.7-text-embedding`（1024 维），Base URL 为
`https://dashscope.aliyuncs.com/compatible-mode/v1`。API Key 只从本机环境变量
`DASHSCOPE_API_KEY` 读取，绝不能写入仓库、`settings.yaml` 或日志。

```powershell
$env:DASHSCOPE_API_KEY="<仅在本机设置，不要写入仓库>"
python scripts\check_dashscope_connectivity.py
```

上述连通性检查是显式执行的：它只发送一次短 LLM 请求和一次单文本 Embedding
请求。正常启动和 Dashboard readiness 只检查本地配置及知识库，不会消耗模型配额。

- ChromaDB 默认目录为 `data/db/chroma`。
- 安全案例 BM25 索引目录为 `data/db/bm25/code_security_cases`。
- 可通过 `DASHSCOPE_BASE_URL` 覆盖 Base URL，适用于后续迁移到业务空间专属域名。

先执行无需 API Key 的基础检查：

```powershell
python main.py
python -m pytest tests\unit\security tests\unit\test_diagnose_code_issue.py -v
```

## 导入安全案例

首次使用或切换 Embedding 模型/维度后，使用当前 DashScope Embedding 重建安全案例
集合：

```powershell
python scripts\ingest_security_cases.py --rebuild
```

`--rebuild` 仅重建 `code_security_cases` 及其 `security_` BM25 索引，不会删除
其他集合或整个数据库目录。导入会调用 Embedding 服务并消耗 token；成功输出应包含
非零案例、Chunk 和向量数量。知识库保存当前 Embedding 的 provider、model、dimensions
标识，若与已有非空集合不一致，系统会要求显式重建，避免混用旧向量。

## 启动 MCP Server

```powershell
python -m src.mcp_server.server
```

MCP Client 的启动配置可使用：

```json
{
  "command": "<project-directory>\\.venv\\Scripts\\python.exe",
  "args": ["-m", "src.mcp_server.server"],
  "cwd": "<project-directory>"
}
```

核心工具输入示例：

```json
{
  "name": "diagnose_code_issue",
  "arguments": {
    "error_message": "",
    "code_snippet": "subprocess.run(user_input, shell=True)",
    "language": "python",
    "top_k": 5
  }
}
```

服务器还保留 `query_knowledge_hub`、`list_collections` 和
`get_document_summary`，方便查看和复用原有 RAG 能力。

## 启动 Dashboard

```powershell
python -m streamlit run src\observability\dashboard\app.py
```

Dashboard 默认打开“漏洞诊断”页，支持粘贴 Python 报错、代码片段或上传单个
UTF-8 `.py` 文件，并可下载 Markdown/JSON 报告。上传内容仅在内存中解析，
不会保存或执行。点击“诊断”后，输入的报错或代码与检索到的案例上下文会发送到
DashScope，用于生成 Qwen 增强的修复说明；诊断同样会消耗 token。请勿提交不应发送到
第三方服务的密钥、个人数据或生产机密。

Embedding 可以稍后配置：未配置 Embedding 或尚未导入
`code_security_cases` 时，页面仍可正常打开，但会提示先完成配置并运行：

```powershell
python scripts\ingest_security_cases.py --rebuild
```

此状态下不会生成模拟诊断结果。

## 诊断示例

输入：

```python
subprocess.run(user_input, shell=True)
```

预期核心结果：

- 分类：`security_vulnerability`
- 类型：`Command Injection`
- CWE：`CWE-78`
- 风险等级：`critical`
- 证据：`subprocess-shell`
- 修复：禁用 `shell=True`，使用参数数组和允许列表校验
- 相似案例：`PY-SEC-002`

完整示例见
[`docs/examples/codeguard-diagnosis-example.md`](docs/examples/codeguard-diagnosis-example.md)。

## 测试与评估

```powershell
python -m pytest tests\unit\security tests\unit\test_diagnose_code_issue.py `
  tests\integration\test_security_case_ingestion.py `
  tests\e2e\test_codeguard_diagnosis.py -v

python -m ruff check src\security `
  src\mcp_server\tools\diagnose_code_issue.py `
  scripts\ingest_security_cases.py `
  tests\unit\security `
  tests\unit\test_diagnose_code_issue.py `
  tests\e2e\test_codeguard_diagnosis.py
```

普通 `python -m pytest` 默认只运行离线测试，并自动移除测试进程及其子进程可见的
DashScope、OpenAI 和 Azure OpenAI API Key。会调用真实模型服务的用例统一标记为
`llm`，必须显式执行；例如：

```powershell
python -m pytest -m llm tests\integration\test_chunk_refiner_llm.py -v
```

只有准备消耗真实模型配额时才运行上述命令。当前支持并验证的客户端版本下限为
`chromadb>=1.5.9` 和 `openai>=2.46.0`。

当前验证覆盖数据模型、案例校验、静态解析、查询扩展、双索引写入、
确定性诊断、MCP 注册和离线端到端输出。项目保留原有 Ragas/Custom
评估模块，但 M1 不提供未经实际实验验证的准确率数字。

## 限制与后续方向

- M1 只解析 Python，不执行待诊断代码。
- 当前危险模式属于可解释规则集合，不等同于完整 SAST。
- 实际 Dense 检索需要可用的 Embedding Provider；没有 API Key 时，
  MCP 初始化和 `tools/list` 仍可工作，但真实混合检索会返回可读的配置错误。
- 无检索结果时返回 `degraded=true`、置信度 `0.0`，提示补充上下文。
- 只有知识库相似性、没有匹配的静态代码证据时，不直接判定漏洞，
  返回 `degraded=true` 和置信度 `0.0`。
- M1 置信度使用可解释的证据分层，不把 RRF、BM25 或 cosine 的异构原始分数
  直接解释为概率。
- 原始代码不会直接拼入远程 Embedding 查询；异常中的常见 API Key、Token、
  Password 和 Bearer 凭据会先被遮盖。
- 后续可增加文件/仓库扫描、Bandit/Semgrep 结果归一化、
  Golden Test Set 指标和多语言支持。

## 简历表述参考

> 独立设计并实现基于 RAG + MCP 的 Python 代码缺陷与漏洞诊断平台，构建
> 30 条结构化安全案例知识库及 JSON/JSONL 校验导入链路；采用 Dense
> Embedding + BM25 双路召回、RRF 融合和可选 Rerank，结合静态危险模式
> 证据输出 CWE、风险等级、根因与修复方案，并通过 MCP 暴露标准化诊断工具；
> 使用 Unit / Integration / E2E 离线测试验证 ChromaDB、BM25 和 MCP stdio
> 全链路。

简历中应只写自己实际运行、理解并能解释的功能，不要填写未经测量的提升比例。