开发者日记

外观

模型供应链投毒扫描：从裸跑到可信校验 ​

问题背景：模型文件不是普通制品 ​

大模型项目通常会同时引入三类外部输入：Hugging Face 等平台上的模型权重、社区封装的推理插件、pip/npm 依赖包。它们共同构成了 AI 供应链。传统应用扫描主要关注源码依赖，但模型权重可能携带恶意 pickle、异常配置路径、陌生远程调用地址；插件依赖则可能通过相似包名、版本劫持、安装脚本植入后门。

一次典型攻击并不需要修改模型推理结果。攻击者只需要在加载权重时触发反序列化，或在插件初始化阶段读取环境变量中的 API Key、向量数据库地址和云厂商凭证。因此，模型供应链安全的目标不是“相信某个热门模型”，而是把权重、配置、代码依赖全部纳入可验证、可阻断、可审计的发布流程。

方案对比：只看病毒库为什么不够 ​

常见做法可以分为四层，成熟度依次提升：

  1. 人工审查：检查作者、下载量、许可证和最近提交。优点是零成本，缺点是无法识别二进制权重中的恶意内容，也容易被高仿账号绕过。
  2. 依赖漏洞扫描：使用 Safety、pip-audit、npm audit 等工具发现已知 CVE。它适合处理依赖风险，但对模型文件、pickle gadget、可疑配置基本无感知。
  3. 模型扫描与反序列化检查：使用 model-scan 或等价扫描器检查 H5、Pickle、PyTorch、NumPy 等载荷中的危险调用，识别 eval、exec、子进程、网络访问等行为。
  4. 哈希、签名与 SBOM 门禁：对权重文件做 SHA-256 锁定，对来源做签名验证，生成依赖清单，并在 CI 中强制“未扫描、未校验、不得发布”。

推荐组合是：model-scan 负责发现模型制品中的危险载荷，依赖扫描器负责发现插件包风险，哈希与签名负责确认来源和完整性，最小权限负责降低失陷后的爆炸半径。

开源权重safetensors/bin模型扫描model-scan哈希签名SHA-256/Sigstore私有制品库只读缓存插件依赖SBOM/审计CI 门禁高危即失败

实操步骤：构建可运行的模型准入脚本 ​

