---
title: "Python 怎么处理用户上传的 ZIP：给 CSV 导入加上文件与解压预算"
date: 2026-11-12
slug: python-zip-import-boundaries
canonical: https://kernelranch.com/blog/2026/11/python-zip-import-boundaries
markdown_url: https://kernelranch.com/blog/2026/11/python-zip-import-boundaries.md
tags: ["python", "security", "zip", "uploads"]
---后台导入只需要 `manifest.json` 和 `data.csv`。把上传的 ZIP 整包解压，会让用户提供的路径和压缩内容影响你的文件系统。先约定包里允许出现的两个名字，验证目录，再将文件流写入新建的临时目录。两个成员都校验通过后，才把暂存文件交给解析器。

## 1. 按导入格式定义接收规则

| 项目 | 本例的约定 |
|---|---|
| 目录 | 根目录恰好包含 `manifest.json` 和 `data.csv`，大小写敏感 |
| 文件类型 | 普通文件，允许 ZIP_STORED 与 ZIP_DEFLATED，拒绝加密成员和 Unix 软链接标记 |
| 压缩包大小 | 至多 2 MiB，在打开 ZIP 前检查 |
| 解压后大小 | 清单至多 64 KiB，CSV 至多 4 MiB，目录预检和分块读取都检查 |
| 失败语义 | 任一成员失败，整包不进入业务解析，临时目录由上下文清理 |

固定入口名能让服务自己决定输出文件路径。`../data.csv`、`dir/data.csv` 和 `DATA.CSV` 均不符合这份导入协议。其他产品若需要支持任意目录，须另外处理路径规范化、大小写碰撞、保留文件名和已有符号链接等问题。

## 2. 按 ZipInfo 打开成员，写到服务指定的路径

保存为 `zip_import_lab.py`，执行 `python zip_import_lab.py`。`stage_import()` 是可复用的暂存函数。`with` 块中完成 JSON、CSV 和业务校验，验证结束后再提交数据库事务。

```python
import os
import stat
import warnings
from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
from zipfile import BadZipFile, ZIP_DEFLATED, ZIP_STORED, ZipFile, ZipInfo


UPLOAD_LIMIT = 2 * 1024 * 1024
LIMITS = {"manifest.json": 64 * 1024, "data.csv": 4 * 1024 * 1024}
CHUNK_SIZE = 64 * 1024


@contextmanager
def stage_import(upload):
    with Path(upload).open("rb") as source:
        if os.fstat(source.fileno()).st_size > UPLOAD_LIMIT:
            raise ValueError("upload too large")
        with ZipFile(source) as archive, TemporaryDirectory() as tmp:
            root = Path(tmp)
            entries = {}
            for entry in archive.infolist():
                # orig_filename 保留 ZipInfo 规范化前的名称。
                name = entry.orig_filename
                if name not in LIMITS or name in entries:
                    raise ValueError("unexpected or duplicate member")
                mode = stat.S_IFMT(entry.external_attr >> 16) if entry.create_system == 3 else 0
                if mode not in (0, stat.S_IFREG) or entry.external_attr & 0x10:
                    raise ValueError("member must be a regular file")
                if entry.flag_bits & 1 or entry.compress_type not in (ZIP_STORED, ZIP_DEFLATED):
                    raise ValueError("unsupported ZIP member")
                if entry.file_size > LIMITS[name]:
                    raise ValueError("expanded member too large")
                entries[name] = entry
            if entries.keys() != LIMITS.keys():
                raise ValueError("required member missing")

            for name, limit in LIMITS.items():
                written = 0
                with archive.open(entries[name]) as member, (root / name).open("xb") as output:
                    while chunk := member.read(CHUNK_SIZE):
                        written += len(chunk)
                        if written > limit:
                            raise ValueError("expanded member too large")
                        output.write(chunk)
            yield root


def write_fixture(path, csv_name, payload, *, compression=ZIP_DEFLATED, extra=None):
    with ZipFile(path, "w", compression=compression) as archive:
        archive.writestr("manifest.json", '{"version":1}')
        archive.writestr(csv_name, payload)
        if extra is not None:
            with warnings.catch_warnings():
                warnings.filterwarnings("ignore", message="Duplicate name:")
                archive.writestr(extra, "duplicate")


def expect_rejection(path, label, error_type, message):
    try:
        with stage_import(path):
            raise AssertionError("rejected archive reached the parser")
    except error_type as error:
        assert message in str(error)
        print(f"{label}: rejected")


if __name__ == "__main__":
    with TemporaryDirectory() as tmp:
        upload = Path(tmp) / "upload.zip"
        csv = b"id,name\n1,Ada\n"
        write_fixture(upload, "data.csv", csv)
        with stage_import(upload) as staged:
            assert (staged / "data.csv").read_bytes() == csv
        assert not staged.exists()
        print("normal: accepted; staging cleaned")

        write_fixture(upload, "../data.csv", csv)
        expect_rejection(upload, "traversal", ValueError, "unexpected or duplicate")
        write_fixture(upload, "data.csv", csv, extra="data.csv")
        expect_rejection(upload, "duplicate", ValueError, "unexpected or duplicate")

        link = ZipInfo("data.csv")
        link.create_system = 3
        link.external_attr = (stat.S_IFLNK | 0o777) << 16
        write_fixture(upload, link, b"../private.csv")
        expect_rejection(upload, "symlink", ValueError, "regular file")

        write_fixture(upload, "data.csv", b"x" * (LIMITS["data.csv"] + 1))
        assert upload.stat().st_size < UPLOAD_LIMIT
        expect_rejection(upload, "expanded size", ValueError, "expanded member too large")

        write_fixture(upload, "data.csv", csv, compression=ZIP_STORED)
        upload.write_bytes(upload.read_bytes().replace(csv, b"X" + csv[1:], 1))
        expect_rejection(upload, "bad CRC", BadZipFile, "Bad CRC")
```

