Python ZIP 安全解压：拒绝危险路径，并给实际输出设上限

接收一个 ZIP 文件，随后调用 extractall()，看起来只是几行胶水代码。问题在于，归档成员名来自输入，它会参与构造输出路径；压缩包还会声明文件类型和大小。这些字段不是应用已经认可的业务事实。处理上传附件、离线数据包或插件资源时，解压实际上是一次受外部输入影响的文件系统写入。

本文用 Python 标准库做一个本地、可复现的小实验：生成正常 ZIP 与恶意 ZIP，在全新临时目录中解压，检查拒绝原因，再清理现场。它不是通用解压器，也不承诺抵挡拥有本机写权限的攻击者。实验没有连接云服务，没有性能对比，结论只覆盖这里展示的策略和验证。

一、先定义接受什么，再讨论怎么拦截

示例接受普通文件和显式目录项，允许嵌套目录，文件内容不做业务解释。名字只允许 ASCII 字母、数字、下划线、连字符及分隔扩展名的点，路径分隔符只接受 /。这会拒绝中文名、空格和很多合法 ZIP；对一个讲清边界的教学示例，这种兼容性代价比隐含的跨平台解释更容易检查。

路径处理采用拒绝策略，不把 a/../b“修复”为 b。绝对路径、空段、.、..、反斜杠和冒号均被拒绝；因此盘符路径、盘符相对路径、UNC 路径及 Windows 备用数据流写法也不能通过。尾随点、空格和常见设备保留名不在允许范围内。单个名字长度和目录深度也有小上限。

重复检查使用已经通过上述规则的完整路径，再转为小写作为键。它会保守地把 A.txt 与 a.txt 视为冲突。由于点段和重复分隔符直接被拒绝，不存在先把它们折叠再覆盖旧文件的步骤；文件与目录的前缀冲突也会被拒绝。这只是本示例受限命名集合内的规则，不是所有文件系统的完整名称等价算法。

二、两道大小限制，分别回答不同问题

第一道检查中央目录：限制成员数、单个成员声明大小、声明总大小，同时拒绝符号链接、其他特殊文件、加密成员及不接受的压缩方式。默认预算是最多十六个成员、单个文件一千零二十四字节、总输出四千零九十六字节，故意设得很小，便于一眼核对实验。业务接入时应先定义可接受的数据规模，再调整这些值，而不是直接搬用演示数字。它们不是推荐的生产容量，也没有来自性能测试的依据。成员数包括目录项，避免攻击者用大量空目录绕过“文件数”。external_attr 的 Unix 文件类型位只作为拒绝依据；程序不会恢复权限、创建链接或执行归档内容。

第二道限制放在复制循环里，按实际读出的未压缩字节计数。每次最多读取当前剩余额度再加一个字节，若越界就在写入前报错，避免把超限块先写到磁盘。读取完成后还会核对该成员的实际长度与声明值。实际计数不能替代声明检查：前者控制复制，后者提前拒绝显然不合要求的包。

也不能把声明字段当成唯一事实。具体畸形 ZIP 在不同 Python 版本中可能先触发 CRC 或格式错误，而不是走到应用的大小错误。下面直接给复制函数喂入超限字节流，验证实际限制确实存在；这两项不是伪造“绕过 zipfile 声明检查”的实测结论。

三、完整代码：只向新建临时目录写入

将下面代码保存为 zip-safe-extract-demo.py。只使用标准库。safe_extract() 是上下文管理器：调用方只能在 with 块内读取结果，退出时结果目录被清理；校验或复制异常同样触发清理。它没有“目标目录”参数，避免把不可信归档合并进已有目录。示例生成 ZIP 时固定时间戳、内容和顺序，不依赖下载样本。

"""A conservative ZIP teaching example; all output lives in fresh temp space."""
from contextlib import contextmanager
from io import BytesIO
from pathlib import Path
import re
import stat
import tempfile
import zipfile


class Rejected(ValueError):
    pass


