AI供应链安全工程

admin 2026-09-14 04:54:15 网络安全文章 来源:ZONE.CI 全球网 0 阅读模式

文章总结: 本文介绍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:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;self._edit_distance(name, trusted) <=&nbsp;2&nbsp;and&nbsp;name != trusted:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;True
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;False

3.2 自动化管线

class&nbsp;SupplyChainCI:
&nbsp; &nbsp;&nbsp;"""CI/CD集成的供应链安全管线"""
&nbsp; &nbsp;&nbsp;def&nbsp;run(self, project):
&nbsp; &nbsp; &nbsp; &nbsp; report = {}
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 1. AI-BOM生成与验证
&nbsp; &nbsp; &nbsp; &nbsp; bom = AIBOMGenerator(project).generate()
&nbsp; &nbsp; &nbsp; &nbsp; report["bom"] = self._verify_bom(bom)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 2. 模型完整性
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;for&nbsp;model&nbsp;in&nbsp;bom.models:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; report[f"model_{model.name}"] = self._verify_model(model)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 3. 依赖审计
&nbsp; &nbsp; &nbsp; &nbsp; report["deps"] = DependencyAuditor().audit(project)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 4. MCP服务器审计
&nbsp; &nbsp; &nbsp; &nbsp; report["mcp"] = MCPAuditor().audit(project)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 5. 综合判定
&nbsp; &nbsp; &nbsp; &nbsp; report["verdict"] =&nbsp;"pass"&nbsp;if&nbsp;not&nbsp;self._has_critical(report)&nbsp;else&nbsp;"fail"
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;report

四、MCP安全网关

4.1 MCP服务器安全审计

class&nbsp;MCPAuditor:
&nbsp; &nbsp;&nbsp;def&nbsp;__init__(self, known_servers):
&nbsp; &nbsp; &nbsp; &nbsp; self.known = known_servers

&nbsp; &nbsp;&nbsp;def&nbsp;audit(self, project):
&nbsp; &nbsp; &nbsp; &nbsp; servers = self._discover_mcp_servers(project)
&nbsp; &nbsp; &nbsp; &nbsp; findings = []
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;for&nbsp;server&nbsp;in&nbsp;servers:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 1. 是否在已知白名单
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;server.name&nbsp;not&nbsp;in&nbsp;self.known:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; findings.append({"server": server.name,&nbsp;"issue":&nbsp;"unknown_server"})
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 2. 能力与登记是否一致
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; registered = self.known.get(server.name)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;registered&nbsp;and&nbsp;server.capabilities != registered.capabilities:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; findings.append({
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"server": server.name,
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"issue":&nbsp;"capability_mismatch",
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"extra": set(server.capabilities) - set(registered.capabilities),
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; })
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 3. 通信是否加密
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;not&nbsp;server.tls_enabled:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; findings.append({"server": server.name,&nbsp;"issue":&nbsp;"no_tls"})
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 4. 证书有效性
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;not&nbsp;server.certificate_valid:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; findings.append({"server": server.name,&nbsp;"issue":&nbsp;"invalid_cert"})
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;findings

4.2 能力验证

class&nbsp;MCPCapabilityVerifier:
&nbsp; &nbsp;&nbsp;def&nbsp;__init__(self, registered_capabilities):
&nbsp; &nbsp; &nbsp; &nbsp; self.registered = registered_capabilities

&nbsp; &nbsp;&nbsp;def&nbsp;verify_call(self, server, tool, args):
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 1. 工具是否在登记能力内
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;tool&nbsp;not&nbsp;in&nbsp;self.registered.get(server, {}):
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;False,&nbsp;"tool_not_registered"
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 2. 参数是否在能力描述的schema内
&nbsp; &nbsp; &nbsp; &nbsp; schema = self.registered[server][tool]["args_schema"]
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;not&nbsp;validate_against_schema(args, schema):
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;False,&nbsp;"args_outside_schema"
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 3. 调用频率是否异常
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;self._rate_exceeded(server, tool):
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;False,&nbsp;"rate_limit_exceeded"
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;True,&nbsp;"ok"

4.3 通信安全

class&nbsp;MCPSecureChannel:
&nbsp; &nbsp;&nbsp;def&nbsp;__init__(self, server, expected_fingerprint):
&nbsp; &nbsp; &nbsp; &nbsp; self.server = server
&nbsp; &nbsp; &nbsp; &nbsp; self.expected_fp = expected_fingerprint

