python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python zipfile读写压缩包

Python利用zipfile读写压缩包的实战指南

作者：codedevin

处理用户上传的 zip、批量打包导出文件、解压第三方数据包,zipfile 是绕不开的标准库,这篇把 zipfile 的正确姿势讲清楚:怎么读、怎么写、怎么流式处理不爆内存,以及解压不可信压缩包时必须做的安全防护,需要的朋友可以参考下

处理用户上传的 zip、批量打包导出文件、解压第三方数据包——zipfile 是绕不开的标准库。但两个坑几乎人人踩过:一个是 read() 一把把整个大文件读进内存,几百 MB 的压缩项直接把服务干 OOM;另一个更危险,直接用 extractall 解压不可信的 zip,一个构造好的压缩包能把文件写到 /etc/ 或者你的项目源码目录里,这就是 Zip Slip 路径穿越漏洞。

这篇把 zipfile 的正确姿势讲清楚:怎么读、怎么写、怎么流式处理不爆内存,以及解压不可信压缩包时必须做的安全防护。

基本读写:先跑通再说

创建一个 zip,把几个文件写进去:

import zipfile

# 'w' 覆盖写;ZIP_DEFLATED 才会真正压缩,默认 ZIP_STORED 只打包不压缩
with zipfile.ZipFile("out.zip", "w", zipfile.ZIP_DEFLATED) as zf:
    zf.write("report.pdf")               # 按原路径存进去
    zf.write("data/a.csv", "a.csv")      # 第二个参数 arcname:改成压缩包内的路径
    zf.writestr("note.txt", "直接写字符串,不用先落地文件")  # 内存内容直接入包

compression 参数很多人忽略:不传的话默认 ZIP_STORED,只是把文件原样塞进去、体积一点不减。要压缩必须显式指定 ZIP_DEFLATED(通用)或 ZIP_LZMA(压得更狠但慢)。writestr 尤其好用,把内存里生成的内容(比如导出的 CSV 字符串)直接写进包,省去先写临时文件再读的麻烦。

读取和查看内容:

with zipfile.ZipFile("out.zip", "r") as zf:
    print(zf.namelist())          # 包内所有条目名列表
    for info in zf.infolist():    # 更详细:每个条目的元信息
        print(info.filename, info.file_size, info.compress_size)

    data = zf.read("note.txt")    # 返回 bytes,整个条目一次读进内存
    print(data.decode("utf-8"))

read() 简单,但注意它把整个条目一次性读进内存——对小文件无所谓,大文件就是隐患,下面细讲。

坑一:read()大文件爆内存,改用流式

假设包里有个 2GB 的日志文件,你要逐行处理:

# ❌ 2GB 条目一次读进内存,直接 OOM
with zipfile.ZipFile("logs.zip") as zf:
    content = zf.read("huge.log")   # 危险
    for line in content.splitlines():
        handle(line)

正确做法是用 zf.open() 拿到一个文件式对象,像普通文件一样流式读,内存里始终只有一小块:

import io

with zipfile.ZipFile("logs.zip") as zf:
    # open 返回二进制流,不会一次性解压全部
    with zf.open("huge.log") as raw:
        # 包裹成 TextIOWrapper 就能按行迭代,编码自己指定
        for line in io.TextIOWrapper(raw, encoding="utf-8"):
            handle(line.rstrip("\n"))

zf.open() 返回的是一个边解压边读的流,配合 TextIOWrapper 就能按行迭代,内存占用和文件大小无关。同理,解压到磁盘时也别用 read() 再自己写,用 shutil.copyfileobj 分块拷贝:

import shutil

with zipfile.ZipFile("logs.zip") as zf:
    with zf.open("huge.log") as src, open("huge.log", "wb") as dst:
        # 分块拷贝,内存里只留一个缓冲区
        shutil.copyfileobj(src, dst, length=1024 * 1024)  # 1MB 一块

记住:只要条目可能很大,就用 zf.open() 流式处理,别碰 zf.read()。

坑二:Zip Slip 路径穿越,extractall不可信包等于开后门

这是最危险的一个。你可能觉得解压很安全,一行 extractall 搞定:

