基于Solidity AST与图神经网络的智能合约行级漏洞检测
2026/9/14 19:23:45 网站建设 项目流程

简介:本资源是一套高分毕业设计级的智能合约漏洞检测实践方案,面向计算机、人工智能、区块链等方向的学生与开发者,聚焦源代码层面的图神经网络建模与多维度漏洞识别问题。项目采用AST-GNN等图融合技术,覆盖重入攻击、算术溢出、时间戳依赖等典型漏洞的标签构建、数据流/控制流信息注入、模型训练与预测全流程,适合作为课程设计、毕设参考或安全研究入门范例。压缩包共48个文件,含35个Python核心脚本(如main.py、model.py、data_process.py及各类attack_label生成模块)、6个XML配置与数据定义文件、2个说明文档(含环境还原指南)、1个docx手册、1个yml环境配置及配套gitignore等,整体14.84MB,结构清晰、模块解耦。已有43人学习下载,提供完整可运行代码、详细文档、预处理数据集与测试用例,附带编译工具链、评估指标实现及多种分类模型(合约级/行级)训练脚本,便于快速复现、调试与二次开发。

1. 这不是又一个“调用 web3.py 扫地址”的检测脚本,而是一套从 Solidity 源码 AST 出发、融合控制流与数据流图、用 GNN 做细粒度行级分类的漏洞识别闭环

你可能已经试过 Slither、Mythril 或 Oyente——它们能报出 reentrancy、integer overflow,但报错位置常是函数级,无法定位到具体哪一行赋值语句触发了危险的数据流;你也可能跑过基于 IR 的模型,但 IR 已丢失原始变量命名、注释、结构体嵌套等语义线索。这个毕设项目反其道而行:不编译、不反编译、不依赖 EVM 字节码,直接吃.sol源文件,解析出 AST,再把 AST 节点、控制流边、数据流边三者构建成异构图,喂给一个定制化的 AST-GNN 模型,最终输出每行代码的漏洞概率分。它不是黑盒扫描器,而是可解释、可调试、可逐层可视化图结构的检测流水线。98 分答辩的核心支撑,正是这套“源码 → AST → 图构建 → GNN 编码 → 行级预测”的全链路可控性。适合两类人:一是刚接触智能合约安全的学生,能看清从require(msg.sender == owner)到图节点再到预测标签的完整映射;二是已有静态分析经验的工程师,可直接复用append_control_flow_information.pyappend_data_flow_information.py中的 CFG/DFG 构建逻辑,嵌入自己的检测框架。


2. AST 解析与多维度图构建:为什么必须用 Solidity AST 而非通用编程语言解析器

2.1 Solidity 特异性 AST 提取:绕过通用 parser 的语义失真

Solidity 的语法糖(如a += b展开为a = a + b)、类型推导(uint x;vsuint256 x;)、继承链(is Ownable, ERC20)在通用 AST 工具(如 tree-sitter)中常被扁平化或忽略。本项目采用py-solc-x配合自定义 visitor,先编译生成标准 JSON AST,再用Node.py封装节点类型系统。关键在于保留ContractDefinitionFunctionDefinitionVariableDeclarationBinaryOperation等原生节点,并打上sourceLocation字段(含起始/结束行号、列号),这是后续行级标注的基础。

# utils.py 中 extract_ast_nodes 的核心逻辑 def extract_ast_nodes(sol_file: str) -> List[Dict]: # 使用 py-solc-x 编译获取 AST JSON compiled = compile_files([sol_file], solc_version="0.8.19") ast_json = compiled[sol_file]["ast"] nodes = [] for node in traverse_ast(ast_json): # 自定义深度优先遍历 if node.get("nodeType") in ["VariableDeclaration", "BinaryOperation", "IfStatement"]: # 严格提取 sourceLocation,过滤掉无位置信息的 synthetic node loc = node.get("src", "").split(":") if len(loc) >= 3: nodes.append({ "type": node["nodeType"], "line_start": int(loc[1]), "line_end": int(loc[1]) + (int(loc[2]) // 100), # 粗略估算行数跨度 "code": node.get("name", "") or node.get("operator", ""), "children": [c["id"] for c in node.get("nodes", [])] }) return nodes

提示py-solc-x必须指定与合约匹配的 Solidity 版本(如0.8.19),否则src字段可能为空或格式错乱。line_end的估算逻辑来自 Solidity 官方 AST 文档中src字段的<start>:<length>:<file>格式,length单位是字符数,除以 100 是经验性换算(因平均行宽约 100 字符),实际部署时建议改用asttokens库精确定位。

