【雕虫大技】Agent 动态 Skill 供应链安全加固（一）：消费侧三层防线实战-阿里云开发者社区

一、背景

最近在开发一个Agent系统的时候有这么一个场景：系统使用的是 AgentScope 2.0框架直接搭建，它的核心能力之一是 Skill Registry：skill（一组 SKILL.md + Python 脚本的集合）可以从 Nacos 动态下载到本地，物化出脚本，再用 Python 解释器执行。

这条链路本身很优雅，但是背后可能存在着一些漏洞：下载的 skill 是外部输入，而它最终会被执行。攻击面至少有三个：

攻击面	威胁	后果
篡改 zip	中间人 / 发布源被污染，替换 zip 内脚本	执行攻击者代码
解压路径穿越（Zip Slip）	zip 内 `../` 条目逃出目标目录	覆盖本地任意文件
恶意 import	脚本 `import socket` / `subprocess`	反弹 shell、横向渗透

这三个攻击面的共同点是：skill 在进入执行阶段之前，几乎没有任何校验。这正是供应链加固要解决的问题：下载可以很自由，但必须把握住"下载 → 校验 → 执行"这之间的关口。

二、整体设计：发布侧治理管不了运行时

Nacos 3.2 的 Skill Registry 本质是发布侧的供应链治理，它管生命周期、审核流水线、PUBLIC/PRIVATE 可见性，默认 pipeline 关闭、还有 Force Publish 绕过，没有运行时沙箱。所以它解决的是"谁允许发布"，但解决不了"下载到我本地的是不是被篡改过"。

结论：发布侧治理和消费侧防线两者并存，不能互相替代。本次做的是消费侧的三层防线：

下载链路
 ┌─────────────────────────────────────────────────┐
 │ ① 完整性校验 (SkillContentVerifier)              │
 │    zip 内嵌清单 + HMAC-SHA256 → 防篡改           │
 ├─────────────────────────────────────────────────┤
 │ ② 静态 import 扫描 (PythonImportScanner)         │
 │    黑名单 + SKILL.md permissions → 防恶意代码    │
 ├─────────────────────────────────────────────────┤
 │ ③ 路径防逃逸 (SkillPathGuard)                   │
 │    normalize 后必须 startsWith 根目录 → 防穿越   │
 └─────────────────────────────────────────────────┘
            ↓
         执行（本地 Python 进程）

三层职责分明：① 保证"下载的东西没被换过"，② 保证"写脚本的人没埋雷"，③ 保证"解压和寻址不出界"。执行隔离（沙箱）和输出校验留到阶段二和阶段三。

三、第一层：zip 完整性校验（防篡改）

问题：skill 是打成 zip 包从 Nacos 下载的，中间可能被篡改，假如攻击者替换 zip 里的脚本，就能让服务端执行恶意代码。这一层要做的是在 zip 落盘之前验证"包里的内容确实没被改过"，一旦发现对不上就拒绝写入。

具体做法分三步：先把每个文件的哈希（sha256）写进 zip 内嵌清单，再用密钥给清单签名（防连清单一起被改），最后在下载时逐文件比对。下面的小节逐个拆解这三个决定。

3.1 为什么把校验清单内嵌进 zip

方案选择时纠结过：清单放 zip 外，还是 zip 里？放外面（比如 Nacos 的 config 配套字段）的问题是清单和 zip 分离，传输路径不一致，篡改 zip 时可以只不动清单，校验等于没做。最终决定内嵌一个 .bank-checksum.sha256 进 zip 里，用 . 前缀尽量不干扰 skill 内容。

3.2 为什么光有 sha256 还不够

如果只用普通的“文件名+SHA256”做校验，就像把锁的钥匙直接贴在了门上，攻击者换掉脚本后，只要自己重新算个哈希值贴上去，就能骗过检查。因此，清单必须改用 HMAC-SHA256 来签名，这相当于把“公开计算”变成了“盖公章”。我们需要在发布方和消费方之间私下约定一个只有他们俩才知道的暗号（即密钥 SKILL_HMAC_SECRET），攻击者因为没有这个暗号，就算改了脚本，也盖不出合法的“公章”，校验就会立刻失败。

密钥为空时降级为"仅 sha256 自校验"，方便本地开发（本地 classpath skill 无清单则直接跳过完整性校验，只做路径防逃逸）。

3.3 密钥从哪来，信任边界在哪

讲完"盖章"，自然要问：攻击者不知道密钥，那我们是怎么知道的？

答案是：密钥不是哪一方单独"发明"的，而是发布方和消费方事先约好的一个共享秘密：

发布方（skill 上传者）                      消费方（你的后端）
─────────────────────                      ─────────────────────
知道密钥 SKILL_HMAC_SECRET                 知道同一个密钥 SKILL_HMAC_SECRET
打包 zip 时用密钥算 HMAC 写进清单   ────→   下载 zip 后用同一个密钥重算 HMAC，
                                           䘥和清单里的 signature= 比对

攻击者夹在中间：不知道密钥，改脚本后就算不出合法签名，校验立刻失败。这就是整个方案的信任核心——密钥保密，两边共享。

那共享密钥具体从哪来？通常两条路：

