用 Python 流式解压超大 ZIP：限制总大小与压缩比防 ZIP bomb

ZIP bomb（压缩炸弹）是一种利用高压缩比文件耗尽磁盘或内存的攻击方式。一个几十 KB 的 ZIP 文件，解压后可能达到数 GB。Python 的 zipfile 提供了 extractall()，但它会一次性把所有内容写到磁盘，缺少总量和压缩比限制。本文演示如何逐条流式解压，并在读取过程中实时校验。

问题场景

假设服务端允许用户上传 ZIP 并解压。如果直接：

import zipfile
zipfile.ZipFile('upload.zip').extractall('output')

攻击者上传一个包含大量重复数据的 ZIP，extractall() 会按声明的大小写入，直到磁盘写满。更隐蔽的是，ZIP 头部的 file_size 字段可能被伪造得很小，但实际解压数据很大。因此不能只信任头部声明。

解决思路

1. 不调用 extractall()，改用 ZipFile.infolist() 遍历每个条目。
2. 用 ZipFile.open(info) 获得文件流，按固定大小分块读取并写入。
3. 预检：单文件压缩比 file_size / compress_size、单文件大小、累计总大小。
4. 实时校验：在分块读取时累计实际解压字节数，超过阈值立即抛异常。
5. 路径检查：拒绝包含 .. 的条目，防止路径穿越。

可运行的最小示例

保存为 safe_extract.py：

import sys
import zipfile
from pathlib import Path

MAX_TOTAL_SIZE = 100 * 1024 * 1024  # 占位值：100 MiB
MAX_SINGLE_SIZE = 20 * 1024 * 1024  # 占位值：20 MiB
MAX_RATIO = 100                     # 占位值：压缩比上限
CHUNK_SIZE = 64 * 1024

def safe_extract(zip_path: str, out_dir: str) -> None:
    out_path = Path(out_dir).resolve()
    out_path.mkdir(parents=True, exist_ok=True)
    total_written = 0

    with zipfile.ZipFile(zip_path, 'r') as zf:
        for info in zf.infolist():
            if info.is_dir() or info.filename.endswith('/'):
                continue

            # 路径安全检查
            if '..' in info.filename.split('/'):
                raise ValueError(f'可疑路径: {info.filename}')

            # 预检压缩比
            if info.compress_size > 0:
                ratio = info.file_size / info.compress_size
                if ratio > MAX_RATIO:
                    raise ValueError(
                        f'压缩比过高: {info.filename} ratio={ratio:.1f}'
                    )
            else:
                if info.file_size > 0:
                    raise ValueError(
                        f'压缩大小为 0 但声明解压大小非 0: {info.filename}'
                    )

            # 预检单文件和总大小
            if info.file_size > MAX_SINGLE_SIZE:
                raise ValueError(f'单文件过大: {info.filename}')
            if total_written + info.file_size > MAX_TOTAL_SIZE:
                raise ValueError(f'总解压大小超过限制: {info.filename}')

            target = (out_path / info.filename).resolve()
            try:
                target.relative_to(out_path)
            except ValueError:
                raise ValueError(f'非法路径: {info.filename}')

            target.parent.mkdir(parents=True, exist_ok=True)
            written = 0
            with zf.open(info, 'r') as src, open(target, 'wb') as dst:
                while True:
                    chunk = src.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    written += len(chunk)
                    total_written += len(chunk)
                    if written > MAX_SINGLE_SIZE:
                        raise ValueError(f'单文件解压超过限制: {info.filename}')
                    if total_written > MAX_TOTAL_SIZE:
                        raise ValueError('总解压大小超过限制')
                    if info.compress_size > 0 and written / info.compress_size > MAX_RATIO:
                        raise ValueError(f'实时压缩比超过限制: {info.filename}')
                    dst.write(chunk)

            if written != info.file_size:
                raise ValueError(f'实际解压大小与声明不一致: {info.filename}')

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print('用法: python safe_extract.py <zip路径> <输出目录>')
        sys.exit(1)
    safe_extract(sys.argv[1], sys.argv[2])

运行方法

准备一个测试 ZIP 文件，例如 input.zip，然后执行：

python safe_extract.py input.zip output_dir

如果 ZIP 正常且未超过阈值，文件会解压到 output_dir。如果某个条目的压缩比、单文件大小或总解压大小超限，脚本会抛出 ValueError 并停止。

边界与限制

• 阈值需要按业务调整。文本文件压缩比可能较高，MAX_RATIO 设得太低会误伤正常文件；图片、视频的压缩比通常接近 1。
• 不处理嵌套 ZIP。如果解压结果里还有 ZIP，需要递归检查或直接拒绝。
• 加密 ZIP 需要密码，zf.open() 会抛 RuntimeError，本文未处理。
• 路径检查拒绝了 ..，能覆盖常见 ZIP 穿越，但生产环境还应使用白名单扩展名、独立低权限用户运行。
• 失败时可能留下部分文件。生产环境建议先解压到临时目录，全部成功后再原子重命名。
• 实时压缩比依赖 compress_size 字段。如果该字段被恶意伪造，zipfile 可能读取失败或提前结束，实时累计写入量是最后防线。
• 总大小限制只约束当前进程的累计写入量，不替代磁盘配额；并发解压时仍需外部配额。
• 分块读取控制了内存占用，但 zlib 解压本身仍有内部缓冲，内存不会随文件线性增长。
• 不要用 extractall() 处理不可信 ZIP。

小结

安全解压不可信 ZIP 的核心是：逐条流式处理、预检压缩比与大小、实时累计校验、路径检查。示例代码可以直接作为工具函数的基础，再根据业务补充磁盘配额、超时和日志。