文章总结: 本文介绍AI供应链安全工程实践,提出AI-BOM物料清单格式,涵盖模型、数据、代码依赖及推理依赖的追踪,实现全链路溯源。通过模型权重签名、safetensors安全加载及来源验证确保模型完整性,并自动化依赖审计扫描。建议企业建立AI资产清单与验证机制,防范供应链攻击。 综合评分: 85 文章分类: 供应链安全,ai安全,安全开发,安全建设
AI供应链安全工程
原创
pandazhengzheng pandazhengzheng
安全分析与研究
2026年9月10日 22:00 广东
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
一、AI-BOM实现
AI系统的物料清单比传统软件复杂:不仅包含代码依赖,还包含模型权重、训练数据、超参数、评估基准。
1.1 AI-BOM格式设计
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class ModelComponent:
name: str
version: str
source: str # 来源仓库/URL
source_hash: str # 来源内容哈希
weights_hash: str # 权重哈希
license: str
dependencies: list = field(default_factory=list) # 依赖的模型/库
training_data: Optional[str] = None # 训练数据引用
training_code_hash: Optional[str] = None # 训练代码哈希
evaluation_results: Optional[dict] = None # 评估结果
@dataclass
class DataComponent:
name: str
version: str
source: str
source_hash: str
license: str
preprocessing: list = field(default_factory=list) # 预处理管线
statistics: Optional[dict] = None # 统计摘要
@dataclass
class AIBOM:
project: str
version: str
created: datetime
models: list = field(default_factory=list)
datasets: list = field(default_factory=list)
code_dependencies: list = field(default_factory=list)
inference_dependencies: list = field(default_factory=list)
signature: str = ""
1.2 自动生成工具
class AIBOMGenerator:
def __init__(self, project_root):
self.root = project_root
def generate(self):
bom = AIBOM(
project=self._project_name(),
version=self._project_version(),
created=datetime.now(),
)
# 1. 扫描模型依赖
bom.models = self._scan_models()
# 2. 扫描数据依赖
bom.datasets = self._scan_datasets()
# 3. 扫描代码依赖(pip/poetry/requirements)
bom.code_dependencies = self._scan_code_deps()
# 4. 扫描推理依赖(ONNX/TensorRT版本等)
bom.inference_dependencies = self._scan_inference_deps()
# 5. 签名
bom.signature = self._sign(bom)
return bom
def _scan_models(self):
models = []
for model_file in self._find_model_files():
comp = ModelComponent(
name=model_file.name,
version=model_file.version,
source=model_file.source_url,
source_hash=model_file.repo_hash,
weights_hash=hash_file(model_file.path),
license=model_file.license,
)
models.append(comp)
return models
1.3 全链路追踪
class ProvenanceTracker:
"""从训练到部署的全链路追踪"""
def __init__(self, storage):
self.storage = storage
def record_training(self, model_id, data_id, code_hash, hyperparams):
record = {
"event": "training",
"model_id": model_id,
"data_id": data_id,
"code_hash": code_hash,
"hyperparams": hyperparams,
"timestamp": datetime.now().isoformat(),
}
self.storage.append(record)
def record_evaluation(self, model_id, benchmark, results):
record = {
"event": "evaluation",
"model_id": model_id,
"benchmark": benchmark,
"results": results,
"timestamp": datetime.now().isoformat(),
}
self.storage.append(record)
def record_deployment(self, model_id, env, config):
record = {
"event": "deployment",
"model_id": model_id,
"env": env,
"config": config,
"timestamp": datetime.now().isoformat(),
}
self.storage.append(record)
def trace(self, model_id):
"""返回某模型从训练到部署的完整链路"""
return [r for r in self.storage if r.get("model_id") == model_id]
二、模型完整性验证
2.1 权重签名
class ModelSignatureManager:
def __init__(self, private_key, public_key):
self.private = private_key
self.public = public_key
def sign_model(self, model_path):
weights_hash = hash_file(model_path)
signature = sign(weights_hash, self.private)
return {
"model_path": model_path,
"weights_hash": weights_hash,
"signature": signature,
"signer": "model_registry",
"timestamp": datetime.now().isoformat(),
}
def verify_model(self, model_path, manifest_entry):
actual_hash = hash_file(model_path)
if actual_hash != manifest_entry["weights_hash"]:
return False, "hash_mismatch"
if not verify(
manifest_entry["weights_hash"],
manifest_entry["signature"],
self.public
):
return False, "signature_invalid"
return True, "verified"
2.2 safetensors安全加载
safetensors格式避免pickle反序列化漏洞,是加载不可信模型的首选:
from safetensors import safe_open
class SafeModelLoader:
def __init__(self, allowed_keys=None, max_size_gb=10):
self.allowed_keys = allowed_keys
self.max_size = max_size_gb
def load(self, path):
# 1. 文件大小检查
if file_size(path) > self.max_size * 1e9:
raise SecurityError("model_too_large")
# 2. 用safetensors加载(无代码执行)
with safe_open(path, framework="pt") as f:
keys = f.keys()
if self.allowed_keys and not set(keys).issubset(self.allowed_keys):
raise SecurityError(f"unexpected_keys: {set(keys) - self.allowed_keys}")
weights = {k: f.get_tensor(k) for k in keys}
# 3. 权重统计检查
self._check_weight_stats(weights)
return weights
def _check_weight_stats(self, weights):
for name, w in weights.items():
if torch.isnan(w).any() or torch.isinf(w).any():
raise SecurityError(f"invalid_weights: {name}")
if w.abs().max() > 1e6:
raise SecurityError(f"suspicious_weights: {name}")
2.3 来源验证
class SourceVerifier:
def __init__(self, trusted_registries):
self.trusted = trusted_registries
def verify(self, model_source):
# 1. URL域名白名单
if not self._trusted_domain(model_source.url):
return False, "untrusted_domain"
# 2. 仓库签名验证
if not model_source.repo_signature_valid:
return False, "repo_signature_invalid"
# 3. 发布者身份验证
if model_source.publisher not in self.trusted_publishers:
return False, "untrusted_publisher"
# 4. 模型卡完整性
if not self._verify_model_card(model_source):
return False, "model_card_incomplete"
return True, "verified"
三、依赖审计自动化
3.1 依赖树扫描
class DependencyAuditor:
def __init__(self, cve_db, poison_db):
self.cve_db = cve_db
self.poison_db = poison_db
def audit(self, project):
# 1. 解析依赖树
deps = self._parse_dependencies(project)
findings = []
# 2. CVE关联
for dep in deps:
cves = self.cve_db.query(dep.name, dep.version)
for cve in cves:
findings.append({
"type": "cve",
"dep": dep.name,
"version": dep.version,
"cve": cve.id,
"severity": cve.severity,
})
# 3. 投毒包检测
for dep in deps:
if self.poison_db.is_known_poisoned(dep.name, dep.version):
findings.append({
"type": "poisoned_package",
"dep": dep.name,
"version": dep.version,
})
# 4. typosquatting检测
for dep in deps:
if self._is_typosquatting(dep.name):
findings.append({
"type": "typosquatting",
"dep": dep.name,
})
return findings
def _is_typosquatting(self, name):
for trusted in self.trusted_packages:
if self._edit_distance(name, trusted) <= 2 and name != trusted:
return True
return False
3.2 自动化管线
class SupplyChainCI:
"""CI/CD集成的供应链安全管线"""
def run(self, project):
report = {}
# 1. AI-BOM生成与验证
bom = AIBOMGenerator(project).generate()
report["bom"] = self._verify_bom(bom)
# 2. 模型完整性
for model in bom.models:
report[f"model_{model.name}"] = self._verify_model(model)
# 3. 依赖审计
report["deps"] = DependencyAuditor().audit(project)
# 4. MCP服务器审计
report["mcp"] = MCPAuditor().audit(project)
# 5. 综合判定
report["verdict"] = "pass" if not self._has_critical(report) else "fail"
return report
四、MCP安全网关
4.1 MCP服务器安全审计
class MCPAuditor:
def __init__(self, known_servers):
self.known = known_servers
def audit(self, project):
servers = self._discover_mcp_servers(project)
findings = []
for server in servers:
# 1. 是否在已知白名单
if server.name not in self.known:
findings.append({"server": server.name, "issue": "unknown_server"})
# 2. 能力与登记是否一致
registered = self.known.get(server.name)
if registered and server.capabilities != registered.capabilities:
findings.append({
"server": server.name,
"issue": "capability_mismatch",
"extra": set(server.capabilities) - set(registered.capabilities),
})
# 3. 通信是否加密
if not server.tls_enabled:
findings.append({"server": server.name, "issue": "no_tls"})
# 4. 证书有效性
if not server.certificate_valid:
findings.append({"server": server.name, "issue": "invalid_cert"})
return findings
4.2 能力验证
class MCPCapabilityVerifier:
def __init__(self, registered_capabilities):
self.registered = registered_capabilities
def verify_call(self, server, tool, args):
# 1. 工具是否在登记能力内
if tool not in self.registered.get(server, {}):
return False, "tool_not_registered"
# 2. 参数是否在能力描述的schema内
schema = self.registered[server][tool]["args_schema"]
if not validate_against_schema(args, schema):
return False, "args_outside_schema"
# 3. 调用频率是否异常
if self._rate_exceeded(server, tool):
return False, "rate_limit_exceeded"
return True, "ok"
4.3 通信安全
class MCPSecureChannel:
def __init__(self, server, expected_fingerprint):
self.server = server
self.expected_fp = expected_fingerprint
def connect(self):
# 1. TLS连接
conn = tls_connect(self.server.url)
# 2. 证书指纹验证(防中间人)
actual_fp = cert_fingerprint(conn.peer_cert)
if actual_fp != self.expected_fp:
raise SecurityError("fingerprint_mismatch")
# 3. 能力协商
capabilities = conn.negotiate()
if not self._capabilities_safe(capabilities):
raise SecurityError("unsafe_capabilities")
return conn
五、实战案例
5.1 企业AI供应链安全方案
某企业构建内部AI平台,需对所有团队使用的模型与工具做供应链安全管控。
架构:
- 内部模型仓库:所有外部模型必须经安全扫描后镜像到内部仓库,禁止直接从HuggingFace等外部源拉取。
- AI-BOM强制:每个AI应用必须提交AI-BOM,CI/CD管线验证BOM完整性与所有组件签名。
- MCP网关:所有MCP服务器必须预先登记,运行时通过统一网关访问,禁止Agent直连。
- 持续监控:已部署模型定期重扫,新发现的CVE或投毒包触发告警与重评估。
实施经验:
- 内部镜像仓库初期被开发团队抵触(增加流程),后通过”自动镜像+透明代理”降低使用成本。
- AI-BOM生成工具集成到训练管线,自动记录训练数据、代码版本、超参数,避免人工填报遗漏。
- 发现过一起”开发者从HuggingFace拉取未审计模型做快速原型”的事件,后通过出网防火墙规则强制所有模型加载走内部仓库。
5.2 HuggingFace模型安全扫描管线
对从HuggingFace引入的模型做安全扫描的完整管线:
class HuggingFaceSecurityPipeline:
def scan(self, model_repo):
report = {}
# 1. 元数据检查
report["metadata"] = self._check_metadata(model_repo)
# 2. 模型卡检查
report["modelcard"] = self._check_modelcard(model_repo)
# 3. 权重安全加载
report["weights"] = self._scan_weights(model_repo)
# 4. 后门检测(见中级篇06)
report["backdoor"] = BackdoorDetectionFramework().full_scan(model)
# 5. 行为探针测试
report["behavior"] = self._behavior_probes(model)
# 6. 依赖审计
report["deps"] = DependencyAuditor().audit(model_repo)
return report
def _scan_weights(self, model_repo):
# 强制用safetensors加载
try:
weights = SafeModelLoader().load(model_repo.safetensors_path)
return {"status": "safe_loaded", "hash": hash(weights)}
except Exception as e:
return {"status": "error", "reason": str(e)}
发现的典型问题:
- 部分模型仅提供pickle格式权重,存在反序列化风险,要求作者提供safetensors版本。
- 模型卡缺失或不完整是常见问题,缺乏训练数据来源与评估结果。
- 少数模型在后门检测中触发Neural Cleanse告警,需人工复核。
六、AI供应链安全的理论基础
6.1 供应链攻击的攻击面
AI系统的供应链比传统软件更复杂:
数据源 → 数据处理 → 模型训练 → 模型分发 → 部署 → 运行
↑ ↑ ↑ ↑ ↑ ↑
注入点 注入点 注入点 注入点 注入点 注入点
每个环节都是潜在攻击面:
- 数据源:恶意数据注入、数据源劫持。
- 数据处理:处理代码漏洞、预处理投毒。
- 模型训练:训练代码漏洞、依赖投毒。
- 模型分发:模型仓库劫持、权重篡改。
- 部署:部署配置漏洞、环境依赖投毒。
- 运行:运行时依赖漏洞、模型更新劫持。
6.2 AI-BOM的理论基础
物料清单的完整性:
AI-BOM的完整性定义为:BOM包含AI系统所有组件及其依赖关系的完整描述。
定理(BOM完整性vs可追踪性):完整BOM是可追踪性的必要条件——若无完整BOM,无法追踪某组件的来源与依赖。
BOM的图结构:
BOM = (V, E)
`
免责声明:
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。
任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我。
本文转载自:安全分析与研究 pandazhengzheng pandazhengzheng《AI供应链安全工程》
版权声明
本站仅做备份收录,仅供研究与教学参考之用。
读者将信息用于其他用途的,全部法律及连带责任由读者自行承担,本站不承担任何责任。











评论