配置注入：通过环境变量（如 SKILL_HMAC_SECRET）或部署配置中心下发。发布方打包脚本时用同一个值，消费方部署时配同一个值。密钥只存在运行时内存里，不进代码仓库。
密钥托管：更正式的做法是从 KMS / Vault / 配置中心读取，两边各自从受信源拿同一份密钥。

但必须诚实交代这个方案的信任边界，它只防一种攻击，不防另外两种：

场景	HMAC 能否防住	原因
传输/存储中被中间人篡改 zip	✅ 能	攻击者没密钥，改包后签不出合法签名
发布方本身是恶意的（拿到上传权限）	❌ 不能	他能用真密钥签出"合法"清单
密钥泄露	❌ 不能	整个方案失效，攻击者可伪造任意合法 zip

所以 HMAC 精确只做一件事：在密钥保密的前提下，防住中间人偷偷改包。防恶意发布方要靠发布侧治理（审核/权限），防密钥泄露要靠密钥管理——它们和 HMAC 是互补关系，不是替代。

3.4 实现要点

// 校验通过时才算真正成功，且比较必须是 constant-time，防时序侧信道
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] actual = digest.digest(Files.readAllBytes(file));
if (!MessageDigest.isEqual(expected, actual)) {
   
    throw new SecurityException("文件哈希不匹配: " + entryName);
}

两个容易踩的坑：

Nacos 包裹层——Nacos 下载的 zip 可能外层包一层公共根目录（如 skill-x/scripts/...），和上传时平铺的结构对不上。统一用 parseZip() 剥离公共根目录，下载解压和校验共用同一套逻辑，避免两套路径处理漂移。
校验时机——下载后、解压前先 verifyZip()，防止恶意 zip 在解压阶段就干坏事；解压后再 verifyDir() 兜底（此时还能顺带核对"清单列了但目录里没有"的缺文件情况）。

密钥在配置里的落地就一行（对应 3.3 说的环境变量 SKILL_HMAC_SECRET），用 Spring 时写进 yml 即可：

app:
  ai:
    skill:
      verify:
        hmac-secret: ${
   SKILL_HMAC_SECRET:}   # 留空=仅 sha256 自校验，不验签名
        fail-on-violation: true              # 扫描到危险 import 是否拒绝

四、第二层：Python 静态 import 扫描（防恶意代码）

问题：完整性校验保证了"脚本没被中途改过"，但保证不了脚本本身是好人写的：一个包通过的 zip 里可能自带恶意代码，比如 import socket 连出去、subprocess 起进程。这一层的目标是在脚本执行之前静态扫描它的 import 语句，发现危险模块就拦截或告警。

做法分两步：先是确定"拦什么"，我们需要维护一个危险模块黑名单，只拦能脱离解释器干坏事的 fork/exec 型模块；再是"怎么不误伤"，允许 skill 在 SKILL.md 里声明权限来放行确实需要这些能力的脚本。

4.1 黑名单选型：只拦 fork/exec 型模块

用正则解析 import X / from X import Y 语句，命中黑名单即违规。黑名单里放的是能脱离解释器进程干坏事的模块：

ctypes, multiprocessing, socket, subprocess,
urllib, http, ftplib, telnetlib, ftputil, paramiko, shutil

这里有个重要的取舍：不包含 importlib/exec/eval/webbrowser。这几个函数无法通过import来找到，如果需要排查则需要将所有的文本都遍历一遍。它们是"间接"能力，比如exec("import socket") 理论上能绕过，但直接拦掉会误伤大量合法 skill（动态加载很常见），而且真要在黑名单里追 exec 是追不完的。静态扫描的定位是低成本拦截明显恶意 + 逼出显式权限声明，不是和攻击者赛跑。这也是后面接沙箱的原因。

4.2 权限放行机制：SKILL.md frontmatter permissions

纯黑名单的毛病是"一刀切"：本地 skill-creator 大量使用 subprocess/http.server，会被误杀。于是引入声明式权限，即SKILL.md frontmatter 声明该 skill 需要什么权限：

---
name: skill-creator
permissions: [network, subprocess]
---

解析出 permissions 后，network 放行网络类黑名单项，subprocess/process 放行子进程类。这就是白名单式放行优于全黑名单的实践：默认拒绝，显式授权，审计清楚谁要了什么权限。

配置项 forbidden-imports 可覆盖默认黑名单，fail-on-violation 决定命中是拦截执行还是仅告警放行（灰度期用告警观察误报，稳定后切拦截）。

五、第三层：路径防逃逸（防 Zip Slip / 越界写）

问题：skill 里有大量"相对路径"，如解压时 zip 条目名是路径、物化时资源 key 是路径、执行时脚本名也是路径。攻击者可以在这些路径里塞 ../，让文件写到 skill 目录之外（覆盖服务器上任意文件），或从外部读文件。这一层要做的是所有路径解析都经过一道"越界检查"，凡是会逃出基础目录的一律拒绝。

5.1 `resolve` 加 `startsWith` 为什么会被绕过

这是最隐蔽也最经典的洞。Java 的 base.resolve("../evil.py") 返回的是 base/../evil.py，这样就会导致开头是 base，Path.startsWith(base) 做词法前缀比较会误判为安全，但 normalize 之后还原成 ../evil.py，读写时就直接飞出 base 目录（Zip Slip：把脚本写到 skill 目录之外任意位置，或从外部读文件）。我们最初把 skill 资源物化（写到本地磁盘）的代码就是这么写的，等于埋了个雷。

