Python 解压不可信 ZIP：同时防住路径穿越、符号链接与解压炸弹

适用场景

本文适用于允许用户上传 ZIP 的报表导入、主题包安装、批量图片处理和离线数据交换服务。压缩包来自外部时，风险不只是文件名中出现 ../：绝对路径、Windows 盘符、反斜杠、同名覆盖、Unix 符号链接以及极高压缩比，都可能让一次普通解压变成任意文件写入或磁盘耗尽。

目标是建立一个可验收的解压边界：先检查归档目录，再把内容流式写入新的隔离目录；限制成员数、单文件大小、总展开大小和压缩比；拒绝链接与路径歧义；任何成员失败时整批失败，不把半成品目录交给后续业务。

常见误区与现场现象

生产代码里经常能看到下面这种处理：

from zipfile import ZipFile


with ZipFile(upload_path) as archive:
    archive.extractall(work_dir)

Python 的 zipfile 会对一部分绝对路径和 .. 做规范化，但官方文档仍明确要求：不要在未经检查的情况下解压不可信归档。更重要的是，路径规范化并不能解决资源耗尽、重复文件、链接语义和业务允许类型等问题。

典型故障或攻击迹象包括：

• 上传文件只有几 MB，解压目录却迅速占满磁盘；
• 归档成员名为 ../../app/config.py、C:/Windows/... 或 ..\\..\\config；
• Linux 上制作的归档携带符号链接，后续处理程序沿链接读写了隔离目录外的文件；
• 归档包含数十万个零字节文件，磁盘容量没满但 inode、CPU 或目录遍历时间耗尽；
• 同一路径出现多次，后一个成员覆盖前一个成员，校验阶段看到的内容与业务实际读取的不一致；
• Windows 大小写不敏感文件系统上，Report.csv 与 report.csv 发生碰撞。

因此，安全检查不能只写一句 if ".." not in filename。字符串包含判断既会误伤 report..csv，也识别不了反斜杠、盘符和规范化后的碰撞。

先定义解压预算

每个上传入口都应根据真实业务定义限制，而不是复制一个“足够大”的全局值：

• 最大成员数：防止海量小文件消耗 inode 和遍历时间；
• 单个文件最大展开大小：阻止单成员撑爆磁盘；
• 全部文件最大展开大小：限制整批任务的最坏空间成本；
• 最大压缩比：尽早拒绝高度可疑的压缩炸弹；
• 允许的文件后缀与目录层级：减少后续解析器接触面；
• 隔离目录配额和任务超时：元数据可能伪造，应用层限制不能替代系统配额。

压缩比只是风险信号，不是恶意判定。大量重复文本本来就可能有很高压缩比；阈值需要结合业务样本调整。反过来，攻击者也能填充随机数据绕过压缩比，所以仍必须检查实际写出的字节数。

一个可复用的安全解压器

下面实现只依赖 Python 标准库，面向 Python 3.11 及以上版本。它有意不调用 extractall()，而是逐个成员校验并流式复制。

from __future__ import annotations

import shutil
import stat
from dataclasses import dataclass
from pathlib import Path, PurePosixPath, PureWindowsPath
from zipfile import BadZipFile, ZipFile, ZipInfo


class UnsafeArchiveError(ValueError):
    """压缩包违反安全约束。"""


@dataclass(frozen=True, slots=True)
class ExtractLimits:
    """解压资源上限。"""

    max_members: int = 1_000
    max_file_bytes: int = 50 * 1024 * 1024
    max_total_bytes: int = 200 * 1024 * 1024
    max_compression_ratio: float = 200.0


@dataclass(frozen=True, slots=True)
class ExtractResult:
    """安全解压结果。"""

    file_count: int
    total_bytes: int


def _is_symlink(info: ZipInfo) -> bool:
    """识别由 Unix 创建的符号链接成员。"""
    unix_mode = info.external_attr >> 16
    return info.create_system == 3 and stat.S_ISLNK(unix_mode)


