mcp-threat-analysis

cp-threat-analysis 全网恶意/漏洞模型上下文协议（MCP）服务器检测平台。该仓库实现了设计文档 [`docs/恶意 MCP 检测可执行技术方案.md`](docs/恶意%20MCP%20检测可执行技术方案.md) 中描述的**静态分析**、**语义/LLM 分析**和**风险聚合/评分**层。 发现、规范化、运行时/时间分析、网络沙盒和披露超出了该代码库的范围，但其接口在 `static_analysis` / `semantic_analysis` / `risk_scoring` 产生或消费数据时定义。 ## 仓库布局 ``` docs/ # 设计文档（主文档 + 各层子系统） sql/ # 数据库模式（Postgres + pgvector） 001_schema.sql # 基础模式 002_llm_calls_response.sql # llm_calls.response_json 列 003_remote_observations.sql # remote_observations 表扩展 枚举 LICENSES/ # 第三方（Apache-2.0 来自 Cisco） example-mcps/ # 16 个件（8 个恶意3 个良性 +src/mcp_threat common/ # 、DB 引擎、、子进程包装器 static_analysis/ # 层 target_loader # 获取/提取工 工作目录 extractors # 工具处理程序、AST IO 摘要袋 analyzers/ Semgrep、CodeQL、SCA、清单誉、混淆 /semgrep/ self # 自有规则（代码/文本/清单） translated_from_cisco/ # 从 cisco-ai-defense/mcp-scanner YARA 规则翻译而来 orchestrator.py persistence.py cli.py # mta-static semantic_analysis/ # 语义/LLM 层 # char_layer、tpa_text_rules、tpa_llmowing、 # schema_code_alignment、toxic_flow、_content alignment/ AlignmentOrchestrator +文件数据流 llm/ # 提供程序可的 LLM 客户端（Anthropic + OpenAI 兼容）， # 异步信号量并发、预算、进程内缓存， # llm_calls 审计行持久性 # 编码器vector shadowing 索引 prompts/ # Markdown 提示模板 orchestrator.py cli.py # mta-semantic risk_scoring/ # 风险聚合、评分、分诊路由 aggregator.py cross_validator triage_router.py lifecycle.py weights popularity.py .py # findings.updated轮询监视器 .py # 读取侧 api/server.py # FastAPI 读取 API + 生命周期 POST（启用 CORS） cli.py # mta-risk remote_analysis/ # MCP 服务器探测层（P1） transport/ # streamable_http（P1）；sse / stdio 位于 P3 detectors/ # tls / auth / protocol — 仅 P1 快照 orchestrator.py # 探测 → 检测 → 持久化 persistence.py # remote_observations + 合成工具/静态 cli.py # mta-remote tests/{static_analysis,semantic_analysis_scoring,remote_analysis}/ # 单元测试 ``` ## 前置条件 - Python 3.11+ -15+，带有 `vector`、`pgcrypto`、`pg`、`fuzzystrmatch` 扩展 - `PATH` 上的外部二进制文件（每个都是可选的；如果缺少二进制文件，分析器将跳过自身）： `semgrep`、`codeql`、`trufflehog`、`gitleaks`、`osv-scanner`、`pip-audit`、`npm` - 用于 `semantic_analysis` LLM 检测器的 LLM API 密钥（Anthropic Messages API *或* AI Chat Completions 兼容端点 — Volcengine Ark、DeepSeek、vLLM、Together 等）。如果没有有效的密钥，LLM 客户端将引发 `LLMUnavailable`，基于规则的检测器仍将运行，而 LLM 检测器将干净地短路。 ## 设置 ```bash # 1. 安装 Python 依赖项 python -m venv .venv && source .venv/bin/activate pip install ".[dev 2. cp .env.example $EDITOR .env 3. 初始化数据库b mta psql -d mta -c EXTENSION IF NOT EXISTS CREATE EXTENSION IF NOT EXISTS pgcrypto; CREATE EXTENSION IF NOT EXISTS pg_trgm; CREATE EXTENSION IF NOTstrmatch;" ps mta -f001_schema.sql psd mta -f sql/002_llm_calls_response.sql # 添加 llm_calls.response_json psql -d mta -f sql/_observations.sql #_observations + 扩展 findings.layer ``` ### LLM 提供 语义层的 LLM 客户端通过环境变量提供程序可插拔： ```bash # Anthropic Messages API（默认） MTA_LLM_PROVIDER=anthropic ANTHROPIC_API_KEY=sk-ant-... # 或任何 OpenAI Chat Completions 兼容端点 MTA_LLM_PROVIDER=openai_compatible MTA_LLM_BASE_URL=https://ark.cn-beijing.volces.com/api/v3 MTA_LLM_API_KEY=... MTA_LLM_MODEL_PRIMARY=glm-5.1 MTA_LLM_MODEL_FALLBACK=... thropic 和 OpenAI 兼容提供商的并发上限3 和 16；覆盖。每次调用（包括缓存命中和错误将持续到 `llm_calls` 审计表中，并evidence.llm_call_id` 链接回其发现。 ## 运行 ### 对本地存档或目录运行静态分析 mta-static ./path/to/artifact.tgz \ version 1.2 \ --lang typescript --artifact-type npm_tarball ``` CLI 以 JSON发现结果，并且（除非 `--no-persist`）将它们插入 `findings` 并写入 `static_summaries` 行，由 `semantic_analysis` 消费。 ### 对 扫描的服务器运行语义分析 ```bash mta-semantic --server-id <UUID> --version 1.2.3 ``` 如果没有有效的 LLM API 密钥，则仅运行基于规则的检测器（`char-layer`、`tpa-text`、`shadowing`、`untrusted-content`）。 ### risk_scoring — 聚合一个服务器，或启动轮询摄取器 ```bash # 一次性 mta-risk aggregate <server-uuid> # 后台守护进程：每 60 秒轮询 findings.updated_at，重新聚合 #服务器，将 P0/P1 推到 triage_queue mta-risk ingest ``` ### remote_analysis — 探测实时 MCP 服务器（P1） `mta-remote scan` 是 remote_analysis 层的入口点。它通过 streamable-HTTP 探测实时 MCP 端点（`initialize` → `tools/list` 加 `resources/list` / `prompts/list` 当被广告宣传时），捕获 TLS 证书详细信息，运行 P1 快照检测器，并写入 `remote_observations` 行和每个报告工具的综合 `tools` 行 — 因此现有的语义层基于规则的检测器（`char:hidden-unicode` / `shadow:*` / `tpa-llm` / `untrusted-content:unmarked`）在远程服务器上不变地运行。 ```bash # 一次性探测 + 持久化 mta-remote scan https://mcp.example.com/mcp # 带有 OAuth 承载者 mta-remote scan https://mcp.example.com/mcp --header "Authorization=Bearer $TOKEN" # 自定义服务器名称 + 更严格的超时 mta-remote scan https://mcp.example.com/mcp --canonical-name acme/email-bot --timeout 8 ``` P1 中活跃的检测器：`remote:tls-self-signed` · `remote:tls-near-expiry≤7d → 高，≤30d → 中 `remote:auth-missing`（非环回，非/list，无 Authorization / API-key 头）· `remote-version-mismatch`（报告的 `protocolVersion` 在已知集合之外）。 漂移检测器添加 / 描述符变异 / 模式宽松等）和循环 `mta-remote watch` 守护进程位于 P2。SSE 和 stdio 传输位于。 退出代码为` 表示探测成功，`2` 表示传输/握；失败路径观察仍然被持久化，因此失败本身分析师可见。 ### 风险评分读取 API（ ```bash mta-api --port # GET / # GET /servers # 具有风险 + 活动计数的语料库列表 # GET /servers/{id}/risk # 服务器 + 顶级发现 # GET /servers/{id}/risk/history # GET /triage?priority=P0&status=pending # GET /findings/{id} # 发现 + 链接的 llm_call # GET /corpus/heatmap # 检测器 × 服务器矩阵 # POST /findings/{id}/{suppress|confirm|false-positive} ``` CORS 为 `http://localhost:7137` 和 `http://localhost:5173` 启用，因此只读浏览器客户端可以直接使用 API 测试 ```bash pytest -q ``` `src/mcp_threat_analysis/tests/` 中的单元测试演练检测器、权重表、交叉验证器、字符串和工具处理程序提取器。DB 和 LLM 接触代码路径需要集成环境（Postgres + Anthropic 密钥），如果这些环境不存在，则会自动跳过。 ## 管道数据流 ``` ScanTarget ──► static_analysis Orchestrator ─┬─► findings (layer=static_analysis) └─► static_summaries │ ▼ semantic_analysis Orchestrator ──► findings (layer=semantic_analysis) │ ▼ risk_scoring Ingestor (poll findings.updated_at) │ ├─► Aggregator → servers.risk_score / risk_priority │ + server_risk_history └─► TriageRouter → triage_queue (P0/P1) │ ▼ disclosure (out of scope) ```