&nbsp; &nbsp;&nbsp;def&nbsp;connect(self):
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 1. TLS连接
&nbsp; &nbsp; &nbsp; &nbsp; conn = tls_connect(self.server.url)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 2. 证书指纹验证(防中间人)
&nbsp; &nbsp; &nbsp; &nbsp; actual_fp = cert_fingerprint(conn.peer_cert)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;actual_fp != self.expected_fp:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;raise&nbsp;SecurityError("fingerprint_mismatch")
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 3. 能力协商
&nbsp; &nbsp; &nbsp; &nbsp; capabilities = conn.negotiate()
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;not&nbsp;self._capabilities_safe(capabilities):
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;raise&nbsp;SecurityError("unsafe_capabilities")
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;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&nbsp;HuggingFaceSecurityPipeline:
&nbsp; &nbsp;&nbsp;def&nbsp;scan(self, model_repo):
&nbsp; &nbsp; &nbsp; &nbsp; report = {}
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 1. 元数据检查
&nbsp; &nbsp; &nbsp; &nbsp; report["metadata"] = self._check_metadata(model_repo)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 2. 模型卡检查
&nbsp; &nbsp; &nbsp; &nbsp; report["modelcard"] = self._check_modelcard(model_repo)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 3. 权重安全加载
&nbsp; &nbsp; &nbsp; &nbsp; report["weights"] = self._scan_weights(model_repo)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 4. 后门检测(见中级篇06)
&nbsp; &nbsp; &nbsp; &nbsp; report["backdoor"] = BackdoorDetectionFramework().full_scan(model)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 5. 行为探针测试
&nbsp; &nbsp; &nbsp; &nbsp; report["behavior"] = self._behavior_probes(model)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 6. 依赖审计
&nbsp; &nbsp; &nbsp; &nbsp; report["deps"] = DependencyAuditor().audit(model_repo)
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;report

&nbsp; &nbsp;&nbsp;def&nbsp;_scan_weights(self, model_repo):
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 强制用safetensors加载
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;try:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; weights = SafeModelLoader().load(model_repo.safetensors_path)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;{"status":&nbsp;"safe_loaded",&nbsp;"hash": hash(weights)}
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;{"status":&nbsp;"error",&nbsp;"reason": str(e)}

发现的典型问题

  • 部分模型仅提供pickle格式权重,存在反序列化风险,要求作者提供safetensors版本。
  • 模型卡缺失或不完整是常见问题,缺乏训练数据来源与评估结果。
  • 少数模型在后门检测中触发Neural Cleanse告警,需人工复核。

六、AI供应链安全的理论基础

6.1 供应链攻击的攻击面

AI系统的供应链比传统软件更复杂:

数据源 → 数据处理 → 模型训练 → 模型分发 → 部署 → 运行
&nbsp; &nbsp;↑ &nbsp; &nbsp; &nbsp; &nbsp; ↑ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;↑ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;↑ &nbsp; &nbsp; &nbsp; &nbsp;↑ &nbsp; &nbsp; &nbsp; ↑
&nbsp;注入点 &nbsp; &nbsp;注入点 &nbsp; &nbsp; &nbsp;注入点 &nbsp; &nbsp; 注入点 &nbsp; &nbsp;注入点 &nbsp; 注入点

每个环节都是潜在攻击面:

  • 数据源:恶意数据注入、数据源劫持。
  • 数据处理:处理代码漏洞、预处理投毒。
  • 模型训练:训练代码漏洞、依赖投毒。
  • 模型分发:模型仓库劫持、权重篡改。
  • 部署:部署配置漏洞、环境依赖投毒。
  • 运行:运行时依赖漏洞、模型更新劫持。

6.2 AI-BOM的理论基础

物料清单的完整性

AI-BOM的完整性定义为:BOM包含AI系统所有组件及其依赖关系的完整描述。

定理(BOM完整性vs可追踪性):完整BOM是可追踪性的必要条件——若无完整BOM,无法追踪某组件的来源与依赖。

BOM的图结构

BOM = (V, E)

`


免责声明:

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。

任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我

本文转载自:安全分析与研究 pandazhengzheng pandazhengzheng《AI供应链安全工程》

AI供应链安全工程 网络安全文章

AI供应链安全工程

文章总结: 本文介绍AI供应链安全工程实践,提出AI-BOM物料清单格式,涵盖模型权重、训练数据等组件,实现全链路追踪与来源验证,并通过模型签名、safeten
评论:0   参与:  0