1. 为什么我要把漏洞利用链分析交给自定义 MCP做安全审计的朋友大概率都经历过这种场景Tabby 跑完代码属性图Neo4j 里躺着几千条从 Controller 到 Sink 的调用路径你一条条点开看眼睛都快瞎了最后发现 90% 是误报。传统 SAST 工具的问题就在这——它能告诉你「这里有一条路径」但没法告诉你「这条路径上的参数到底可不可控、有没有经过过滤」。我试过把整份源码丢给大模型让它审结果 Token 烧得飞快模型还经常在无关的 JDK 代码里绕圈。后来想明白了模型需要的不是全量代码而是精准的调用链上下文。这正是自定义 MCPModel Context Protocol服务能解决的问题——它像一个中间层把图数据库里的调用链、源码里的方法片段按需喂给模型。这篇要讲的就是怎么用自定义 MCP 把「Tabby 生成调用链 → Neo4j 提取路径 → MCP 提取上下文 → 模型语义审计」串成闭环。适合有 Java 审计基础、想用 AI 提效的安全研究员也适合想理解 MCP 实战用法的开发者。全程可跟做配置和代码都能直接复制。2. 前置准备TaoToken 统一 Key 与 MCP 运行环境在写 MCP Server 之前得先解决模型调用的问题。自定义 MCP 服务本身不产生推理能力它需要调用大模型 API 来做语义审计。这里我用 TaoToken 作为统一通道好处是一个 Key 能覆盖多种模型切换模型不用改代码MCP Server 里只维护一份配置。TaoToken 的定位是模型 API 聚合通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要先去控制台创建一个 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 复制出来。环境依赖清单如下建议用 Python 3.10 以上组件版本/说明用途Python3.10运行 MCP Serverfastmcp最新版快速构建 MCP 服务openpyxl最新版读取 Tabby 导出的 xlsxNeo4j4.x/5.x存储代码属性图Tabby最新 jar生成 CPGTaoToken Key控制台获取模型调用凭证安装依赖一条命令搞定pip install fastmcp openpyxl neo4j requestsTaoToken 的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面写了 OpenAI 兼容的调用方式。MCP Server 里我会用 requests 直接调避免引入过重的 SDK。3. 可复制配置MCP Server 骨架与 config.toml先给 MCP 服务端一个清晰的目录结构这样后面排障时能快速定位文件sink_mcp/ ├── config.toml # 服务配置 ├── server.py # MCP 主入口 ├── chain_loader.py # 调用链加载 ├── source_locator.py # 源码定位 └── output.xlsx # Tabby 导出的调用链3.1 config.toml 骨架把模型通道、Neo4j 连接、项目路径都收进配置文件避免硬编码[model] # TaoToken 统一通道 base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model gemini-3-flash timeout 60 [neo4j] uri bolt://localhost:7687 user neo4j password your_password [project] # Java 项目源码根目录 source_root ./springboot-manager-master/src/main/java # Tabby 导出的调用链文件 chain_file ./output.xlsx [audit] # 调用链最大深度 max_depth 5 # 是否跳过末端 Sink 节点JDK/三方库代码 skip_sink_node true这里base_url指向 TaoToken 的 API 入口api_key从控制台拿。模型名可以按需换TaoToken 支持多种模型改一行配置就行不用动代码。3.2 MCP Server 主入口server.py 的核心是暴露两个工具list_all_chains和analyze_chain。前者让模型知道有哪些链后者按需返回某条链的源码上下文。import re import tomllib from pathlib import Path from typing import Any, Dict, List from fastmcp import FastMCP from openpyxl import load_workbook # 加载配置 with open(config.toml, rb) as f: CFG tomllib.load(f) mcp FastMCP(sink_mcp) ROOT Path(CFG[project][source_root]) CHAIN_FILE Path(CFG[project][chain_file]) def load_chains() - List[List[str]]: 读取 Tabby 导出的调用链每行一条链 wb load_workbook(CHAIN_FILE, read_onlyTrue) ws wb.active chains [] for row in ws.iter_rows(values_onlyTrue): cells [str(c) for c in row if c] if cells: chains.append(cells) return chains mcp.tool() def list_all_chains() - List[Dict[str, Any]]: 列出所有潜在调用链的摘要供模型初步筛选 chains load_chains() result [] for idx, chain in enumerate(chains): result.append({ row_index: idx, length: len(chain), head: chain[0], tail: chain[-1], }) return resultlist_all_chains只返回摘要不返回源码这样模型第一轮扫描时 Token 消耗极低。等它锁定可疑链再调analyze_chain拿详细上下文。3.3 源码定位与方法提取这是整个 MCP 最关键的逻辑。调用链里的节点格式是method(class.fqn)需要根据全限定类名找到 Java 文件再提取对应方法的代码片段。def find_class_file(fqn: str) - Path | None: 根据全限定类名定位 Java 源码文件 parts fqn.split(.) candidate ROOT.joinpath(*parts).with_suffix(.java) if candidate.exists(): return candidate # 处理内部类Outer$Inner - Outer.java simple parts[-1].split($)[0] .java for path in ROOT.rglob(simple): return path return None def extract_method_code(content: str, method_name: str) - str: 用大括号计数法提取完整方法体 lines content.splitlines() start_re re.compile( rf^\s*(?:public|protected|private|static|\s)*[\w\[\]\s]\s rf{re.escape(method_name)}\s*\( ) start -1 for i, line in enumerate(lines): if start_re.search(line): start i break if start -1: return depth 0 snippet [] for line in lines[start:]: snippet.append(line) depth line.count({) - line.count(}) if depth 0 and { in .join(snippet): break return \n.join(snippet)extract_method_code只返回方法体不返回整个文件。实测下来一个 800 行的 Java 文件提取单个方法后通常只剩 30 到 50 行Token 消耗能降一个数量级。3.4 analyze_chain 工具定义把上面两个函数组装成 MCP 工具返回结构化上下文mcp.tool() def analyze_chain(row_index: int) - Dict[str, Any]: 获取指定调用链的完整审计上下文仅含相关方法源码 chains load_chains() if row_index len(chains): return {error: row_index 越界} chain chains[row_index] involved {} for cell in chain: m re.match(r(\w)\(([\w\.\$])\), cell) if not m: continue method, fqn m.groups() # 跳过末端 Sink 节点 if CFG[audit][skip_sink_node] and cell chain[-1]: continue if fqn in involved: continue path find_class_file(fqn) if path: code extract_method_code(path.read_text(encodingutf-8), method) involved[fqn] {method: method, code: code} return { row_index: row_index, chain_sequence: - .join(chain), involved_classes: involved, prompt_hint: 请扮演高级安全审计专家判断 source 参数是否可控并到达 sink。 }prompt_hint是给模型的引导语让它聚焦在参数可控性判断上而不是泛泛地描述代码。3.5 在编辑器中加载 MCP以 Trae 为例在mcpServers配置里加上我们的服务{ mcpServers: { sink_mcp: { command: python, args: [d:\\sink_mcp\\server.py], env: { PYTHONPATH: d:\\sink_mcp } } } }保存后重启编辑器MCP 面板里应该能看到sink_mcp已连接工具列表里有list_all_chains和analyze_chain。4. 验证请求跑通一条漏洞链分析配置完成后得验证整条链路能不能跑通。分两步先用 Neo4j 提取调用链再用 MCP 让模型审计。4.1 用 Cypher 提取调用链假设 Tabby 已经把 CPG 导入 Neo4j执行下面这条查询从 Web 入口找通往危险函数的路径MATCH (source:Method) MATCH (source_c:Class) WHERE source.CLASSNAME source_c.NAME AND source_c.ANNOTATIONS IN [RequestMapping,PostMapping,GetMapping] MATCH (sink:Method)-[:CALL]-(m1:Method) WHERE sink.NAME IN [exec,eval,upload,delete] CALL apoc.algo.allSimplePaths(source, m1, ALIAS|CALL, 5) YIELD path RETURN * LIMIT 100把结果导出为output.xlsx每行一条链格式类似upload(com.demo.controller.FileController) - save(com.demo.service.FileService) - write(com.demo.util.FileUtil)4.2 让模型审计这条链在编辑器里对模型说「调用 list_all_chains找出涉及文件上传的链然后 analyze_chain 分析它」。模型会先拿到摘要列表锁定目标行号再请求详细上下文。MCP 返回的involved_classes里包含每个方法的源码片段模型据此推理参数传递。比如它会判断FileController.upload的MultipartFile参数有没有做后缀校验FileService.save有没有拼接用户可控路径最终FileUtil.write的落盘路径是否可被外部控制。一次典型的成功返回长这样{ row_index: 12, chain_sequence: upload(FileController) - save(FileService) - write(FileUtil), involved_classes: { com.demo.controller.FileController: { method: upload, code: public String upload(MultipartFile file) { ... } }, com.demo.service.FileService: { method: save, code: public void save(MultipartFile f) { ... } } }, prompt_hint: 请扮演高级安全审计专家... }模型拿到这段上下文后会输出类似「upload方法未校验文件后缀save直接使用原始文件名拼接路径存在任意文件上传风险」的结论。如果它判断参数不可控也会明确说明理由比如「删除路径由数据库 ID 决定用户无法直接控制」。4.3 模型调用的接入代码MCP Server 内部调用 TaoToken 的代码很简单OpenAI 兼容格式import requests def call_model(prompt: str) - str: resp requests.post( f{CFG[model][base_url]}/v1/chat/completions, headers{ Authorization: fBearer {CFG[model][api_key]}, Content-Type: application/json, }, json{ model: CFG[model][model], messages: [{role: user, content: prompt}], temperature: 0.2, }, timeoutCFG[model][timeout], ) resp.raise_for_status() return resp.json()[choices][0][message][content]temperature设低一点审计场景需要稳定判断不需要发散。模型名从 config.toml 读想换模型改配置即可。5. 本篇常见错排查跑这套流程时我踩过几个坑列出来帮你省时间。MCP 工具不显示检查编辑器 MCP 配置里的args路径是不是绝对路径Windows 下反斜杠要转义。另外fastmcp版本太旧会导致工具注册失败升级到最新版。find_class_file 返回 None多半是source_root配错了。注意source_root要指向src/main/java这一层不是项目根目录。内部类的情况用$分割处理如果还是找不到检查包名和目录结构是否一致。extract_method_code 提取不完整大括号计数法对 Lambda 表达式和匿名内部类会误判。如果方法体里有{出现在字符串里也会算错。这种情况可以退化成按行提取或者用 JavaParser 做精确解析。模型调用返回 401TaoToken 的 Key 没配对或者base_url写成了带/v1的地址。注意base_url只到https://taotoken.net/api路径拼接在代码里做。Key 去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 重新复制一次。Token 消耗还是很大检查skip_sink_node有没有开。末端 Sink 通常是 JDK 或三方库代码模型不需要看。另外list_all_chains只返回摘要别在第一步就把源码全塞进去。Neo4j 查询超时allSimplePaths在深度 5 以上会爆炸把max_depth降到 4 或 5并且给sink.NAME加索引。Tabby 导入后记得跑CREATE INDEX加速。MCP 返回的源码乱码Java 文件编码不是 UTF-8 时read_text会报错。加个errorsignore参数或者显式指定encodinggbk。6. 长期编码与 Agent 场景的接入建议如果你只是偶尔跑一次审计上面这套配置够用了。但如果想把 MCP 审计做成日常流程比如每次代码提交后自动跑一遍那模型调用的稳定性和成本就变得重要。这种长期编码或 Agent 场景建议用 TaoToken 的 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。它针对高频调用做了优化比按次计费更适合持续运行的 MCP 服务。配置方式一样把base_url和 Key 换过去就行。另外几个实战建议MCP Server 里加日志把每次analyze_chain的输入输出记到sink_analysis.log方便回溯误报把模型判断结果结构化存成 JSON积累一批后可以反过来优化 Cypher 查询的 Sink 列表如果审计的链特别多可以先用小模型做初筛再用大模型精审TaoToken 的多模型切换能力在这里很省事。模型对话调试可以直接用 https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 不用写代码就能验证 prompt 效果。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题先查这里。整套流程跑通后你会发现漏洞利用链分析从「人工翻几千条路径」变成了「模型按需拉取上下文做判断」。MCP 的价值不在于替代分析而在于把正确的上下文在正确的时机喂给模型。