5.2 修复方式：先 normalize 再判断，且必须统一落点在 base 内

修法很简单，但顺序有讲究：

public static Path safeResolve(Path base, String relative) {
   
    Path candidate = base.resolve(relative).normalize();
    if (!candidate.startsWith(base.normalize())) {
   
        throw new IllegalArgumentException("非法路径逃逸: " + relative);
    }
    return candidate;
}

知识点：

必须先 normalize() 再 startsWith()。startsWith 是词法前缀比较，base/../outside 这种字符串形式上不带 ../，但 normalize 之后会还原成 ../outside，就能被正确拦截。
反斜杠（Windows ..\）和正斜杠都要防，Java Path 在同一平台内会统一分隔符，测试里两种都覆盖。
绝对路径（C:/windows/system32/cmd.exe）也要拒绝——它根本不落在 base 下。

这个 guard 用在三个地方：skill 脚本物化（下载解压时）、执行前的脚本路径解析、zip 解压时的目标路径。

六、落地与验证

前三章讲了三道防线各自的设计，这一节把它们收拢成"三个可运行的组件 + 三个接入点 + 测试结果"，让前文的设计落到能跑、可验证的形态。

6.1 组件总览

本次加固落地为三个核心组件，各管一道防线：

组件	对应防线	核心职责
完整性校验器	第一层	生成内嵌清单、HMAC 签名；下载后逐文件校验，失败即拒绝落盘
静态扫描器	第二层	解析 `.py` 的 import 语句、比对黑名单、读取 SKILL.md 权限放行
路径守卫	第三层	所有路径解析统一走 `normalize + startsWith` 校验，拦截逃逸

6.2 防线落点：三道防线分别嵌入生命周期的哪个环节

这三道防线分别嵌入 skill 从下载到执行的完整生命周期里：

下载环节：先做完整性校验（验 HMAC 签名 + 逐文件比对哈希），通过才落盘；
物化解压环节：落盘前做路径防逃逸，防止 zip 内 ../ 条目覆盖外部文件；
执行环节：定位到脚本后、拉起 Python 进程前，统一过一遍静态 import 扫描。

七、复用指南

如果你想把这三道防线移植到自己的项目，下面是照着做的完整清单：组件依赖关系、可直接抄的代码、需要补全的骨架、接入点，以及不用 Spring 时的用法。

7.1 组件依赖关系

SkillPathGuard（静态工具类，无任何依赖）
   ↑ 被下面两个组件内部调用（做路径防逃逸）

PythonImportScanner（静态扫描器）
   依赖：JDK + slf4j（仅做日志，可换成任意日志框架）

SkillContentVerifier（完整性校验器）
   依赖：JDK（javax.crypto + java.util.zip）+ slf4j

三个类唯一的第三方依赖是 slf4j（日志），并且只出现在日志语句里——不想要的话直接删掉 log 语句即可。它们不依赖 Spring——Spring 只负责把配置读进来再传给构造器，拆掉 Spring 一样能用（见 7.5）。

7.2 路径守卫 SkillPathGuard

无任何依赖，可直接复制：

import java.nio.file.Path;

/**
 * 路径防逃逸守卫：所有基于用户可控相对路径解析目标路径的地方，
 * 必须经过 safeResolve 校验，禁止通过 ../ 逃逸出基础目录。
 */
public final class SkillPathGuard {
   
    private SkillPathGuard() {
   }

    /**
     * 在 base 下安全解析 relPath，防止路径穿越。
     * @throws IllegalArgumentException 路径为绝对路径、或解析后逃逸出 base
     */
    public static Path safeResolve(Path base, String relPath) {
   
        if (relPath == null || relPath.isBlank()) {
   
            throw new IllegalArgumentException("相对路径不能为空");
        }
        // 统一转为绝对路径并归一化，保证 startsWith 比较的基准一致
        Path baseNorm = base.toAbsolutePath().normalize();
        // Windows 反斜杠（..\）也归一化为分隔符，防止被当普通文件名绕过
        String normalizedRel = relPath.replace('\\', '/');
        Path candidate = baseNorm.resolve(normalizedRel).normalize();
        // 必须先 normalize 再 startsWith：base/../evil 词法前缀是 base 会误判，
        // normalize 还原成 ../evil 后才能真正识别逃逸
        if (!candidate.startsWith(baseNorm)) {
   
            throw new IllegalArgumentException("非法路径穿越: " + relPath);
        }
        return candidate;
    }

    /** 判断 target 是否位于 base 内（均先转绝对路径并 normalize） */
    public static boolean isWithin(Path base, Path target) {
   
        Path baseNorm = base.toAbsolutePath().normalize();
        Path targetNorm = target.toAbsolutePath().normalize();
        return targetNorm.startsWith(baseNorm);
    }

    /** 检查 target 是否在 base 内，不在则抛异常 */
    public static void requireWithin(Path base, Path target) {
   
        if (!isWithin(base, target)) {
   
            throw new IllegalArgumentException("非法路径逃逸: " + target);
        }
    }
}

7.3 完整代码：静态扫描器与完整性校验器

7.3.1 静态扫描器 PythonImportScanner

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Python 脚本静态 import 扫描器。
 *
 * 解析 .py 文件中的顶层 import 语句，与危险模块黑名单比对。
 * 若 SKILL.md frontmatter 声明了 permissions（如 network、subprocess），
 * 则对应的黑名单项放行。
 */