# ❌ 对不可信的上传 zip 这样写,等于把写文件的权力交给了攻击者
with zipfile.ZipFile(uploaded_zip) as zf:
    zf.extractall("/data/uploads/")

问题在于:zip 内条目的文件名可以是任意字符串,包括 ../../../etc/cron.d/evil 或者绝对路径 /etc/passwd。攻击者构造这样一个包,解压时文件就被写到了 /data/uploads/ 之外——覆盖系统配置、往定时任务目录塞脚本、篡改你的项目代码,都成为可能。这就是 Zip Slip(CVE 遍地都是,Java、Python、JS 生态全中过招)。

先看攻击是怎么造出来的,理解了才好防:

# 演示:构造一个带路径穿越条目的恶意包(仅用于理解攻击)
with zipfile.ZipFile("evil.zip", "w") as zf:
    zf.writestr("../../../../tmp/pwned.txt", "escaped!")  # 条目名带 ../

好消息是:Python 3.6.2 起,extractall / extract 已经会把带 .. 和绝对路径的条目名做「净化」,把它们重定向到目标目录内,不会真的穿越。但——这个净化只防最经典的 ../,不能全信,而且不同版本行为有差异,符号链接、特殊字符仍可能有边界情况。生产环境处理不可信压缩包,应当自己显式校验每一个解压目标路径:

import os
import zipfile

def safe_extract(zip_path: str, dest_dir: str) -> None:
    dest_dir = os.path.realpath(dest_dir)  # 解析成绝对真实路径,消除符号链接
    with zipfile.ZipFile(zip_path) as zf:
        for member in zf.namelist():
            # 拼出目标路径,再算真实绝对路径
            target = os.path.realpath(os.path.join(dest_dir, member))
            # 核心校验:目标必须仍在 dest_dir 之内,否则就是路径穿越
            if not target.startswith(dest_dir + os.sep) and target != dest_dir:
                raise ValueError(f"检测到路径穿越,拒绝解压条目:{member}")
        # 全部校验通过才解压
        zf.extractall(dest_dir)

# safe_extract("evil.zip", "/data/uploads")  # 会抛 ValueError,拒之门外

关键是用 os.path.realpath 把拼出来的目标解析成真实绝对路径(它会展开 .. 和符号链接),再判断结果是否仍以 dest_dir 为前缀。只要跳出了目标目录,一律拒绝。这层校验不依赖标准库版本,自己守住最踏实。

坑三:zip 里的中文文件名乱码

Windows 下用资源管理器打的 zip,文件名常用 GBK 编码,而 zip 规范默认 CP437,Python 解出来就是一串乱码:

with zipfile.ZipFile("windows.zip") as zf:
    for info in zf.infolist():
        name = info.filename
        # 条目没标记 UTF-8 时(flag_bits 第 11 位为 0),往往是 CP437 误读的 GBK
        if not info.flag_bits & 0x800:
            # 先按 CP437 编码回 bytes,再用 GBK 正确解码
            name = info.filename.encode("cp437").decode("gbk", errors="replace")
        print(name)

判断依据是 flag_bits & 0x800(通用位标记的第 11 位):置位表示文件名是 UTF-8,不用处理;没置位且出现乱码,基本就是 CP437 误读了 GBK,encode("cp437").decode("gbk") 绕回去即可。

追加与防「zip 炸弹」

往已有 zip 追加文件用 'a' 模式:

with zipfile.ZipFile("out.zip", "a", zipfile.ZIP_DEFLATED) as zf:
    zf.writestr("added.txt", "追加进来的内容")

处理不可信包还要防 zip 炸弹——一个几十 KB 的包解压出几十 GB,撑爆磁盘。解压前用 infolist() 检查累计的 file_size,超阈值就拒绝:

MAX_TOTAL = 500 * 1024 * 1024  # 解压上限 500MB
with zipfile.ZipFile(untrusted) as zf:
    total = sum(info.file_size for info in zf.infolist())
    if total > MAX_TOTAL:
        raise ValueError(f"解压后体积 {total} 超限,疑似 zip 炸弹")

小结

以上就是Python利用zipfile读写压缩包的实战指南的详细内容，更多关于Python zipfile读写压缩包的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:
阅读全文