def checked_parts(info):
    # orig_filename retains a NUL that ZipInfo.filename can truncate.
    name = info.orig_filename
    if "\x00" in name:
        raise Rejected("ambiguous-name")
    if name.startswith("/"):
        raise Rejected("absolute-path")
    if "\\" in name or ":" in name:
        raise Rejected("windows-path")
    if name != info.filename:
        raise Rejected("ambiguous-name")
    directory = info.is_dir()
    body = name[:-1] if directory else name
    parts = body.split("/")
    if any(p in ("", ".", "..") for p in parts):
        raise Rejected("dot-or-empty-segment")
    if len(name) > 200 or len(parts) > 12:
        raise Rejected("name-budget")
    reserved = {"CON", "PRN", "AUX", "NUL"}
    reserved |= {f"{p}{i}" for p in ("COM", "LPT") for i in range(1, 10)}
    for part in parts:
        if not re.fullmatch(r"[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+)*", part):
            raise Rejected("name-policy")
        if part.split(".", 1)[0].upper() in reserved:
            raise Rejected("reserved-name")
    mode = info.external_attr >> 16
    if stat.S_ISLNK(mode):
        raise Rejected("symlink")
    kind = stat.S_IFMT(mode)
    if kind not in (0, stat.S_IFREG, stat.S_IFDIR):
        raise Rejected("special-file")
    if kind == stat.S_IFDIR and not directory:
        raise Rejected("type-mismatch")
    if kind == stat.S_IFREG and directory:
        raise Rejected("type-mismatch")
    return tuple(parts), directory


def make_plan(archive, max_members, max_file, max_total):
    members = archive.infolist()
    if len(members) > max_members:
        raise Rejected("member-count")
    seen, files, dirs, plan = set(), set(), set(), []
    declared = 0
    for info in members:
        parts, directory = checked_parts(info)
        key = "/".join(parts).lower()
        if key in seen:
            raise Rejected("duplicate-path")
        seen.add(key)
        if (directory and key in files) or (not directory and key in dirs):
            raise Rejected("file-directory-conflict")
        for i in range(1, len(parts)):
            parent = "/".join(parts[:i]).lower()
            if parent in files:
                raise Rejected("file-directory-conflict")
            dirs.add(parent)
        (dirs if directory else files).add(key)
        if info.flag_bits & 1:
            raise Rejected("encrypted-entry")
        if info.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED):
            raise Rejected("compression-method")
        if directory and info.file_size != 0:
            raise Rejected("directory-payload")
        if info.file_size < 0 or info.file_size > max_file:
            raise Rejected("declared-file-size")
        declared += info.file_size
        if declared > max_total:
            raise Rejected("declared-total-size")
        plan.append((info, parts, directory))
    return plan


def copy_limited(source, target, used, max_file, max_total):
    count = 0
    while True:
        # Read at most one byte beyond either remaining allowance.
        allowance = min(max_file - count, max_total - used - count)
        block = source.read(min(65536, allowance + 1))
        if not block:
            return count
        if len(block) > max_file - count:
            raise Rejected("actual-file-size")
        if len(block) > max_total - used - count:
            raise Rejected("actual-total-size")
        target.write(block)
        count += len(block)


@contextmanager
def safe_extract(blob, *, max_members=16, max_file=1024, max_total=4096):
    if min(max_members, max_file, max_total) < 0:
        raise ValueError("limits must be nonnegative")
    # No caller-supplied destination, no merge into an existing directory.
    with tempfile.TemporaryDirectory(prefix="zip-safe-") as temp:
        root = Path(temp)
        with zipfile.ZipFile(BytesIO(blob)) as archive:
            plan = make_plan(archive, max_members, max_file, max_total)
            used = 0
            for info, parts, directory in plan:
                target = root.joinpath(*parts)
                if directory:
                    target.mkdir(parents=True, exist_ok=True)
                    continue
                target.parent.mkdir(parents=True, exist_ok=True)
                with archive.open(info) as source, target.open("xb") as output:
                    count = copy_limited(source, output, used, max_file, max_total)
                if count != info.file_size:
                    raise Rejected("size-mismatch")
                used += count
        yield root  # Valid only inside this with block; cleanup also on errors.


