AI 生成的智能合约代码安全审计框架与自动化方案
一、ChatGPT 写的合约代码，部署后被黑了
团队用 LLM 生成了一个 ERC-20 代币合约，测试网跑了两周一切正常。上主网第三天，攻击者利用 transferFrom 函数中未校验 _value 的漏洞，转走了合约中所有代币。事后分析：这段合约代码包含了一个常见的重入漏洞变种，LLM 在训练数据里见过 1000 次反例但还是生成了漏洞代码。
LLM 生成的代码"看起来对"和"逻辑上安全"之间隔着巨大的差距。智能合约的特殊性：一旦部署不可修改，漏洞的代价可能是全部资产。AI 生成 + 自动审计的组合才是安全的工程实践。
二、AI 合约审计的三层防御框架
flowchart TD
A[AI 生成合约代码] –> B[第一层: 静态分析]
B –> B1[Slither: 已知漏洞模式匹配]
B –> B2[Mythril: 符号执行]
B –> B3[自定义规则: 业务逻辑检查]
B1 –> C{发现漏洞?}
B2 –> C
B3 –> C
C –>|是| FIX[标记 + 修复建议]
C –>|否| D[第二层: 动态验证]
FIX –>|模型修复| A
D –> D1[模糊测试: Echidna/Foundry]
D –> D2[形式化验证: Certora]
D –> D3[分叉测试: 主网状态模拟]
D1 –> E{不变量被破坏?}
D2 –> E
D3 –> E
E –>|是| FIX
E –>|否| F[第三层: AI 专项审计]
F –> F1[重入攻击检查]
F –> F2[整数溢出检查]
F –> F3[访问控制检查]
F –> F4[预言机操纵检查]
F1 –> G[审计报告]
F2 –> G
F3 –> G
F4 –> G
G –> H{风险评分 < 阈值}
H –>|是| I[允许部署]
H –>|否| FIX
三层防御的核心思想：静态分析做广度（扫常见漏洞）、动态验证做深度（测逻辑不变量）、AI 专项审计做针对性检查（合约特有的攻击向量）。
三、Python 实现：自动化安全审计流水线
import json
import subprocess
import tempfile
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional
# ========== 漏洞与风险定义 ==========
class Severity(Enum):
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
INFO = "info"
@dataclass
class Vulnerability:
"""安全漏洞"""
id: str
title: str
description: str
severity: Severity
location: str # 代码位置
recommendation: str # 修复建议
cwe_id: str = "" # 通用弱点枚举 ID
confidence: float = 0.0 # AI 检测置信度
@dataclass
class AuditReport:
"""审计报告"""
contract_name: str
contract_code: str
vulnerabilities: list[Vulnerability] = field(default_factory=list)
risk_score: float = 0.0
passed: bool = False
tool_results: dict = field(default_factory=dict)
@property
def summary(self) -> str:
if not self.vulnerabilities:
return "未发现漏洞"
counts = {}
for v in self.vulnerabilities:
counts[v.severity.value] = counts.get(v.severity.value, 0) + 1
return f"发现漏洞: {counts}"
# ========== 静态分析器（Slither 封装） ==========
class SlitherAnalyzer:
"""Slither 静态分析工具封装"""
def __init__(self, solc_version: str = "0.8.20"):
self.solc_version = solc_version
def analyze(self, contract_code: str, contract_name: str) -> list[Vulnerability]:
"""运行 Slither 分析合约代码"""
vulns = []
# 1. 将合约代码写入临时文件
with tempfile.TemporaryDirectory() as tmpdir:
filepath = Path(tmpdir) / f"{contract_name}.sol"
filepath.write_text(contract_code)
try:
# 2. 运行 Slither
result = subprocess.run(
["slither", str(filepath), "–json", "-"],
capture_output=True,
text=True,
timeout=120,
)
if result.returncode != 0 and result.stderr:
vulns.append(Vulnerability(
id="SLITHER-ERROR",
title="静态分析执行失败",
description=result.stderr[:500],
severity=Severity.INFO,
location="N/A",
recommendation="检查合约代码语法是否正确",
))
return vulns
# 3. 解析 Slither 输出
if result.stdout:
data = json.loads(result.stdout)
for detector in data.get("results", {}).get("detectors", []):
vulns.append(Vulnerability(
id=detector.get("check", "unknown"),
title=detector.get("check", "未知漏洞"),
description=detector.get("description", ""),
severity=self._map_severity(detector.get("impact", "")),
location=self._extract_location(detector),
recommendation=self._get_recommendation(detector),
))
except subprocess.TimeoutExpired:
vulns.append(Vulnerability(
id="SLITHER-TIMEOUT",
title="静态分析超时",
description="Slither 分析超过 120 秒",
severity=Severity.INFO,
location="N/A",
recommendation="考虑拆分合约复杂度",
))
except FileNotFoundError:
vulns.append(Vulnerability(
id="SLITHER-NOT-FOUND",
title="Slither 未安装",
description="pip install slither-analyzer",
severity=Severity.INFO,
location="N/A",
recommendation="安装 Slither: pip install slither-analyzer",
))
return vulns
def _map_severity(self, impact: str) -> Severity:
impact_map = {
"High": Severity.HIGH,
"Medium": Severity.MEDIUM,
"Low": Severity.LOW,
"Optimization": Severity.INFO,
}
return impact_map.get(impact, Severity.INFO)
def _extract_location(self, detector: dict) -> str:
elements = detector.get("elements", [])
if elements:
elem = elements[0]
return f"{elem.get('source_mapping', {}).get('filename_used', '')}:" \\
f"{elem.get('source_mapping', {}).get('lines', ['?'])[0]}"
return "N/A"
def _get_recommendation(self, detector: dict) -> str:
recommendations = {
"reentrancy-eth": "使用 ReentrancyGuard 或 CEI 模式（Check-Effects-Interactions）",
"unchecked-lowlevel": "检查低级调用的返回值",
"arbitrary-send-erc20": "使用 SafeERC20 库的 safeTransfer",
"divide-before-multiply": "先乘后除以避免精度损失",
"uninitialized-state": "确保所有状态变量在构造函数中正确初始化",
}
check = detector.get("check", "")
return recommendations.get(check, "参考 Slither 文档获取修复建议")
# ========== AI 专项检测器 ==========
class AIAuditChecker:
"""AI 驱动的专项安全检查"""
# 已知危害模式的正则检查
DANGER_PATTERNS = {
"reentrancy": {
"pattern": r'\\.call\\{value:\\s*\\w+\\}\\(""\\)',
"description": "低级 call 可能在状态更新前被重入",
"severity": Severity.CRITICAL,
},
"tx_origin_auth": {
"pattern": r'==\\s*tx\\.origin',
"description": "使用 tx.origin 做权限检查易受钓鱼攻击",
"severity": Severity.HIGH,
},
"unchecked_call": {
"pattern": r'\\.call\\(',
"description": "未检查 call 返回值可能导致静默失败",
"severity": Severity.MEDIUM,
},
"timestamp_dependence": {
"pattern": r'block\\.timestamp',
"description": "block.timestamp 可被矿工小幅操纵",
"severity": Severity.MEDIUM,
},
"selfdestruct": {
"pattern": r'selfdestruct\\(',
"description": "selfdestruct 可能导致资金永久锁定",
"severity": Severity.HIGH,
},
}
def check_patterns(self, contract_code: str) -> list[Vulnerability]:
"""检查已知危害模式"""
vulns = []
import re
for check_name, check_info in self.DANGER_PATTERNS.items():
matches = list(re.finditer(check_info["pattern"], contract_code))
for match in matches:
# 获取上下文（前后各一行）
lines = contract_code[:match.start()].count('\\n')
context = contract_code.split('\\n')[max(0, lines-1):lines+2]
vulns.append(Vulnerability(
id=f"AI-{check_name.upper()}",
title=f"可疑模式: {check_name}",
description=check_info["description"],
severity=check_info["severity"],
location=f"第 {lines+1} 行",
recommendation=f"检查此处的 {check_name} 模式，确认是否为预期行为",
confidence=0.85,
))
return vulns
def check_business_logic(self, contract_code: str) -> list[Vulnerability]:
"""业务逻辑安全检查"""
vulns = []
# 检查是否有访问控制
has_access_control = any(
kw in contract_code for kw in ["onlyOwner", "Ownable", "AccessControl"]
)
if not has_access_control:
vulns.append(Vulnerability(
id="AI-NO-ACCESS-CONTROL",
title="缺少访问控制",
description="合约未使用 Ownable 或 AccessControl 模式",
severity=Severity.HIGH,
location="合约整体",
recommendation="实现 Ownable 或基于角色的访问控制",
confidence=0.9,
))
# 检查是否有 transferOwner 两步确认
has_transfer_ownership = "transferOwnership" in contract_code
has_accept_ownership = "acceptOwnership" in contract_code
if has_transfer_ownership and not has_accept_ownership:
vulns.append(Vulnerability(
id="AI-NO-2STEP-OWNERSHIP",
title="Owner 转移缺少两步确认",
description="transferOwnership 应配合 acceptOwnership 使用",
severity=Severity.MEDIUM,
location="transferOwnership 函数",
recommendation="实现两步确认的 Owner 转移流程",
confidence=0.8,
))
return vulns
# ========== 审计流水线 ==========
class ContractAuditPipeline:
"""智能合约审计流水线"""
def __init__(self):
self.slither = SlitherAnalyzer()
self.ai_checker = AIAuditChecker()
def audit(self, contract_code: str, contract_name: str) -> AuditReport:
"""执行完整的审计流水线"""
report = AuditReport(
contract_name=contract_name,
contract_code=contract_code,
)
all_vulns = []
# 第一层：静态分析
print("[Audit] 第一层: 静态分析 (Slither)…")
slither_vulns = self.slither.analyze(contract_code, contract_name)
all_vulns.extend(slither_vulns)
report.tool_results["slither"] = {
"vulns_found": len(slither_vulns),
}
# 第二层：AI 模式检测
print("[Audit] 第二层: AI 模式检测…")
pattern_vulns = self.ai_checker.check_patterns(contract_code)
all_vulns.extend(pattern_vulns)
# 第三层：业务逻辑检查
print("[Audit] 第三层: 业务逻辑检查…")
logic_vulns = self.ai_checker.check_business_logic(contract_code)
all_vulns.extend(logic_vulns)
# 去重（按 ID）
seen_ids = set()
unique_vulns = []
for v in all_vulns:
if v.id not in seen_ids:
seen_ids.add(v.id)
unique_vulns.append(v)
report.vulnerabilities = sorted(
unique_vulns,
key=lambda v: (v.severity != Severity.CRITICAL,
v.severity != Severity.HIGH),
)
# 计算风险评分
report.risk_score = self._calculate_risk_score(report.vulnerabilities)
# 判断是否通过
critical_count = sum(1 for v in report.vulnerabilities if v.severity == Severity.CRITICAL)
high_count = sum(1 for v in report.vulnerabilities if v.severity == Severity.HIGH)
report.passed = critical_count == 0 and high_count == 0
return report
def _calculate_risk_score(self, vulns: list[Vulnerability]) -> float:
"""计算风险评分 (0-100)"""
severity_weights = {
Severity.CRITICAL: 25,
Severity.HIGH: 15,
Severity.MEDIUM: 5,
Severity.LOW: 2,
Severity.INFO: 0.5,
}
score = sum(severity_weights.get(v.severity, 0) for v in vulns)
return min(score, 100)
四、审计框架的边界与局限性
自动化工具不能替代人工审计。 Slither 和 Mythril 能发现 60-70% 的常见漏洞，但不能理解业务逻辑。一个逻辑上正确但会导致资金锁定的合约——自动工具会说"通过"。
AI 生成的审计建议本身可能不准确。 如果用 LLM 来分析合约漏洞，它可能忽略关键上下文或给出错误的修复建议。AI 辅助审计是"帮助发现模式"，不是"最终决策者"。
第三方库依赖也需要审计。 OpenZeppelin 等成熟库已经被审计过无数次，但你使用的特定版本可能包含已知漏洞。审计流水线应包含依赖版本检查。
审计频率要跟上代码变更。 不要只在部署前审计一次。每次合约升级、每次依赖更新，都应该重新运行审计流水线。
五、总结
AI 生成智能合约的安全审计需要三层防护：静态分析扫已知漏洞、动态验证测逻辑不变量、AI 专项检查找智能合约特有的攻击向量。实施建议：先把 Slither 跑起来，每分钟都能帮你发现代码里的问题；再叠加 AI 模式检查覆盖业务逻辑；最后在测试网上用 Echidna 做模糊测试。安全不是一次审计，而是持续集成在 CI/CD 流水线中的自动化检查。