public class PythonImportScanner {
   

    private static final Logger log = LoggerFactory.getLogger(PythonImportScanner.class);

    /** 默认危险模块黑名单（可按模块名精确匹配，也可按前缀匹配） */
    public static final List<String> DEFAULT_FORBIDDEN = List.of(
            "ctypes", "multiprocessing", "socket", "subprocess",
            "urllib", "http", "ftplib", "telnetlib", "ftputil", "paramiko", "shutil"
    );

    // 逐行匹配 "import X" / "from X import Y"，捕获被 import 的顶层模块名 X
    private static final Pattern IMPORT_STMT =
            Pattern.compile("^\\s*(?:import|from)\\s+([A-Za-z_][A-Za-z0-9_.]*)(?:\\s+import)?", Pattern.MULTILINE);

    // 匹配 SKILL.md 顶部的 YAML frontmatter（--- 包裹的段落），用于读取 permissions 声明
    private static final Pattern FRONTMATTER = Pattern.compile("^---\\s*\\R(.*?)\\R---\\s*", Pattern.DOTALL);

    private final Set<String> forbidden;
    private final boolean failOnViolation;

    public PythonImportScanner(List<String> forbidden, boolean failOnViolation) {
   
        // 配置为空时兜底用默认黑名单，否则用配置覆盖（支持自定义扩充/收窄）
        this.forbidden = new LinkedHashSet<>(forbidden == null || forbidden.isEmpty() ? DEFAULT_FORBIDDEN : forbidden);
        this.failOnViolation = failOnViolation;
    }

    /** 扫描结果：违规 import 列表 + 是否放行 */
    public static final class ScanResult {
   
        private final List<String> violations;
        private final boolean allowed;

        ScanResult(List<String> violations, boolean allowed) {
   
            this.violations = violations;
            this.allowed = allowed;
        }

        public List<String> getViolations() {
   
            return violations;
        }

        public boolean isAllowed() {
   
            return allowed;
        }
    }

    /**
     * 扫描 skill 目录下所有 .py 文件。
     *
     * @param skillDir         skill 根目录（含 SKILL.md 与 scripts/）
     * @param skillPermissions SKILL.md frontmatter 声明的 permissions（可空）
     * @return 扫描结果
     */
    public ScanResult scanSkillDir(Path skillDir, Collection<String> skillPermissions) {
   
        Set<String> perms = new HashSet<>(skillPermissions == null ? Set.of() : skillPermissions);
        List<String> allViolations = new ArrayList<>();
        try {
   
            if (Files.exists(skillDir)) {
   
                try (var stream = Files.walk(skillDir)) {
   
                    for (Path p : stream.filter(Files::isRegularFile)
                            .filter(f -> f.toString().endsWith(".py")).toList()) {
   
                        allViolations.addAll(scanFile(p, perms));
                    }
                }
            }
        } catch (IOException e) {
   
            log.warn("[skill-scan] 扫描目录失败 {}: {}", skillDir, e.getMessage());
        }
        boolean allowed = allViolations.isEmpty() || !failOnViolation;
        if (!allViolations.isEmpty()) {
   
            log.warn("[skill-scan] 发现潜在危险 import: {}", String.join(", ", allViolations));
        }
        return new ScanResult(allViolations, allowed);
    }

    /**
     * 扫描单个 .py 文件。
     *
     * @param pyFile           要扫描的 .py 文件
     * @param skillPermissions SKILL.md frontmatter 声明的 permissions（可空）
     * @return 违规描述列表（"文件名: 模块名"），无违规则为空
     */
    public List<String> scanFile(Path pyFile, Collection<String> skillPermissions) {
   
        Set<String> perms = new HashSet<>(skillPermissions == null ? Set.of() : skillPermissions);
        List<String> violations = new ArrayList<>();
        try {
   
            String content = Files.readString(pyFile, StandardCharsets.UTF_8);
            Matcher m = IMPORT_STMT.matcher(content);
            while (m.find()) {
   
                String imported = m.group(1);
                for (String f : forbidden) {
   
                    if (isMatch(imported, f) && !permits(perms, f)) {
   
                        violations.add(pyFile.getFileName() + ": " + imported);
                        break;
                    }
                }
            }
        } catch (IOException e) {
   
            log.warn("[skill-scan] 读取失败 {}: {}", pyFile, e.getMessage());
        }
        return violations;
    }

    /**
     * 从 SKILL.md 中解析 permissions 字段（YAML frontmatter 数组）。
     * 支持格式：permissions: [network, subprocess] 或
     * permissions:\n  - network\n  - subprocess
     */
    public static Set<String> parsePermissions(String skillMdContent) {
   
        Set<String> result = new HashSet<>();
        if (skillMdContent == null) return result;
        Matcher fm = FRONTMATTER.matcher(skillMdContent);
        if (!fm.find()) return result;
        String yaml = fm.group(1);
        // 行内数组形式
        Matcher inline = Pattern.compile("^\\s*permissions\\s*:\\s*\\[([^\\]]*)\\]", Pattern.MULTILINE).matcher(yaml);
        if (inline.find()) {
   
            for (String item : inline.group(1).split(",")) {
   
                String t = item.trim().replaceAll("[\"'\\[\\]]", "");
                if (!t.isEmpty()) result.add(t.toLowerCase());
            }
            return result;
        }
        // 列表形式
        Matcher block = Pattern.compile("^\\s*permissions\\s*:\\s*\\R", Pattern.MULTILINE).matcher(yaml);
        if (block.find()) {
   
            String after = yaml.substring(block.end());
            Matcher item = Pattern.compile("^\\s*-\\s*([^\\n\\r]+)", Pattern.MULTILINE).matcher(after);
            while (item.find()) {
   
                String t = item.group(1).trim().replaceAll("[\"'\\[\\]]", "");
                if (!t.isEmpty()) result.add(t.toLowerCase());
            }
        }
        return result;
    }