2.2 控制流图(CFG)与数据流图(DFG)的联合注入

仅靠 AST 节点无法捕获跨语句依赖。例如reentrancy漏洞需识别“外部调用前未更新状态”这一模式,这要求连接state_var = valueexternal_call()之间的数据依赖边。项目通过两阶段注入:

  • append_control_flow_information.py:基于FunctionDefinition节点,用if/else/for/while关键字识别基本块边界,用return/break/continue识别跳转目标,生成CFGEdge(from_node_id, to_node_id, edge_type="conditional"|"unconditional")
  • append_data_flow_information.py:对每个VariableDeclarationBinaryOperation,追踪leftHandSide变量在后续ExpressionStatement中是否作为rightHandSide出现,生成DFGEdge(var_node_id, use_node_id, data_type="taint"|"control")
# append_data_flow_information.py 中的关键追踪逻辑 def build_dfg_for_function(ast_nodes: List[Dict]) -> List[Dict]: dfg_edges = [] # 第一步:收集所有变量声明节点及其作用域 var_decls = {n["id"]: n for n in ast_nodes if n["type"] == "VariableDeclaration"} # 第二步:遍历所有表达式节点,检查其右操作数是否引用已声明变量 for expr_node in [n for n in ast_nodes if n["type"] == "BinaryOperation"]: rhs_vars = extract_identifiers(expr_node.get("rightHandSide", {})) for var_name in rhs_vars: # 在当前函数作用域内查找该变量声明 decl = find_var_declaration_in_scope(var_name, var_decls, expr_node) if decl: dfg_edges.append({ "from": decl["id"], "to": expr_node["id"], "type": "data_dependence", "var_name": var_name }) return dfg_edges

注意extract_identifiers不是简单正则匹配,而是递归解析IdentifierMemberAccess(如msg.sender)、IndexAccess(如arr[i])节点,确保捕获所有变量引用路径。find_var_declaration_in_scope实现了作用域链查找(支持for (uint i=0; i<10; i++)中的i仅在 for 块内有效)。

2.3 异构图融合:AST-GNN 的输入图结构设计

最终输入 GNN 的图不是单一类型,而是三类节点、四类边构成的异构图:

节点类型示例数量级(单合约)
AST_NodeBinaryOperation,IfStatement200–800
ControlFlowBlockBasicBlock_0x1a,ExitBlock10–50
DataVariableowner,balance[msg.sender]5–30
边类型连接语义
ast_child_ofBinaryOperationIdentifierAST 层次结构
cfg_nextBasicBlock_ABasicBlock_B控制流顺序
dfg_taintsownerrequire(msg.sender == owner)数据污染传播
ast_in_blockBinaryOperationBasicBlock_AAST 节点归属基本块

该结构被序列化为 PyTorch Geometric 的HeteroData对象,存入built_vector_dataset.py生成的.pt文件。model.py中的ASTGNN模型会为每类节点初始化独立 embedding,再通过HeteroConv层进行跨类型消息传递。


3. AST-GNN 模型训练与行级漏洞标注:如何让模型学会“看懂” Solidity 语义

3.1 漏洞标签的生成逻辑:从合约级标签到行级概率

公开数据集(如 SmartBugs)只提供合约级标签(has_reentrancy: True/False),但本项目要求行级预测。解决方案是规则驱动+人工校验的半自动标注

  • make_reentry_attack_label.py:扫描所有call.value()delegatecallstaticcall调用点,向前追溯最近的state_var = ...赋值语句(在同函数内且无条件分支阻断),将该赋值行标记为reentrancy_risk: 0.9
  • make_timestamp_attack_label.py:匹配block.timestampnow的使用,并检查是否用于访问控制(如if (block.timestamp > deadline)),将该行标记为timestamp_dependency: 0.8
  • make_arithmetic_attack_label.py:用asttokens精确定位+,-,*,/操作符所在行,若操作数含uint类型且无require(a > b)校验,则标记为arithmetic_overflow: 0.7
# make_reentry_attack_label.py 中的追溯逻辑 def find_state_update_before_call(call_node: Dict, ast_nodes: List[Dict]) -> Optional[Dict]: # 获取 call_node 所在行号 call_line = call_node["line_start"] # 向上搜索同一函数内的 VariableDeclaration 节点 for node in reversed(ast_nodes): if node["type"] == "VariableDeclaration" and node["line_start"] < call_line: # 检查该变量是否为状态变量(有 storage 关键字或无 memory 关键字) if "storage" in node.get("visibility", "") or "memory" not in node.get("visibility", ""): # 检查从该赋值到 call 之间是否有 if/for/while 阻断(简单启发式) if not has_control_flow_between(node["line_end"], call_line, ast_nodes): return node return None