下面示例保存为 model_admission.py。它会完成四件事：扫描模型目录中的危险载荷、校验权重 SHA-256、检查插件依赖是否包含安装脚本或高危版本、生成机器可读的准入结果。示例优先演示安全流程，实际项目可把 model_scan 替换为团队统一安装的扫描器 CLI 或 SDK。

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""模型供应链准入检查：权重扫描 + 哈希校验 + 插件依赖检查。"""

import hashlib
import json
import subprocess
import sys
from pathlib import Path

# 只允许这些权重后缀进入加载流程
ALLOWED_SUFFIXES = {".safetensors", ".bin", ".pt", ".pth", ".h5", ".onnx"}

# 危险调用示例：真实规则应由 model-scan 官方规则集维护
DANGEROUS_TOKENS = [
    b"__reduce__",
    b"subprocess",
    b"os.system",
    b"eval(",
    b"exec(",
    b"requests.post",
    b"socket.create_connection",
]


def sha256_file(path: Path, block_size: int = 1024 * 1024) -> str:
    """以流式方式计算大文件哈希，避免一次性读取几十 GB 权重。"""
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(block_size), b""):
            digest.update(block)
    return digest.hexdigest()


def load_hash_manifest(manifest_path: Path) -> dict:
    """清单格式：{"pytorch_model.bin": "sha256:..."}"""
    return json.loads(manifest_path.read_text(encoding="utf-8"))


def verify_hashes(model_dir: Path, manifest: dict) -> list:
    findings = []
    for file_path in model_dir.rglob("*"):
        if not file_path.is_file():
            continue
        if file_path.suffix.lower() not in ALLOWED_SUFFIXES:
            continue
        rel = str(file_path.relative_to(model_dir))
        expected = manifest.get(rel)
        if not expected:
            findings.append({"level": "HIGH", "type": "missing_hash", "file": rel})
            continue
        actual = "sha256:" + sha256_file(file_path)
        if actual != expected:
            findings.append({
                "level": "CRITICAL",
                "type": "hash_mismatch",
                "file": rel,
                "expected": expected,
                "actual": actual,
            })
    return findings


def scan_model_files(model_dir: Path) -> list:
    """轻量静态检查；生产环境应同时执行官方 model-scan。"""
    findings = []
    for file_path in model_dir.rglob("*"):
        if not file_path.is_file() or file_path.suffix.lower() not in ALLOWED_SUFFIXES:
            continue
        rel = str(file_path.relative_to(model_dir))

        # 大文件只抽样检查头尾和若干窗口，适合作为快速预检
        with file_path.open("rb") as f:
            head = f.read(2 * 1024 * 1024)
            f.seek(max(0, file_path.stat().st_size - 2 * 1024 * 1024))
            tail = f.read(2 * 1024 * 1024)

        blob = head + tail
        for token in DANGEROUS_TOKENS:
            if token in blob:
                findings.append({
                    "level": "CRITICAL",
                    "type": "dangerous_payload",
                    "file": rel,
                    "indicator": token.decode(errors="ignore"),
                })
    return findings


def run_model_scan(model_dir: Path) -> list:
    """调用外部 model-scan；若 CI 镜像未安装则直接失败，而不是静默跳过。"""
    cmd = [sys.executable, "-m", "model_scan", "--path", str(model_dir), "--output", "json"]
    result = subprocess.run(cmd, text=True, capture_output=True)
    if result.returncode != 0:
        return [{"level": "CRITICAL", "type": "scanner_error", "detail": result.stderr[:500]}]
    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        return [{"level": "CRITICAL", "type": "invalid_scan_report"}]
    return data.get("findings", [])


def scan_plugin_manifest(plugin_file: Path) -> list:
    """检查插件包元数据，拦截安装脚本、过度权限和未固定版本。"""
    findings = []
    text = plugin_file.read_text(encoding="utf-8", errors="ignore")

    if "setup.py" in text or "postinstall" in text:
        findings.append({"level": "HIGH", "type": "install_hook", "file": plugin_file.name})

    # 依赖必须锁定到精确版本或内部制品库哈希
    for line in text.splitlines():
        clean = line.strip()
        if clean and not clean.startswith("#") and (">=" in clean or "~=" in clean or "*" in clean):
            findings.append({"level": "MEDIUM", "type": "floating_version", "value": clean})
    return findings


def main():
    if len(sys.argv) != 3:
        print("用法: python model_admission.py <模型目录> <哈希清单>")
        sys.exit(2)

    model_dir = Path(sys.argv[1])
    manifest = load_hash_manifest(Path(sys.argv[2]))

    findings = []
    findings.extend(verify_hashes(model_dir, manifest))
    findings.extend(scan_model_files(model_dir))
    findings.extend(run_model_scan(model_dir))

    for name in ["requirements.txt", "package.json", "plugin.json"]:
        candidate = model_dir / name
        if candidate.exists():
            findings.extend(scan_plugin_manifest(candidate))

    report = {"model": str(model_dir), "findings": findings}
    print(json.dumps(report, ensure_ascii=False, indent=2))

    blocked = {"CRITICAL", "HIGH"}
    sys.exit(1 if any(item.get("level") in blocked for item in findings) else 0)


if __name__ == "__main__":
    main()

配套的 sha256-manifest.json 示例如下：

json
{
  "model.safetensors": "sha256:替换为官方公布或内部入库时计算的值",
  "tokenizer.json": "sha256:建议配置文件也纳入校验"
}

CI 中可以增加如下门禁：

bash
python model_admission.py ./models/embedding-v1 ./models/embedding-v1/sha256-manifest.json
pip-audit -r ./plugins/requirements.txt --strict

插件依赖：用最小权限限制失陷影响 ​

插件是另一条高频攻击路径。建议所有模型插件运行在独立虚拟环境或容器中，只授予完成任务所需的最小权限：只读模型目录、只允许访问指定推理端点、禁止挂载云主机元数据服务、默认禁止访问公网。若插件需要检索知识库，也应通过后端服务签发短期 Token，而不是把长期数据库密码放进环境变量。

对于智能体插件，还要关注提示注入导致的越权操作。即使依赖没有木马，被检索文档或网页中的恶意指令也可能诱导插件调用删除、外发、支付等工具。工具层应把读写、外发、管理类动作分级；高危动作必须二次确认，并且不能把“模型输出的自然语言”当作授权依据。

避坑总结：三个高频错误与修正方法 ​

  1. 只扫描 .py 文件，忽略 .bin、.pt、.pth。现象是代码审计通过，但加载模型时执行了 pickle 链。修正方法是把全部模型制品纳入 model-scan，优先使用 safetensors；历史 pickle 权重必须在隔离环境中扫描、复现哈希并重新入库。
  2. 只记录下载地址，不锁定哈希和签名。现象是同一模型版本第二天拉取内容发生变化，排查发现镜像缓存或发布者覆盖文件。修正方法是入库时生成 SHA-256 清单，发布流程只从私有只读制品库拉取；关键模型进一步使用签名验证。
  3. 依赖版本写成范围约束。例如 transformers>=4.40.0 在今天通过，下周可能解析到被投毒或引入不兼容安装脚本的新版本。修正方法是锁定精确版本、保留内部 wheel 缓存，并生成 SBOM；确需升级时重新扫描和回归测试。
  4. 扫描失败被 CI 忽略。常见脚本写成 model-scan ... || true，最终报告显示绿色流水线。修正方法是把扫描器不存在、规则更新失败、报告格式异常都视为准入失败，并把 JSON 报告归档到审计系统。
  5. 插件拥有过大环境变量和网络权限。一旦插件被提示注入或依赖投毒，攻击者可读取 API Key 并访问元数据服务。修正方法是使用独立运行身份、最小网络出口、短期凭证，并对高危工具调用做策略校验。

模型供应链安全不是一次性扫描，而是准入、缓存、加载、运行、审计的闭环。把不可信内容挡在私有制品库之外，把插件权限收敛到最小，再用 CI 门禁强制执行，才能让开源模型从“裸跑下载”变成“可验证、可追踪、可阻断”的工程资产。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。