def _normalize_member_name(name: str) -> PurePosixPath:
    """把成员名收敛为无歧义的相对 POSIX 路径。"""
    if not name or "\x00" in name:
        raise UnsafeArchiveError("成员名为空或包含空字符")
    if "\\" in name:
        raise UnsafeArchiveError(f"成员名包含反斜杠: {name!r}")

    posix_path = PurePosixPath(name)
    windows_path = PureWindowsPath(name)
    if posix_path.is_absolute() or windows_path.is_absolute() or windows_path.drive:
        raise UnsafeArchiveError(f"成员名不能使用绝对路径或盘符: {name!r}")
    if any(part in {"", ".", ".."} for part in posix_path.parts):
        raise UnsafeArchiveError(f"成员名包含非法路径段: {name!r}")

    return posix_path


def _check_declared_size(info: ZipInfo, limits: ExtractLimits) -> None:
    """根据中央目录元数据做快速拒绝。"""
    if info.file_size < 0 or info.compress_size < 0:
        raise UnsafeArchiveError(f"成员大小非法: {info.filename!r}")
    if info.file_size > limits.max_file_bytes:
        raise UnsafeArchiveError(f"成员展开后过大: {info.filename!r}")

    if info.file_size == 0:
        return
    if info.compress_size == 0:
        raise UnsafeArchiveError(f"非空成员的压缩大小为零: {info.filename!r}")

    ratio = info.file_size / info.compress_size
    if ratio > limits.max_compression_ratio:
        raise UnsafeArchiveError(
            f"成员压缩比过高: {info.filename!r}, ratio={ratio:.1f}"
        )


def safe_extract_zip(
    archive_path: Path,
    destination: Path,
    *,
    limits: ExtractLimits = ExtractLimits(),
) -> ExtractResult:
    """将 ZIP 安全解压到一个预期为空的私有目录。"""
    if min(
        limits.max_members,
        limits.max_file_bytes,
        limits.max_total_bytes,
    ) <= 0:
        raise ValueError("成员数和字节数上限必须大于零")
    if limits.max_compression_ratio <= 0:
        raise ValueError("压缩比上限必须大于零")

    destination.mkdir(parents=True, exist_ok=False)
    root = destination.resolve(strict=True)
    seen_paths: set[str] = set()
    total_declared = 0
    total_written = 0
    file_count = 0

    try:
        with ZipFile(archive_path, mode="r") as archive:
            members = archive.infolist()
            if len(members) > limits.max_members:
                raise UnsafeArchiveError("压缩包成员数量超过限制")

            for info in members:
                relative_path = _normalize_member_name(info.filename)
                collision_key = relative_path.as_posix().casefold().rstrip("/")
                if collision_key in seen_paths:
                    raise UnsafeArchiveError(
                        f"成员路径重复或大小写冲突: {info.filename!r}"
                    )
                seen_paths.add(collision_key)

                if _is_symlink(info):
                    raise UnsafeArchiveError(
                        f"压缩包不允许包含符号链接: {info.filename!r}"
                    )

                target = (root.joinpath(*relative_path.parts)).resolve(strict=False)
                if not target.is_relative_to(root):
                    raise UnsafeArchiveError(
                        f"成员目标越过解压目录: {info.filename!r}"
                    )

                if info.is_dir():
                    target.mkdir(parents=True, exist_ok=True)
                    continue

                _check_declared_size(info, limits)
                total_declared += info.file_size
                if total_declared > limits.max_total_bytes:
                    raise UnsafeArchiveError("压缩包声明的总展开大小超过限制")

                target.parent.mkdir(parents=True, exist_ok=True)
                written_for_file = 0
                with archive.open(info, mode="r") as source, target.open("xb") as output:
                    while chunk := source.read(1024 * 1024):
                        written_for_file += len(chunk)
                        total_written += len(chunk)
                        if written_for_file > limits.max_file_bytes:
                            raise UnsafeArchiveError(
                                f"成员实际展开大小超过限制: {info.filename!r}"
                            )
                        if total_written > limits.max_total_bytes:
                            raise UnsafeArchiveError("实际总展开大小超过限制")
                        output.write(chunk)
                file_count += 1
    except (BadZipFile, OSError, RuntimeError) as exc:
        shutil.rmtree(root, ignore_errors=True)
        raise UnsafeArchiveError(
            f"压缩包读取或写入失败: error_type={type(exc).__name__}"
        ) from exc
    except Exception:
        shutil.rmtree(root, ignore_errors=True)
        raise

    return ExtractResult(file_count=file_count, total_bytes=total_written)