提示has_control_flow_between并非精确 CFG 分析,而是扫描行号区间内是否存在if(for(while(字符串,这是在毕设精度与实现复杂度间的务实折中。真实工业场景应替换为CFGEdge查询。

3.2 AST-GNN 模型架构:三层异构消息传递的设计意图

model.py中的ASTGNN不是简单堆叠 GCN 层,而是针对 Solidity 语义设计的三阶段处理:

  1. AST Embedding Layer:对AST_Node类型节点,用TransformerConv处理ast_child_of边,学习语法结构(如IfStatement下必有conditiontrueBody子节点)
  2. Cross-Modal Fusion Layer:用HeteroConv聚合ast_in_block边,将 AST 节点特征注入ControlFlowBlock节点,使基本块获得其内部 AST 的语义摘要
  3. Line-Level Prediction Head:对每个AST_Node节点,拼接其自身 embedding 与所属ControlFlowBlock的 embedding,经Linear(256, 128) → ReLU → Linear(128, 4)输出四类漏洞概率(reentrancy / timestamp / overflow / unchecked_call)
# model.py 中的 forward 方法关键片段 class ASTGNN(torch.nn.Module): def __init__(self, hidden_channels=128): super().__init__() # Step 1: AST-specific encoding self.ast_conv = TransformerConv( in_channels=64, out_channels=hidden_channels, heads=4, dropout=0.1, edge_dim=16 ) # Step 2: Heterogeneous fusion self.hetero_conv = HeteroConv({ ('AST_Node', 'ast_in_block', 'ControlFlowBlock'): SAGEConv((-1, -1), hidden_channels), ('ControlFlowBlock', 'cfg_next', 'ControlFlowBlock'): GATv2Conv((-1, -1), hidden_channels), }, aggr='sum') # Step 3: Line-level classification self.classifier = Sequential( Linear(hidden_channels * 2, 128), ReLU(), Dropout(0.3), Linear(128, 4) # 4 vulnerability types ) def forward(self, data: HeteroData) -> Tensor: # Encode AST nodes x_dict = {} x_dict['AST_Node'] = self.ast_conv( data['AST_Node'].x, data['AST_Node', 'ast_child_of', 'AST_Node'].edge_index ) # Fuse with control flow blocks x_dict = self.hetero_conv(x_dict, data.edge_index_dict) # Predict per AST node (i.e., per line) ast_x = x_dict['AST_Node'] block_x = x_dict['ControlFlowBlock'][data['AST_Node'].block_id] # block_id 是预计算的索引 combined = torch.cat([ast_x, block_x], dim=1) return self.classifier(combined)

注意data['AST_Node'].block_id是在dataset.py中预计算并存入HeteroData的张量,避免运行时重复查询。SAGEConv用于ast_in_block边因其聚合邻居时更鲁棒(对 block 内节点数量变化不敏感),GATv2Conv用于cfg_next边因其能学习不同控制流路径的重要性权重。

3.3 训练配置与验证指标:为何选用 F1-macro 而非准确率

数据集由 127 个真实合约(含 OpenZeppelin、Uniswap V2 子集)和 38 个人工构造漏洞合约组成,按 7:2:1 划分训练/验证/测试集。关键配置如下:

参数说明
batch_size8因图大小差异大,小 batch 避免 OOM
learning_rate2e-4AdamW 优化器,配合线性 warmup
epochs120验证集 F1-macro 连续 15 轮不升则早停
loss_fnFocalLoss(alpha=0.75, gamma=2.0)解决四类漏洞样本不均衡(reentrancy 占 42%,unchecked_call 仅 8%)

验证时不用整体准确率,因漏洞行占比常低于 5%,准确率易被大量负样本拉高。metric.py中的compute_f1_macro计算每类漏洞的 F1 值后取平均:

# metric.py def compute_f1_macro(y_true: np.ndarray, y_pred: np.ndarray) -> float: """ y_true: (N,) int array, 0=reentrancy, 1=timestamp, 2=overflow, 3=unchecked_call, 4=clean y_pred: (N, 5) float array, softmax output """ y_pred_labels = np.argmax(y_pred, axis=1) # 计算每类的 precision/recall/f1,忽略 clean 类(类别 4) f1_scores = [] for cls in [0, 1, 2, 3]: tp = np.sum((y_true == cls) & (y_pred_labels == cls)) fp = np.sum((y_true != cls) & (y_pred_labels == cls)) fn = np.sum((y_true == cls) & (y_pred_labels != cls)) precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0 f1_scores.append(f1) return np.mean(f1_scores)

4. 本地环境还原与端到端推理:从解压 ZIP 到输出带高亮的漏洞报告

4.1 MVD-HG 环境还原:Conda 环境与依赖冲突的解决路径

项目附带environment.yml,但直接conda env create -f environment.yml常失败,原因有三:pytorch-geometricpytorch版本强绑定、py-solc-x依赖特定solc二进制、Windows 下tree-sitter编译失败。推荐分步还原:

# 步骤1:创建基础环境(指定 Python 3.9,避坑 pytorch-geometric 2.4+) conda create -n astgnn python=3.9 conda activate astgnn # 步骤2:安装 PyTorch(根据 CUDA 版本选,此处以 CPU 为例) pip install torch==2.0.1+cpu torchvision==0.15.2+cpu torchaudio==2.0.2+cpu -f https://download.pytorch.org/whl/torch_stable.html # 步骤3:安装 PyTorch Geometric(严格对应 torch 2.0.1) pip install torch-scatter torch-sparse torch-cluster torch-spline-conv -f https://data.pyg.org/whl/torch-2.0.1+cpu.html pip install torch-geometric==2.3.0 # 步骤4:安装 Solidity 工具链 pip install py-solc-x==8.4.0 # 手动下载 solc 0.8.19:https://github.com/ethereum/solidity/releases/download/v0.8.19/solc-static-linux chmod +x solc-static-linux sudo mv solc-static-linux /usr/local/bin/solc

提示:若pip install torch-geometric==2.3.0No matching distribution,请确认torch版本是否为2.0.1+cpu(而非2.0.1),末尾+cpu是关键标识。py-solc-x==8.4.0是最后一个兼容solc 0.8.19的版本。

4.2 端到端推理流程:main.py的三阶段执行链

main.py是入口脚本,执行data_process.py → train.py → load_model_to_predict.py三阶段。新手可跳过训练,直接用预训练模型:

# 解压资源包后,进入项目根目录 unzip "基于源代码的图融合的智能合约漏洞检测全部资料+详细文档(高分毕设).zip" cd "基于源代码的图融合的智能合约漏洞检测全部资料+详细文档(高分毕设)" # 步骤1:预处理单个合约(生成 AST + CFG + DFG) python data_process.py --input contracts/sample.sol --output processed/sample.pt # 步骤2:加载预训练模型并预测(模型文件在 model/ 目录下) python load_model_to_predict.py \ --model_path model/best_astgnn_epoch112.pt \ --data_path processed/sample.pt \ --output_report reports/sample_vuln.json # 步骤3:生成 HTML 报告(高亮风险行) python Test.py --json reports/sample_vuln.json --sol contracts/sample.sol --html reports/sample_report.html

Test.py会读取sample_vuln.json中的{"line": 42, "vul_type": "reentrancy", "score": 0.93},在sample.sol第 42 行添加<span class="vul-reentrancy">标签,最终生成带颜色高亮的 HTML。

4.3 漏洞报告解读与可信度验证技巧

生成的sample_vuln.json结构如下:

[ { "line": 87, "vul_type": "reentrancy", "score": 0.89, "explanation": "External call at line 87 uses msg.sender without prior state update; data flow from balance[msg.sender] (line 72) to call.value() (line 87) detected", "ast_path": ["FunctionDefinition", "Block", "ExpressionStatement", "BinaryOperation"] } ]

验证可信度的三个动作

  1. 查 AST 路径:用print_tree.py sample.sol输出 AST,确认ast_path中的节点确实存在且顺序正确
  2. 追数据流:打开processed/sample.pt,用torch.load()查看data['DFGEdge'],确认from节点 ID 是否指向balance[msg.sender]的声明
  3. 比对基线:用 Slither 运行同一合约slither contracts/sample.sol --solc-allow-paths .,看其报告的reentrancy位置是否与本项目第 87 行重合——若重合,可信度大幅提升;若不重合,检查make_reentry_attack_label.py的追溯逻辑是否漏掉中间赋值

注意print_tree.py输出的是缩进式文本 AST,非 JSON。它用solc --ast-json生成原始 AST,再用Node.pypretty_print方法美化,适合快速定位节点类型与行号,无需启动 IDE。


5. 进阶技巧:如何将此框架迁移到新漏洞类型或新语言

5.1 新漏洞类型的快速适配:以 “delegatecall 代理劫持” 为例

要检测delegatecall被恶意合约利用(如Proxy模式中 implementation 地址被篡改),只需新增make_delegatecall_hijack_label.py,其逻辑与make_reentry_attack_label.py高度相似,但检查重点变为:

  • delegatecall调用点是否使用address(this).delegatecall(...)
  • implementation变量是否为publicexternal函数可修改
  • implementation的赋值是否发生在构造函数之外(即运行时可变)
# make_delegatecall_hijack_label.py(伪代码) def detect_delegatecall_hijack(ast_nodes: List[Dict]) -> List[Dict]: hijack_lines = [] # 步骤1:找到所有 delegatecall 调用 for call_node in [n for n in ast_nodes if "delegatecall" in n.get("code", "")]: # 步骤2:检查是否为 address(this).delegatecall(...) if "address(this)" in get_caller_expr(call_node): # 步骤3:向上追溯 implementation 变量 impl_var = find_implementation_var(call_node, ast_nodes) if impl_var and is_publicly_writable(impl_var, ast_nodes): # 步骤4:检查 impl_var 是否在构造函数外被赋值 if not is_assigned_in_constructor(impl_var, ast_nodes): hijack_lines.append({ "line": call_node["line_start"], "vul_type": "delegatecall_hijack", "score": 0.85, "explanation": f"delegatecall on line {call_node['line_start']} uses mutable implementation {impl_var['name']}" }) return hijack_lines

提示is_publicly_writable的判断逻辑是:扫描impl_var的所有赋值节点,若存在FunctionDefinition节点且visibility"public""external",则返回True。这比检查modifier onlyOwner更底层,也更可靠。

5.2 迁移至其他语言:AST-GNN 的跨语言泛化能力边界

本框架的 AST 解析部分(utils.py,Node.py)是 Solidity 特化的,但图构建与 GNN 模型(model.py,dataset.py)具有强泛化性。迁移到 Vyper 或 Rust(Solana)的关键动作是:

动作Solidity 当前做法迁移至 Vyper 的改动
AST 提取py-solc-x+ JSON AST改用vyper-parser库,其输出ast.ASTNode对象,需重写traverse_ast以适配 Vyper 的FunctionDef/Assign/Call节点类型
CFG 构建基于if/for/while字符串匹配Vyper 无for循环,需改为识别for i in range(...)with语句块
DFG 构建追踪Identifier引用Vyper 的self.storage.var访问需特殊处理,extract_identifiers需增加MemberAccess解析逻辑

真正不可迁移的是漏洞规则引擎make_*.py)。Vyper 的reentrancy检测逻辑完全不同(因其默认禁用重入),此时应废弃make_reentry_attack_label.py,转而实现make_vyper_storage_access_label.py,聚焦self.storage的越界读写。

5.3 模型轻量化部署:ONNX 导出与推理加速

train.py训练完成后,可用以下脚本导出 ONNX 模型,供 C++ 或 WebAssembly 调用:

# export_onnx.py import torch from model import ASTGNN from torch_geometric.data import HeteroData # 加载训练好的模型 model = ASTGNN(hidden_channels=128) model.load_state_dict(torch.load("model/best_astgnn_epoch112.pt")) model.eval() # 构造 dummy input(需与 dataset.py 中的 HeteroData 结构一致) dummy_data = HeteroData() dummy_data['AST_Node'].x = torch.randn(50, 64) dummy_data['AST_Node', 'ast_child_of', 'AST_Node'].edge_index = torch.randint(0, 50, (2, 100)) # ... 其他节点和边的 dummy tensor # 导出 torch.onnx.export( model, (dummy_data,), # 注意:HeteroData 是 tuple 输入 "astgnn.onnx", input_names=["hetero_data"], output_names=["vul_probs"], dynamic_axes={ "hetero_data": {0: "num_nodes"}, # 节点数动态 "vul_probs": {0: "num_predictions"} }, opset_version=15 )

导出的astgnn.onnx可用onnxruntime在 CPU 上达到 120ms/合约(RTX 3090 测试),比 PyTorch 原生推理快 3.2 倍,且内存占用降低 60%。这对于集成到 Remix IDE 插件或 CI/CD 流水线至关重要。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询