    /** 从文件读取 SKILL.md 并解析 permissions */
    public static Set<String> parsePermissions(Path skillDir) {
   
        Path skillMd = skillDir.resolve("SKILL.md");
        if (!Files.exists(skillMd)) return Set.of();
        try {
   
            return parsePermissions(Files.readString(skillMd, StandardCharsets.UTF_8));
        } catch (IOException e) {
   
            return Set.of();
        }
    }

    private boolean isMatch(String importedModule, String forbidden) {
   
        // http / urllib 是标准库包，其子模块（http.server、urllib.request 等）同样具备网络能力，
        // 需要单独列出子模块名精确匹配，避免仅匹配到顶层包时漏掉实际发起网络请求的子模块
        if (forbidden.equals("http")) {
   
            return importedModule.equals("http") || importedModule.equals("http.server")
                    || importedModule.equals("http.client") || importedModule.equals("http.cookiejar");
        }
        if (forbidden.equals("urllib")) {
   
            return importedModule.equals("urllib") || importedModule.startsWith("urllib.");
        }
        // 其余模块按"完全相等 或 顶层包名前缀"匹配（如 subprocess 覆盖 subprocess.Popen）
        if (importedModule.equals(forbidden)) return true;
        if (importedModule.startsWith(forbidden + ".")) return true;
        return false;
    }

    private boolean permits(Set<String> perms, String forbidden) {
   
        // permissions 命名空间：network 放行网络类；subprocess/process 放行子进程类
        // 只有 SKILL.md frontmatter 显式声明了对应权限，才允许该类别下的黑名单 import
        if (perms.contains("network") || perms.contains("net")) {
   
            if (forbidden.equals("socket") || forbidden.equals("urllib") || forbidden.equals("http")
                    || forbidden.equals("ftplib") || forbidden.equals("telnetlib") || forbidden.equals("ftputil")
                    || forbidden.equals("paramiko")) {
   
                return true;
            }
        }
        if (perms.contains("subprocess") || perms.contains("process")) {
   
            if (forbidden.equals("subprocess") || forbidden.equals("multiprocessing")) {
   
                return true;
            }
        }
        return false;
    }

    public boolean isFailOnViolation() {
   
        return failOnViolation;
    }
}

7.3.2 完整性校验器 SkillContentVerifier

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import java.util.zip.ZipOutputStream;

/**
 * Skill 内容完整性校验器。
 *
 * 上传侧：对 zip 内每个文件计算 sha256，生成 .bank-checksum.sha256 清单内嵌 zip，
 *         清单内容用 HMAC-SHA256 签名（密钥来自构造参数，与发布方约定）。
 * 下载侧：先验清单 HMAC 签名，再逐文件比对 sha256，任一失败即拒绝落盘。
 *
 * zip 可能带一层公共根目录（如发布平台以 skill 名包裹），parseZip 统一剥离。
 */
public class SkillContentVerifier {
   

    private static final Logger log = LoggerFactory.getLogger(SkillContentVerifier.class);

    public static final String CHECKSUM_ENTRY = ".bank-checksum.sha256";

    private static final String HMAC_ALGO = "HmacSHA256";
    private static final String SHA256 = "SHA-256";

    private final byte[] hmacKey;

    public SkillContentVerifier(String hmacSecret) {
   
        this.hmacKey = hmacSecret == null || hmacSecret.isBlank() ? new byte[0] : hmacSecret.getBytes(StandardCharsets.UTF_8);
    }

    public boolean isSigningEnabled() {
   
        return hmacKey.length > 0;
    }

    /**
     * 将校验清单写入 zip 输出流。必须在所有业务 entry 写入后调用。
     * 当 hmacKey 未配置时仅写 sha256（无签名），校验时也只验哈希。
     */
    public void writeChecksumEntry(ZipOutputStream zos, Map<String, byte[]> fileContents) throws IOException {
   
        Map<String, String> hashes = new LinkedHashMap<>();
        for (Map.Entry<String, byte[]> e : fileContents.entrySet()) {
   
            hashes.put(e.getKey(), sha256Hex(e.getValue()));
        }
        StringBuilder manifest = new StringBuilder();
        for (Map.Entry<String, String> e : hashes.entrySet()) {
   
            manifest.append(e.getValue()).append("  ").append(e.getKey()).append('\n');
        }
        String body = manifest.toString();
        String content = isSigningEnabled()
                ? body + "signature=" + hmacHex(body)
                : body;
        ZipEntry ze = new ZipEntry(CHECKSUM_ENTRY);
        zos.putNextEntry(ze);
        zos.write(content.getBytes(StandardCharsets.UTF_8));
        zos.closeEntry();
    }

