安全查找并提取硬盘中的压缩包:清点、隔离与校验

维护层,核查于 2026-09-01。 2019 年原文并非空白:正文只有一条递归执行的 Windows compact 命令,文末已完整保留。维护版处理标题的另一种常见理解——在硬盘中查找 ZIP/TAR 压缩包——但不建议盲目解压整个硬盘。

先确认“解压”指什么

两种完全不同的操作常被统称为解压:

  • NTFS 文件系统压缩: Windows 以压缩形式存储普通文件,应用程序仍然照常打开它。微软的 `compact` 文档说明,/u 移除这一属性,/s 则递归处理某个目录。
  • 压缩包提取: ZIP 或 TAR 系列容器保存带名称的成员,提取时把它们写入目标目录。Python 分别通过 `zipfile``tarfile` 支持这些格式。

compact 不会打开 ZIP 文件,提取 ZIP 也不会改变 NTFS 压缩属性。存档中的命令会递归改变当前驱动器上的文件系统存储方式。不要仅仅因为标题看起来像“解压所有压缩包”就运行它。处理 NTFS 压缩时,应先查询并测试一个经过审查的目录;大范围更改卷可能消耗大量空间,还会遇到受保护的操作系统文件。

为什么不能盲目解压整个硬盘

压缩包并不只是一袋无害文件。提取前至少要考虑:

  • 绝对路径、.. 路径段、Windows 驱动器路径和不可移植名称;
  • 符号链接、硬链接、设备、FIFO 或其他特殊 TAR 成员;
  • 重复名称、文件与目录冲突,以及仅大小写不同的冲突;
  • 声明了巨量输出或极端膨胀率的压缩包;
  • 会让无人值守任务变成密码处理问题的加密成员;
  • 两个压缩包写入相同目标名称;
  • 已经创建部分文件后才发生的失败。

Python 官方文档明确警告:提取前必须检查压缩包。即使 TAR 使用 data 提取过滤器,也无法阻止所有拒绝服务或恶意输入问题。安全流程应先清点,再提取到全新的隔离目录,同时施加资源限制,并在接受结果前完成校验。

保守的工作流程

  1. 不更改源驱动器,把输出与日志放在扫描树之外。
  2. 先寻找疑似压缩包的文件名,再根据文件内容确认 ZIP/TAR 类型,不只相信扩展名。
  3. 记录压缩包 SHA-256、成员数量、声明的未压缩大小、膨胀率以及全部拒绝原因。
  4. 审查 JSON Lines 清单。任何含有 problems 的记录都不得提取。
  5. 每次只选择一个相对压缩包路径来提取。
  6. 提取到全新的临时目录,其中不能预先存在链接或文件。
  7. 精确校验所有普通文件的路径与大小,写入状态记录,再把完成的目录重命名到最终位置。
  8. 保留原压缩包。只有完成备份和应用层检查后,才能把删除作为单独的保留策略决定。

这一流程有意比递归一行命令更慢,因为它能让部分失败、冲突和重试变得可观测。

Python 3.12+ 清点与提取参考工具

下面的纯标准库脚本支持 ZIP 和可读取的 TAR 变体。默认模式只做清点;--extract 每次只接受清单中的一个压缩包路径。它会拒绝链接及特殊文件、路径穿越和 Windows 驱动器路径、重复或仅大小写不同的名称、加密 ZIP 成员、超出限制的成员数或声明字节数,以及过高的膨胀率。

这些限制只是策略示例,并非通用安全值。面对不可信来源时应进一步降低阈值。请使用低权限账户,在受到适当约束的环境内处理不可信内容;此脚本只是护栏,不是恶意软件扫描器或安全边界。

<details>
<summary>显示 <code>archive_audit.py</code></summary>

#!/usr/bin/env python3
import argparse
import hashlib
import json
import os
import re
import shutil
import stat
import tarfile
import tempfile
import zipfile
from pathlib import Path, PurePosixPath, PureWindowsPath

SUFFIXES = (
    ".zip", ".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2",
    ".tar.xz", ".txz", ".gz", ".bz2", ".xz",
)
WINDOWS_RESERVED = {
    "con", "prn", "aux", "nul",
    *(f"com{i}" for i in range(1, 10)),
    *(f"lpt{i}" for i in range(1, 10)),
}