本地实测，正常包可读取并在退出时清理。路径穿越、重名、软链接和解压大小超限在预检阶段被拒绝，篡改过正文的 ZIP 在读 `data.csv` 时触发 CRC 错误。五种异常包均未进入解析器。

## 3. 把预算加在对应的入口

- **上传入口限制实际接收字节。** 代码假定传入已上传完毕、由服务控制的本地文件。HTTP 接收层还需要在写入临时文件时计数，达到限制就停止接收。仅检查声明的 `Content-Length` 会遗漏部分请求。
- **归档读取分块并读到 EOF。** 目录声明大小可用于提前拒绝。代码还累计实际写入的字节，直到完整读取成员，让 ZIP 库有机会检查 CRC。CRC 用于发现损坏，内容来源需要独立鉴权。
- **业务解析单独限额。** CSV 还要检查行数、列数、字段大小与预期表头，清单也要校验结构。服务端给导入任务设并发上限，避免多个各自合规的上传同时耗尽磁盘或工作线程。

## 小结

导入格式已知时，逐项打开所需成员，把名字和大小变成接口契约。解压先写私有临时目录，所有成员校验完成后才交给业务层。

## 数据来源与检索时间

检索时间 2026-11-12。

| 来源 | URL | 对应内容 |
|---|---|---|
| Python 3.11 `zipfile` | https://docs.python.org/3.11/library/zipfile.html | `ZipInfo`、按成员打开、归档资源限制 |
| CPython 3.11.9 `zipfile` 实现 | https://github.com/python/cpython/blob/v3.11.9/Lib/zipfile.py | 原始文件名、ZIP 流读取和 CRC 校验 |
| Microsoft Windows 文件命名规范 | https://learn.microsoft.com/en-us/windows/win32/fileio/naming-a-file | 任意目录解包涉及的保留名称与路径差异 |

## 核验说明

- 在 Python 3.11.9 下生成并读取六类本地 ZIP，逐个断言接收或拒绝结果。
- 示例覆盖暂存和归档校验。HTTP 上传流、实际 CSV 字段校验与数据库写入由导入服务实现。