几个关键点值得单独说明：

• ZIP 内部路径约定使用 /。直接拒绝反斜杠，可以避免同一名称在 Linux 与 Windows 上被解释成不同路径；
• 同时用 PurePosixPath 和 PureWindowsPath 判断绝对路径与盘符，不能只按当前服务器操作系统解释攻击者提供的名称；
• resolve(strict=False) 消除 .. 并解析已经存在的链接，再用 is_relative_to() 验证目标仍在根目录内；
• external_attr 的高 16 位可携带 Unix 文件模式。这里拒绝符号链接，而不是尝试在目标机还原它；
• 先按中央目录中的 file_size 快速拒绝，再对实际流出的字节计数。元数据检查提高效率，实际计数才是最终边界；
• 输出文件使用 xb 创建，已存在文件会失败，避免静默覆盖；路径还用 casefold() 去重，结果在大小写敏感和不敏感文件系统上保持一致；
• 任何异常都会删除本次目标目录。调用方应给每个任务创建新的随机私有目录，不要把多个请求解压到共享目录。

这段代码解决的是应用层边界，不是强隔离。若本地其他账号能在解压期间修改目标目录，路径检查与文件创建之间仍存在竞态条件。生产环境应让任务目录仅对服务账号可写，并在容器、独立卷或受配额文件系统中执行；对高风险上传，可在无网络、只读根文件系统和最小权限的沙箱进程中解压。

用测试固定攻击边界

下面的 pytest 用例全部在临时目录中运行，不依赖真实外部服务：

import stat
from pathlib import Path
from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo

import pytest

from safe_zip import ExtractLimits, UnsafeArchiveError, safe_extract_zip


def _write_zip(path: Path, files: dict[str, bytes]) -> None:
    """创建测试归档。"""
    with ZipFile(path, mode="w", compression=ZIP_DEFLATED) as archive:
        for name, content in files.items():
            archive.writestr(name, content)


def test_extracts_normal_files(tmp_path: Path) -> None:
    """正常归档应写入隔离目录并返回实际字节数。"""
    archive_path = tmp_path / "normal.zip"
    destination = tmp_path / "output"
    _write_zip(archive_path, {"reports/summary.txt": "完成".encode("utf-8")})

    result = safe_extract_zip(archive_path, destination)

    assert (destination / "reports" / "summary.txt").read_text("utf-8") == "完成"
    assert result.file_count == 1
    assert result.total_bytes == len("完成".encode("utf-8"))


@pytest.mark.parametrize(
    "member_name",
    ["../outside.txt", "/etc/cron.d/job", "C:/Windows/task.txt", "..\\outside.txt"],
)
def test_rejects_ambiguous_or_escaping_paths(
    tmp_path: Path,
    member_name: str,
) -> None:
    """路径穿越、绝对路径和跨平台歧义路径必须被拒绝。"""
    archive_path = tmp_path / "traversal.zip"
    _write_zip(archive_path, {member_name: b"blocked"})

    with pytest.raises(UnsafeArchiveError):
        safe_extract_zip(archive_path, tmp_path / "output")

    assert not (tmp_path / "outside.txt").exists()