def sample(entries):
    stream = BytesIO()
    with zipfile.ZipFile(stream, "w", compression=zipfile.ZIP_STORED) as archive:
        for name, data, mode in entries:
            info = zipfile.ZipInfo(name, date_time=(2026, 11, 4, 0, 0, 0))
            # Windows ZipInfo creation can convert backslashes to slashes.
            # Preserve the attack spelling in the generated archive itself.
            info.filename = info.orig_filename = name
            info.create_system = 3
            info.external_attr = mode << 16
            archive.writestr(info, data)
    return stream.getvalue()


def cases():
    regular = stat.S_IFREG | 0o600
    def one(name, data=b"x", mode=regular):
        return sample([(name, data, mode)])
    return [
        ("normal", sample([("docs/", b"", stat.S_IFDIR | 0o700),
                           ("docs/readme.txt", b"hello\n", regular),
                           ("data.csv", b"a,b\n", regular)]), {}, None),
        ("traversal", one("../escape.txt"), {}, "dot-or-empty-segment"),
        ("absolute", one("/escape.txt"), {}, "absolute-path"),
        ("drive", one("C:/escape.txt"), {}, "windows-path"),
        ("backslash", one("dir\\escape.txt"), {}, "windows-path"),
        ("symlink", one("link", b"outside", stat.S_IFLNK | 0o777), {}, "symlink"),
        ("duplicate", sample([("A.txt", b"1", regular),
                              ("a.txt", b"2", regular)]), {}, "duplicate-path"),
        ("dot-segment", one("a/./b.txt"), {}, "dot-or-empty-segment"),
        ("count", sample([("a", b"1", regular), ("b", b"2", regular)]),
         {"max_members": 1}, "member-count"),
        ("file-size", one("large", b"12345"), {"max_file": 4}, "declared-file-size"),
        ("total-size", sample([("a", b"123", regular), ("b", b"456", regular)]),
         {"max_total": 5}, "declared-total-size"),
    ]


def main():
    for label, blob, limits, expected in cases():
        try:
            with safe_extract(blob, **limits) as root:
                assert expected is None, label
                result = sorted((p.relative_to(root).as_posix(), p.read_bytes())
                                for p in root.rglob("*") if p.is_file())
                assert result == [("data.csv", b"a,b\n"),
                                  ("docs/readme.txt", b"hello\n")]
                print(f"{label}: OK files=2 bytes=10")
            assert not root.exists()
        except Rejected as error:
            assert str(error) == expected, (label, str(error), expected)
            print(f"{label}: REJECT {error}")
    # Exercise the same copy loop independently of ZIP header declarations.
    for label, used, per_file, total, expected in [
        ("actual-file", 0, 4, 10, "actual-file-size"),
        ("actual-total", 3, 10, 7, "actual-total-size"),
    ]:
        output = BytesIO()
        try:
            copy_limited(BytesIO(b"12345"), output, used, per_file, total)
        except Rejected as error:
            assert str(error) == expected
            assert len(output.getvalue()) <= min(per_file, total - used)
            print(f"{label}: REJECT {error}")
        else:
            raise AssertionError(label)
    print("all 13 checks passed")


if __name__ == "__main__":
    main()

orig_filename 的检查用于避免内嵌 NUL 被文件名处理截断后形成歧义。目录项需要是空内容。输出文件通过 xb 排他创建；这样意外存在的同名文件会触发异常，不能静默覆盖。整个计划审查通过后才开始复制，符号链接条目不会被当作普通文件“兼容处理”。

这里还有一个容易让测试产生假阳性的细节：本机 Windows 环境中的 ZipInfo 会把反斜杠转换成正斜杠。如果直接拿它生成恶意包，最终包里的名字可能已经被改成正常路径，所谓“反斜杠攻击测试”便没有测到目标。生成器明确保留输入拼写，审查器也先看原始名字，确保拒绝理由对应归档中的真实字段。生成样本与消费样本使用同一库时，尤其需要检查这种预处理。