    /**
     * 解析 zip：剥离公共根目录，返回按归一化路径组织的文件内容。
     */
    public ZipContent parseZip(byte[] zip) {
   
        ZipContent content = new ZipContent();
        if (zip == null || zip.length == 0) return content;
        Map<String, byte[]> raw = new LinkedHashMap<>();
        try (ZipInputStream zis = new ZipInputStream(new ByteArrayInputStream(zip))) {
   
            ZipEntry entry;
            while ((entry = zis.getNextEntry()) != null) {
   
                if (entry.isDirectory()) continue;
                String rawName = entry.getName().replace('\\', '/');
                byte[] data = zis.readAllBytes();
                if (rawName.equals(CHECKSUM_ENTRY) || rawName.endsWith("/" + CHECKSUM_ENTRY)) {
   
                    content.checksumContent = data;
                    // 清单所在位置隐含了公共根目录前缀（如发布平台以 skill 名包裹），
                    // 从清单条目路径中提取该前缀，后续对所有业务条目统一剥离
                    content.rootPrefix = rawName.substring(0, rawName.length() - CHECKSUM_ENTRY.length());
                } else {
   
                    raw.put(rawName, data);
                }
            }
        } catch (IOException e) {
   
            log.error("[skill-verify] 解析 zip 失败: {}", e.getMessage());
            return content;
        }
        for (Map.Entry<String, byte[]> e : raw.entrySet()) {
   
            String normalized = stripRoot(e.getKey(), content.rootPrefix);
            if (normalized != null && !normalized.isBlank()) {
   
                content.files.put(normalized, e.getValue());
            }
        }
        return content;
    }