def digest(path):
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def normalized_name(raw):
    if not raw or "" in raw:
        raise ValueError("empty or NUL-containing member name")
    windows = PureWindowsPath(raw)
    posix = PurePosixPath(raw.replace("\", "/"))
    if windows.drive or windows.is_absolute() or posix.is_absolute():
        raise ValueError(f"absolute or drive-qualified path: {raw!r}")
    if ".." in posix.parts:
        raise ValueError(f"parent traversal: {raw!r}")
    parts = tuple(part for part in posix.parts if part not in ("", "."))
    if not parts:
        return None
    for part in parts:
        stem = part.rstrip(" .").split(".", 1)[0].casefold()
        if ":" in part or part.endswith((" ", ".")) or stem in WINDOWS_RESERVED:
            raise ValueError(f"non-portable path component: {raw!r}")
    return "/".join(parts)


def validate_members(raw_members):
    members = []
    exact = {}
    folded = {}
    problems = []
    for raw, kind, size in raw_members:
        try:
            name = normalized_name(raw)
        except ValueError as exc:
            problems.append(str(exc))
            continue
        if name is None:
            continue
        if kind not in {"file", "dir"}:
            problems.append(f"unsupported {kind}: {raw!r}")
            continue
        if name in exact:
            if kind == exact[name]["kind"] == "dir":
                continue
            problems.append(f"duplicate or file/type collision: {raw!r}")
            continue
        folded_name = name.casefold()
        if folded_name in folded and folded[folded_name] != name:
            problems.append(
                f"case-insensitive collision: {folded[folded_name]!r} / {name!r}"
            )
            continue
        member = {"name": name, "kind": kind, "size": size}
        exact[name] = member
        folded[folded_name] = name
        members.append(member)

    for member in members:
        for parent in PurePosixPath(member["name"]).parents:
            parent_name = parent.as_posix()
            if parent_name == ".":
                break
            if exact.get(parent_name, {}).get("kind") == "file":
                problems.append(
                    f"file blocks child path: {parent_name!r} / {member['name']!r}"
                )
    return members, sorted(set(problems))


def inspect_archive(path, kind, limits):
    raw_members = []
    encrypted = False
    if kind == "zip":
        with zipfile.ZipFile(path) as archive:
            for info in archive.infolist():
                mode = (info.external_attr >> 16) & 0xFFFF
                if info.is_dir():
                    member_kind = "dir"
                elif stat.S_ISLNK(mode):
                    member_kind = "symlink"
                elif mode and stat.S_IFMT(mode) not in (0, stat.S_IFREG):
                    member_kind = "special file"
                else:
                    member_kind = "file"
                encrypted |= bool(info.flag_bits & 0x1)
                raw_members.append((info.filename, member_kind, info.file_size))
    else:
        with tarfile.open(path, "r:*") as archive:
            for info in archive.getmembers():
                if info.isfile():
                    member_kind = "file"
                elif info.isdir():
                    member_kind = "dir"
                elif info.issym() or info.islnk():
                    member_kind = "symlink or hard link"
                else:
                    member_kind = "special file"
                raw_members.append((info.name, member_kind, info.size))

    members, problems = validate_members(raw_members)
    files = [member for member in members if member["kind"] == "file"]
    unpacked = sum(member["size"] for member in files)
    archive_bytes = path.stat().st_size
    ratio = unpacked / max(archive_bytes, 1)
    if encrypted:
        problems.append("encrypted ZIP: password input is deliberately unsupported")
    raw_member_count = len(raw_members)
    if raw_member_count > limits.max_members:
        problems.append(
            f"member limit exceeded: {raw_member_count} > {limits.max_members}"
        )
    if unpacked > limits.max_bytes:
        problems.append(f"byte limit exceeded: {unpacked} > {limits.max_bytes}")
    if ratio > limits.max_ratio:
        problems.append(f"expansion ratio exceeded: {ratio:.1f} > {limits.max_ratio}")
    return {
        "archive_bytes": archive_bytes,
        "file_count": len(files),
        "member_count": raw_member_count,
        "accepted_member_count": len(members),
        "unpacked_bytes": unpacked,
        "expansion_ratio": round(ratio, 2),
        "problems": sorted(set(problems)),
        "members": members,
    }


def verify_tree(root, members):
    expected = {
        member["name"]: member["size"]
        for member in members if member["kind"] == "file"
    }
    actual = {}
    problems = []
    for base, dirs, files in os.walk(root, followlinks=False):
        base_path = Path(base)
        kept_dirs = []
        for name in dirs:
            path = base_path / name
            if path.is_symlink():
                problems.append(f"extracted symlink: {path.relative_to(root)}")
            else:
                kept_dirs.append(name)
        dirs[:] = kept_dirs
        for name in files:
            path = base_path / name
            relative = path.relative_to(root).as_posix()
            if relative == ".archive-state.json":
                continue
            if path.is_symlink() or not path.is_file():
                problems.append(f"non-regular output: {relative}")
            else:
                actual[relative] = path.stat().st_size
    if actual != expected:
        problems.append("output file names or sizes differ from the reviewed inventory")
    return problems


def extract_one(plan, output):
    if plan["problems"]:
        raise RuntimeError("selected archive did not pass preflight")
    destination = output / plan["destination"]
    if destination.exists():
        state_path = destination / ".archive-state.json"
        state = json.loads(state_path.read_text("utf-8")) if state_path.is_file() else {}
        problems = verify_tree(destination, plan["members"])
        if state.get("sha256") == plan["sha256"] and not problems:
            return {"status": "already-verified", "destination": str(destination)}
        raise FileExistsError(f"refusing to merge with existing destination: {destination}")

    output.mkdir(parents=True, exist_ok=True)
    required = plan["unpacked_bytes"] + max(plan["unpacked_bytes"] // 10, 64 << 20)
    if shutil.disk_usage(output).free < required:
        raise RuntimeError("insufficient free space including safety margin")
    if digest(plan["path"]) != plan["sha256"]:
        raise RuntimeError("archive changed after inventory")

    with tempfile.TemporaryDirectory(prefix=".partial-", dir=output) as temporary:
        temporary_path = Path(temporary)
        if plan["kind"] == "zip":
            with zipfile.ZipFile(plan["path"]) as archive:
                archive.extractall(temporary_path)
        else:
            if not hasattr(tarfile, "data_filter"):
                raise RuntimeError("safe tar extraction requires the data filter")
            with tarfile.open(plan["path"], "r:*") as archive:
                archive.extractall(temporary_path, filter="data")
        if digest(plan["path"]) != plan["sha256"]:
            raise RuntimeError("archive changed during extraction")
        problems = verify_tree(temporary_path, plan["members"])
        if problems:
            raise RuntimeError("; ".join(problems))
        state = {
            "source": plan["relative_path"],
            "sha256": plan["sha256"],
            "file_count": plan["file_count"],
            "unpacked_bytes": plan["unpacked_bytes"],
        }
        (temporary_path / ".archive-state.json").write_text(
            json.dumps(state, indent=2, sort_keys=True) + "n", "utf-8"
        )
        temporary_path.rename(destination)
    return {"status": "extracted-and-verified", "destination": str(destination)}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--source", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--manifest", required=True, type=Path)
    parser.add_argument(
        "--extract", metavar="RELATIVE_ARCHIVE",
        help="extract exactly one reviewed archive; default is inventory only",
    )
    parser.add_argument("--max-members", type=int, default=100_000)
    parser.add_argument("--max-bytes", type=int, default=50 * 1024**3)
    parser.add_argument("--max-ratio", type=float, default=200.0)
    args = parser.parse_args()

    if args.max_members < 1 or args.max_bytes < 1 or args.max_ratio <= 0:
        parser.error("resource limits must be positive")

    source = args.source.resolve()
    output = args.output.resolve()
    manifest = args.manifest.resolve()
    if not source.is_dir():
        raise SystemExit("--source must be an existing directory")
    if output == source or output.is_relative_to(source):
        raise SystemExit("--output must be outside --source")
    if manifest.is_relative_to(source):
        raise SystemExit("--manifest must be outside --source")

    plans = {}
    records = []
    walk_errors = []

    def on_walk_error(error):
        walk_errors.append(f"{type(error).__name__}: {error}")

    for base, dirs, files in os.walk(source, followlinks=False, onerror=on_walk_error):
        base_path = Path(base)
        dirs[:] = [name for name in dirs if not (base_path / name).is_symlink()]
        for name in files:
            path = base_path / name
            if path.is_symlink() or not name.casefold().endswith(SUFFIXES):
                continue
            relative = path.relative_to(source).as_posix()
            try:
                if zipfile.is_zipfile(path):
                    kind = "zip"
                elif tarfile.is_tarfile(path):
                    kind = "tar"
                else:
                    records.append({
                        "record_type": "archive",
                        "relative_path": relative,
                        "kind": "unsupported-or-damaged",
                        "problems": ["candidate suffix but no supported ZIP/TAR signature"],
                    })
                    continue
                sha256 = digest(path)
                inspected = inspect_archive(path, kind, args)
                label = re.sub(r"[^A-Za-z0-9._-]+", "_", relative).strip("._")
                path_token = hashlib.sha256(relative.encode("utf-8")).hexdigest()[:12]
                destination = f"{(label or 'archive')[:60]}-{path_token}-{sha256[:12]}"
                plan = {
                    "record_type": "archive",
                    "relative_path": relative,
                    "kind": kind,
                    "sha256": sha256,
                    "destination": destination,
                    "path": path,
                    **inspected,
                }
                plans[relative] = plan
                records.append({key: value for key, value in plan.items()
                                if key not in {"path", "members"}})
            except (OSError, tarfile.TarError, zipfile.BadZipFile) as exc:
                records.append({
                    "record_type": "archive",
                    "relative_path": relative,
                    "kind": "unreadable",
                    "problems": [f"{type(exc).__name__}: {exc}"],
                })

    summary = {
        "record_type": "scan-summary",
        "source": str(source),
        "archives": len(records),
        "walk_errors": walk_errors,
        "mode": "extract-one" if args.extract else "inventory-only",
    }
    manifest.parent.mkdir(parents=True, exist_ok=True)
    with manifest.open("x", encoding="utf-8") as stream:
        for record in (*records, summary):
            stream.write(json.dumps(record, sort_keys=True) + "n")
            print(json.dumps(record, sort_keys=True))

    if args.extract:
        if walk_errors:
            raise SystemExit("scan was incomplete; refusing extraction")
        plan = plans.get(args.extract)
        if plan is None:
            raise SystemExit("--extract must exactly match a supported relative_path")
        print(json.dumps(extract_one(plan, output), sort_keys=True))


if __name__ == "__main__":
    main()

</details>

第一步只做清点,不提取

把脚本保存为 archive_audit.py。显式指定三个路径,并确保它们互不包含:

python3 archive_audit.py 
  --source /path/to/mounted-drive 
  --output /path/to/separate-output/extracted 
  --manifest /path/to/private-logs/inventory-2026-09-01.jsonl

不使用 --extract 时,脚本不会创建提取目录。它会扫描常见的 ZIP/TAR 后缀,但通过 zipfile.is_zipfile()tarfile.is_tarfile() 确认类型。这样既不盲信改过的扩展名,也不必昂贵地尝试解析驱动器上的每一个普通文件。

清单以独占创建模式打开,因此不会静默覆盖已有日志。它包含压缩包相对名称、源根目录信息、哈希、声明大小、预期目标名称和拒绝详情;如果文件名或挂载路径敏感,请将日志作为私密资料保存。

审查空间、成员与拒绝原因

每个受支持的压缩包至少要检查:

  • problems:提取前必须为空列表;
  • sha256:标识被检查的确切压缩包字节;
  • file_countunpacked_bytes:声明的普通文件工作量;
  • expansion_ratio:声明输出大小除以压缩包大小;
  • destination:净化后的名称加相对路径哈希与内容哈希前缀;
  • 最终摘要中的 walk_errors:任何扫描错误都会使清单不完整,因而不能安全提取。

默认上限是每个压缩包 100,000 个成员、50 GiB 声明输出和 200:1 膨胀率。这些值并不保证提取安全。压缩包头可以恶意伪造,文件系统元数据会占额外空间,检查后其他进程也可能继续消耗空间。因此提取路径还要求“声明输出 + 10% 或 64 MiB 中较大者”的可用空间,并通过 Python 的 `shutil.disk_usage()` 查询。

每次只提取一个已审查压缩包

从无问题的清单记录中复制精确的 relative_path,并使用新的清单文件名:

python3 archive_audit.py 
  --source /path/to/mounted-drive 
  --output /path/to/separate-output/extracted 
  --manifest /path/to/private-logs/extract-photos-2026-09-01.jsonl 
  --extract 'relative/path/photos.zip'

脚本在提取前后都会重新计算源文件哈希,在全新的临时目录中工作,并且绝不合并到已有目标目录。TAR 提取会显式要求 filter="data";如果该 API 不存在就直接失败。随后,它会把每个普通输出文件的路径与大小同已审查清单比较,全部通过后才把目录重命名到最终位置。

源目录树永远不会被写入。输出必须位于源目录之外,因此新提取出的压缩包不会反馈到同一次扫描。每个压缩包相对路径都有按路径与内容共同寻址的隔离目标目录;即使两个不同名称的压缩包字节完全相同,也不会发生冲突。

密码与加密压缩包

该工具有意阻止加密 ZIP 成员。不要把明文密码加入命令行、脚本、JSONL 清单、Shell 历史、进程列表或源代码仓库。应先确认压缩包来源,再使用经过审查的格式专用工具,以交互方式或经批准的秘密管理器取得密码,同时继续提取到隔离目标。

Python 的 zipfile 文档指出,其解密速度极慢。更重要的是,密码成功只能解密字节,并不能证明成员路径、膨胀大小或内容安全。任何独立的加密压缩包流程都应重新执行等效的清点与校验控制。

日志、幂等性与校验边界

成功后,目标目录内会有 .archive-state.json,记录源相对路径、SHA-256、文件数和声明输出字节数。对未变更的字节重复执行时,脚本不会覆盖文件:它会校验现有目录并报告 already-verified。压缩包一旦变化,按哈希生成的目标名称也会变化;已有目录若不完整或被修改,工具会拒绝合并。

路径与大小一致是有用的非破坏性结构检查,但不能证明文档真实或可执行文件无害。SHA-256 能标识被检查的压缩包,却不能认证发布者。若发布者提供签名或可信校验和,应另行验证;还要按本地安全策略扫描内容,并只在隔离查看器中打开风险格式。提取流程不应自动执行任何程序。

支持范围与已知限制

这个参考工具有意缩小范围:

  • 支持已安装 Python 能读取的标准 ZIP 与 TAR,包括常见的 gzip/bzip2/xz 压缩 TAR。
  • 独立 .gz.bz2.xz、分卷 ZIP、RAR、7z,或需要缺失的 Python 可选压缩模块的格式,会被记录或失败;工具不会猜测并换用其他解压器。
  • 候选发现从常见后缀开始,因此名称完全无关的有效压缩包不会自动找到。应单独调查这类文件,而不是探测驱动器上的每个字节序列。
  • 链接和特殊 TAR 文件系统对象会被拒绝,不会重建。
  • 资源限制只能降低风险;面对恶意输入时仍需限制进程权限、CPU、内存、文件系统配额和时间。

这些约束正是安全特性:不支持或含糊的输入应停下来审查,而不是静默选择更强大的提取模式。

一次性环境中的验证证据

内嵌脚本已在一次性本地目录中用 Python 3.12.3 完成语法检查和运行测试。测试集包括普通 ZIP、同时含 A.txta.txt 的 ZIP、含符号链接的 TAR,以及成员名为 ../escape.txt 的 TAR。

  • 清点模式没有创建任何提取结果。
  • 普通 ZIP 提取到了隔离的哈希命名目录,并通过路径/大小校验。
  • 对同一目标重复执行时返回 already-verified,没有覆盖文件。
  • 大小写冲突、符号链接和路径穿越压缩包都在提取前被拒绝。

验证期间没有修改真实驱动器、生产路径或网络压缩包。

官方一手文档

---

2019 年英文原文存档(逐字保留)

原始导出并非空白。以下是其完整可见正文,发布及最后修改日期均为 2019-06-13。仅为仓库格式规范化了不可见的行尾空白。它作为来源记录保留,并不是更改整个驱动器的建议。


First move present working directory to root of each drive, then type

C:
compact /u /s /a /q /i

Leave a Reply