四、复制命令与可核对的输出

文章所附演示代码是完整脚本，可直接复制保存后运行。在脚本所在目录打开终端，确认 python 指向本机 Python 3，执行：

python -B ./zip-safe-extract-demo.py

本地实验的标准输出如下，包含正常文件内容核对、拒绝原因和两项实际复制限制。程序没有把随机临时路径、耗时或时钟写进输出，适合比较两次运行是否一致。

normal: OK files=2 bytes=10
traversal: REJECT dot-or-empty-segment
absolute: REJECT absolute-path
drive: REJECT windows-path
backslash: REJECT windows-path
symlink: REJECT symlink
duplicate: REJECT duplicate-path
dot-segment: REJECT dot-or-empty-segment
count: REJECT member-count
file-size: REJECT declared-file-size
total-size: REJECT declared-total-size
actual-file: REJECT actual-file-size
actual-total: REJECT actual-total-size
all 13 checks passed

作者本地另用补充验证器实际运行演示两次并逐字节比较标准输出，另外验证失败清理、CRC 异常、文件目录冲突和复制中途失败。该验证器不是运行本文完整演示代码的前置条件。临时目录的随机名字不同并不妨碍输出一致。这证明的是测试场景在本机运行通过，不代表已经穷举 ZIP 格式、所有操作系统或文件系统。

五、失败清理不是事务提交，也不是磁盘配额

全新目录让失败处理更直观：异常向外传播，由 TemporaryDirectory 回收本次产生的内容。即使正常解压成功，消费者的 with 块抛出异常也会清理。真实程序若需要持久化，应在成功校验后设计单独的接收和提交步骤；这里没有把临时结果发布为长期文件，更没有演示跨卷移动的原子性。

清理仍依赖操作系统允许删除。文件被外部进程占用、权限变化、进程被强杀、机器断电，都可能使自动清理失败或来不及执行。因此不能写成“发生任何错误都不会留下文件”。生产环境通常需要独立暂存区、残留回收任务与可观察的失败记录。

总输出字节上限也不是磁盘配额。目录和文件系统元数据占空间，多个并发解压任务会累计占用；归档本体、内存副本和其他进程同样消耗资源。部署时需要对暂存文件系统设置配额或容量隔离，并限制并发任务数。磁盘空间不足是正常错误路径，不能因为预先检查过剩余空间就假定之后一定够用。

六、边界必须写在“安全”二字旁边

本例接收 bytes，调用前归档已在内存中；ZipFile 解析中央目录后才执行成员数检查。因此它没有限制上传体积，也没有给解析阶段提供严格内存上界。接入网络时还需在接收阶段限制压缩输入字节，配合任务超时、CPU 和内存隔离。解压输出受限不等于压缩算法计算耗时受限，本文没有测量压缩炸弹的性能。

临时目录应位于受信任的暂存父目录，并由运行账号控制。名字检查、xb 和新目录减少了误写风险，但不是完整的抗竞态机制：若另一进程能替换父目录或把内部目录换成符号链接、junction 等重解析对象，路径检查与实际打开之间仍有 TOCTOU 窗口。恶意同用户进程和拥有更高权限的进程不在这段代码的保障范围内。

不同系统需要不同的目录句柄和无跟随打开策略，不能拿一次 resolve() 后的前缀检查冒充跨平台完整隔离。面对主动本机攻击者，应把解压放在低权限隔离进程或容器中，给它专用输出挂载和资源额度，再讨论平台特有的文件打开原语。

最后，内容本身仍是不可信数据：普通文件也可能包含脚本、宏或有害配置。解压后如何检查内容、谁能读取、是否执行，是下一层业务策略。这个实验提供的是一套可运行、可复核的保守路径与输出预算规则；它的价值在于把允许项、拒绝项和未解决的边界同时摆出来。