def test_rejects_unix_symlink(tmp_path: Path) -> None:
    """Unix 符号链接成员不得进入解压目录。"""
    archive_path = tmp_path / "symlink.zip"
    info = ZipInfo("latest")
    info.create_system = 3
    info.external_attr = (stat.S_IFLNK | 0o777) << 16
    with ZipFile(archive_path, mode="w") as archive:
        archive.writestr(info, "../../etc/passwd")

    with pytest.raises(UnsafeArchiveError):
        safe_extract_zip(archive_path, tmp_path / "output")


def test_rejects_excessive_compression_ratio(tmp_path: Path) -> None:
    """高度压缩的成员应在写盘前被快速拒绝。"""
    archive_path = tmp_path / "bomb.zip"
    _write_zip(archive_path, {"zeros.bin": b"0" * 1024 * 1024})
    limits = ExtractLimits(max_compression_ratio=10.0)

    with pytest.raises(UnsafeArchiveError):
        safe_extract_zip(archive_path, tmp_path / "output", limits=limits)

项目内还应增加以下用例：成员数刚好等于和超过上限、单文件与总大小边界、重复文件名、仅大小写不同的文件名、损坏的中央目录、加密归档、磁盘写入失败，以及业务后缀允许列表。安全测试不仅要断言抛出异常，还要断言隔离目录已清理、目录外没有新文件。

接入上传服务时的完整流程

安全解压只是上传链路中的一步，推荐流程如下：

1. 上传流先写入有字节数上限的临时文件，不把整个请求体读入内存；
2. 计算服务端任务 ID，并为该任务创建不可预测、仅服务账号可写的目标目录；
3. 校验 ZIP 结构和资源预算，安全解压到隔离目录；
4. 对解压后的普通文件按业务格式重新校验，不信任扩展名和压缩包内的 MIME 声明；
5. 图片、Office 文档或其他复杂格式交给低权限、无网络的解析进程；
6. 全部验证通过后，以目录重命名等原子方式发布结果；失败则删除整批临时产物；
7. 记录成员数、压缩大小、实际展开大小、耗时和拒绝原因，不记录文件内容或用户敏感数据。

日志字段可使用 archive_id、member_count、compressed_bytes、extracted_bytes、duration_ms、reject_reason。固定消息使用中文；不要记录原始文件内容、访问令牌、完整本地路径或未经清洗的成员名，以免日志注入和敏感信息泄露。

监控与上线验证

上线后至少观察以下指标：

• 上传 ZIP 的压缩大小与实际展开大小分布；
• 按 reject_reason 分类的拒绝次数；
• 解压任务耗时、超时数和并发数；
• 临时卷使用率、inode 使用率和任务目录清理失败数；
• 后续解析器的崩溃、内存峰值和异常文件类型。

发布前用一组固定攻击样本做回归：路径穿越、Windows 盘符、反斜杠、符号链接、重复成员、高压缩比、大量空文件和损坏 ZIP。再在受限容器或测试卷中验证磁盘配额和任务超时确实生效。只有应用限制、文件系统配额和进程隔离同时存在，才能把元数据欺骗、竞态和未知解析器漏洞的影响收敛在可控范围内。

总结

处理不可信 ZIP 的核心不是“把 ../ 替换掉”，而是把解压视为一次受预算约束的文件写入操作。路径必须采用跨平台规则解析并确认留在隔离根目录；链接、重复名称和覆盖行为应显式拒绝；大小既看声明值，也统计实际写入值；失败产物不能进入后续业务。

在这个应用边界之外，还要依靠私有临时目录、磁盘和 inode 配额、任务超时、低权限进程以及格式级校验。这样即使收到畸形归档或压缩炸弹，也能在文件进入正式存储前终止处理，并留下足够但不泄密的诊断证据。

参考资料

• Python 官方文档：zipfile
• Python 官方文档：pathlib
• Python 官方文档：安全注意事项