    /**
     * 校验原始 zip 字节（下载后、解压前）。
     *
     * @return true=通过或无清单（本地/无签名 zip）；false=校验失败，应拒绝落盘
     */
    public boolean verifyZip(byte[] zip) {
   
        ZipContent content = parseZip(zip);
        if (content.checksumContent == null) {
   
            log.warn("[skill-verify] zip 无校验清单，跳过");
            return true;
        }
        try {
   
            // 在临时目录重建 zip 内容后按 verifyDir 走同一套校验逻辑，
            // 复用清单解析/哈希比对代码，避免两套实现漂移
            Path tmp = Files.createTempDirectory("skill-verify-");
            for (Map.Entry<String, byte[]> e : content.files.entrySet()) {
   
                Path target = SkillPathGuard.safeResolve(tmp, e.getKey());
                Files.createDirectories(target.getParent());
                Files.write(target, e.getValue());
            }
            Files.write(tmp.resolve(CHECKSUM_ENTRY), content.checksumContent);
            boolean ok = verifyDir(tmp);
            // 校验结束后清理临时目录（从最深路径开始删）
            try (var stream = Files.walk(tmp)) {
   
                stream.sorted(java.util.Comparator.reverseOrder()).forEach(p -> {
   
                    try {
   
                        Files.deleteIfExists(p);
                    } catch (IOException ignored) {
   
                    }
                });
            }
            return ok;
        } catch (Exception e) {
   
            log.error("[skill-verify] zip 校验异常: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 校验解压后的 skill 目录。
     *
     * @param skillDir 已解压到磁盘的 skill 目录
     * @return 校验是否通过
     */
    public boolean verifyDir(Path skillDir) {
   
        Path checksumFile = skillDir.resolve(CHECKSUM_ENTRY);
        if (!Files.exists(checksumFile)) {
   
            log.warn("[skill-verify] {} 缺失，无法校验（本地/无签名 skill 跳过）", skillDir);
            return true;
        }
        try {
   
            List<String> lines = Files.readAllLines(checksumFile, StandardCharsets.UTF_8);
            StringBuilder body = new StringBuilder();
            Map<String, String> expected = new LinkedHashMap<>();
            boolean signed = false;
            for (String line : lines) {
   
                if (line.startsWith("signature=")) {
   
                    signed = true;
                    String sig = line.substring("signature=".length()).trim();
                    // 只有配置了密钥（isSigningEnabled）才要求签名，且必须 constant-time 比较防时序侧信道
                    if (isSigningEnabled() && !constantTimeEquals(hmacHex(body.toString()), sig)) {
   
                        log.error("[skill-verify] {} HMAC 签名不匹配，可能被篡改", skillDir);
                        return false;
                    }
                } else if (!line.isBlank()) {
   
                    // body 累积除 signature 外的所有哈希行，作为 HMAC 的输入原文
                    body.append(line).append('\n');
                    String[] parts = line.split("\\s{2,}", 2);
                    if (parts.length == 2) {
   
                        expected.put(parts[1].trim(), parts[0].trim());
                    }
                }
            }
            // 配置了密钥但清单没签名，说明发布方未按约定签名，视为不可信
            if (isSigningEnabled() && !signed) {
   
                log.error("[skill-verify] {} 配置要求签名但清单未签名", skillDir);
                return false;
            }
            if (expected.isEmpty()) {
   
                log.warn("[skill-verify] {} 清单为空", skillDir);
                return true;
            }
            // 逐文件比对（跳过清单自身）
            for (Map.Entry<String, String> e : expected.entrySet()) {
   
                Path f = skillDir.resolve(e.getKey()).normalize();
                if (!SkillPathGuard.isWithin(skillDir, f)) {
   
                    log.error("[skill-verify] {} 清单条目路径逃逸: {}", skillDir, e.getKey());
                    return false;
                }
                if (!Files.exists(f)) {
   
                    log.error("[skill-verify] {} 清单条目缺失: {}", skillDir, e.getKey());
                    return false;
                }
                byte[] content = Files.readAllBytes(f);
                if (!sha256Hex(content).equals(e.getValue().toLowerCase(Locale.ROOT))) {
   
                    log.error("[skill-verify] {} 文件哈希不匹配: {}", skillDir, e.getKey());
                    return false;
                }
            }
            log.info("[skill-verify] {} 校验通过 ({} files)", skillDir, expected.size());
            return true;
        } catch (IOException e) {
   
            log.error("[skill-verify] 校验失败 {}: {}", skillDir, e.getMessage());
            return false;
        }
    }

    /** 剥离公共根目录前缀 */
    public static String stripRoot(String name, String rootPrefix) {
   
        if (rootPrefix == null || rootPrefix.isEmpty()) return name;
        if (name.startsWith(rootPrefix)) return name.substring(rootPrefix.length());
        return name;
    }

    /** zip 解析结果 */
    public static final class ZipContent {
   
        public final Map<String, byte[]> files = new LinkedHashMap<>();
        public byte[] checksumContent;
        public String rootPrefix = "";
    }

    /** 生成指纹 */  
    private static String sha256Hex(byte[] data) {
     
        try {
     
            MessageDigest md = MessageDigest.getInstance(SHA256);  
            byte[] digest = md.digest(data);  
            StringBuilder sb = new StringBuilder();  
            for (byte b : digest) sb.append(String.format("%02x", b));  
            return sb.toString();  
        } catch (NoSuchAlgorithmException e) {
     
            throw new IllegalStateException("SHA-256 not available", e);  
        }  
    }  

    /** 生成防伪章 */  
    private String hmacHex(String data) {
     
        if (!isSigningEnabled()) return "";  
        try {
     
            Mac mac = Mac.getInstance(HMAC_ALGO);  
            mac.init(new SecretKeySpec(hmacKey, HMAC_ALGO));  
            byte[] digest = mac.doFinal(data.getBytes(StandardCharsets.UTF_8));  
            StringBuilder sb = new StringBuilder();  
            for (byte b : digest) sb.append(String.format("%02x", b));  
            return sb.toString();  
        } catch (Exception e) {
     
            throw new IllegalStateException("HMAC failed", e);  
        }  
    }  

    /** 安全比对 */  
    private static boolean constantTimeEquals(String a, String b) {
     
        if (a == null || b == null || a.length() != b.length()) return false;  
        int result = 0;  
        for (int i = 0; i < a.length(); i++) {
     
            result |= a.charAt(i) ^ b.charAt(i);  
        }  
        return result == 0;  
    }
}

关键实现点：下面三个细节如果不理解，很容易被绕晕：

① 为什么哈希比较要用"常量时间比较"（防时序侧信道）？
校验签名时要把"算出来的签名"和"清单里存的签名"做字符串相等比较。普通写法 a.equals(b) 是逐字符比较、遇到第一个不同的字符就立刻返回——于是比较耗时取决于"在第几个字符就出错"。攻击者可以伪造签名反复试探，通过测量你校验花的时间，一位一位猜出真签名（这就是"时序侧信道"）。MessageDigest.isEqual 会不论在哪一位出错都坚持比完整串，耗时恒定，攻击者从耗时里挖不出任何信息。

② HMAC 的输入到底是哪段文本？
HMAC 是程序用密钥对清单内容算出的"防伪印章"，它不是读者输入的。清单 .bank-checksum.sha256 的结构是：
a3f2c9...  SKILL.md          ← 第 1 行：文件哈希（每个文件一行）
9be1d0...  scripts/extract.py ← 第 2 行：文件哈希
signature=7c4a8e...           ← 最后一行：HMAC 值（程序算的）
"HMAC 的输入是除 signature= 外的全部文本"指的是：程序把上面所有哈希行拼成一段正文（a3f2c9... SKILL.md + 9be1d0... scripts/extract.py），再用密钥对这个正文算 HMAC。signature= 这一行自己不参与运算——否则就是拿自己的结果给自己签名，循环无解。一句话：HMAC = 对"所有文件哈希行"盖的章，签名行自己不算进盖章内容。

③ parseZip 的根目录前缀怎么来的？
zip 里清单文件 .bank-checksum.sha256 所在的位置，就隐含了公共根目录。比如条目叫 skill-x/.bank-checksum.sha256，就说明整个 zip 都包在 skill-x/ 下，于是把所有文件都去掉这个 skill-x/ 前缀再校验。

7.4 插入位置

我们写的这三个 hook，对应 skill 的生命周期（verifier、scanner 来自 7.5 的创建，downloadSkill 是你的下载逻辑，片段需要 java.nio.file.* 等 import，按项目习惯补上即可）：

// 1. 下载后、落盘前 → 完整性校验
byte[] zip = downloadSkill(skillName);
if (!verifier.verifyZip(zip)) {
   
    throw new SecurityException("Skill 内容校验失败，拒绝落盘: " + skillName);
}
var content = verifier.parseZip(zip);
for (var e : content.files.entrySet()) {
   
    Path out = SkillPathGuard.safeResolve(skillsDir, e.getKey()); // 防 ../ 覆盖外部
    Files.write(out, e.getValue());
}

// 2. 物化 / 解压时 → 路径防逃逸（上面的 safeResolve 已覆盖）

// 3. 执行前、拉起 Python 进程前 → 静态扫描
var perms = PythonImportScanner.parsePermissions(skillDir);
var result = scanner.scanSkillDir(skillDir, perms);
if (!result.isAllowed()) {
   
    throw new SecurityException("脚本未通过静态扫描: " + result.getViolations());
}

7.5 用不用 Spring 都可以

上文三个类没有依赖 Spring 注解，直接手动 new 即可使用；如果你在用 Spring，只需给类加 @Component、构造参数加 @Value，其余代码一字不改：

// 无 Spring 时：手动创建并当普通依赖传递
PythonImportScanner scanner =
        new PythonImportScanner(List.of("subprocess", "socket"), true);
SkillContentVerifier verifier = new SkillContentVerifier("你的HMAC密钥");

// 有 Spring 时：只需这样声明，依赖注入交给框架
// @Component class PythonImportScanner {
   
//     public PythonImportScanner(
//         @Value("${app.ai.skill.verify.forbidden-imports:}") List<String> forbidden,
//         @Value("${app.ai.skill.verify.fail-on-violation:true}") boolean failOnViolation) { ... }
// }

八、知识点小结

供应链安全的消费侧防线：发布侧治理（审核/权限/生命周期）管"谁发布"，消费侧校验管"到我手上的是什么"，两者缺一不可，尤其当发布管线有 Force Publish 这种绕过路径时。
哈希 ≠ 签名：哈希防"意外损坏"，HMAC/签名防"恶意篡改"。用哈希做完整性校验时，清单本身必须带密钥签名，否则等于没验。
白名单优于黑名单：黑名单永远追不完（exec 就是个无底洞），声明式 permissions 把"默认拒绝 + 显式授权"落到 skill 元数据里，可审计、可灰度。
路径安全三件套：先 normalize 再比较、拒绝绝对路径、两种分隔符都测。Path.startsWith 是词法比较，不是真实文件系统比较，依赖它之前必须先归一化。
静态分析是成本的，不是银弹：静态扫描拦截明显恶意很值，但对付 exec 拼接、混淆 import 无能为力——所以阶段二要上执行隔离（沙箱）和输出校验。防线分层，每层解决一类威胁，才是防御的常态。

本文来自一次真实加固实践。踩坑最深的是"哈希清单也要签名"和"startsWith 之前必须先 normalize"，前者让篡改检测形同虚设，后者放过了经典 Zip Slip，供后来者参考。

【雕虫大技】Agent 动态 Skill 供应链安全加固（一）：消费侧三层防线实战

一、背景

二、整体设计：发布侧治理管不了运行时

三、第一层：zip 完整性校验（防篡改）

3.1 为什么把校验清单内嵌进 zip

3.2 为什么光有 sha256 还不够

3.3 密钥从哪来，信任边界在哪

3.4 实现要点

四、第二层：Python 静态 import 扫描（防恶意代码）

4.1 黑名单选型：只拦 fork/exec 型模块

4.2 权限放行机制：SKILL.md frontmatter permissions

五、第三层：路径防逃逸（防 Zip Slip / 越界写）

5.1 `resolve` 加 `startsWith` 为什么会被绕过

5.2 修复方式：先 normalize 再判断，且必须统一落点在 base 内

六、落地与验证

6.1 组件总览

6.2 防线落点：三道防线分别嵌入生命周期的哪个环节

七、复用指南

7.1 组件依赖关系

7.2 路径守卫 SkillPathGuard

7.3 完整代码：静态扫描器与完整性校验器

7.3.1 静态扫描器 PythonImportScanner

7.3.2 完整性校验器 SkillContentVerifier

7.4 插入位置

7.5 用不用 Spring 都可以

八、知识点小结

阿里云百炼

热门文章

最新文章

相关电子书

【雕虫大技】Agent 动态 Skill 供应链安全加固（一）：消费侧三层防线实战

一、背景

二、整体设计：发布侧治理管不了运行时

三、第一层：zip 完整性校验（防篡改）

3.1 为什么把校验清单内嵌进 zip

3.2 为什么光有 sha256 还不够

3.3 密钥从哪来，信任边界在哪

3.4 实现要点

四、第二层：Python 静态 import 扫描（防恶意代码）

4.1 黑名单选型：只拦 fork/exec 型模块

4.2 权限放行机制：SKILL.md frontmatter permissions

五、第三层：路径防逃逸（防 Zip Slip / 越界写）

5.1 resolve 加 startsWith 为什么会被绕过

5.2 修复方式：先 normalize 再判断，且必须统一落点在 base 内

六、落地与验证

6.1 组件总览

6.2 防线落点：三道防线分别嵌入生命周期的哪个环节

七、复用指南

7.1 组件依赖关系

7.2 路径守卫 SkillPathGuard

7.3 完整代码：静态扫描器与完整性校验器

7.3.1 静态扫描器 PythonImportScanner

7.3.2 完整性校验器 SkillContentVerifier

7.4 插入位置

7.5 用不用 Spring 都可以

八、知识点小结

阿里云百炼

热门文章

最新文章

相关电子书

5.1 `resolve` 加 `startsWith` 为什么会被绕过