功能说明
这段代码在解压 tar.gz 前逐项审查成员,阻止经典的“zip slip / tar traversal”攻击:恶意归档可以包含 ../../etc/passwd 或绝对路径,让解压过程把文件写到目标目录之外。代码还拒绝符号链接、硬链接和设备文件,只允许普通文件与目录。
它适合处理用户上传的归档、插件包或构建产物。示例先用内存构造一个正常归档和一个恶意归档,验证正常文件被解压、穿越路径被拒绝且目标目录外没有产生文件。
关键代码
safe_members 用 PurePosixPath 解析归档内的 POSIX 路径,检查是否为绝对路径、是否包含 .. 段,并拒绝 issym()、islnk() 以及非文件非目录的成员。
随后把 root / path 解析成真实路径,确认 root 位于 target.parents 中,防止通过特殊路径绕过字符串检查。只有通过审查的成员才会传给 extractall,因此不会解压任何未经验证的条目。
使用注意
这类检查存在检查与使用之间的时间窗口,在高安全场景应把归档解压到全新的临时目录,并限制进程权限。代码没有限制单个文件大小、总解压大小或文件数量,压缩炸弹仍可能耗尽磁盘;应增加配额。
大小写不敏感的文件系统、Windows 保留设备名、Unicode 规范化都可能带来额外风险。Python 3.12 的 filter="data" 提供官方防护,可作为纵深防御叠加使用。归档中的权限位和属主信息也应谨慎处理。
验证方式
正常归档解压后返回 ["a.txt", "nested/b.txt"],并读取到 beta 内容。恶意归档 ../evil.txt 会抛出 ValueError,随后断言父目录下不存在 evil.txt。可以再加入绝对路径 /tmp/x、符号链接成员和 FIFO 成员,确认它们分别被拒绝。
代码示例
from __future__ import annotations
import io
import tarfile
import tempfile
from pathlib import Path, PurePosixPath
def safe_members(archive: tarfile.TarFile, destination: Path) -> list[tarfile.TarInfo]:
root = destination.resolve()
accepted: list[tarfile.TarInfo] = []
for member in archive.getmembers():
path = PurePosixPath(member.name)
if path.is_absolute() or ".." in path.parts:
raise ValueError(f"unsafe archive path: {member.name}")
if member.issym() or member.islnk():
raise ValueError(f"links are not allowed: {member.name}")
if not (member.isfile() or member.isdir()):
raise ValueError(f"unsupported member type: {member.name}")
target = (root / path).resolve()
if root not in target.parents and target != root:
raise ValueError(f"member escapes destination: {member.name}")
accepted.append(member)
return accepted
def extract_archive(data: bytes, destination: Path) -> list[str]:
destination.mkdir(parents=True, exist_ok=True)
with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as archive:
members = safe_members(archive, destination)
archive.extractall(destination, members=members)
return sorted(
path.relative_to(destination).as_posix()
for path in destination.rglob("*")
if path.is_file()
)
def build_archive(entries: list[tuple[str, bytes]]) -> bytes:
buffer = io.BytesIO()
with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
for name, payload in entries:
info = tarfile.TarInfo(name)
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))
return buffer.getvalue()
def main() -> None:
with tempfile.TemporaryDirectory() as raw_root:
root = Path(raw_root)
destination = root / "extract"
archive = build_archive([("a.txt", b"alpha"), ("nested/b.txt", b"beta")])
files = extract_archive(archive, destination)
print(files)
assert files == ["a.txt", "nested/b.txt"]
assert (destination / "nested" / "b.txt").read_bytes() == b"beta"
malicious = build_archive([("../evil.txt", b"owned")])
try:
extract_archive(malicious, root / "unsafe")
except ValueError as exc:
print(f"rejected: {exc}")
else:
raise AssertionError("traversal archive was accepted")
assert not (root / "evil.txt").exists()
if __name__ == "__main__":
main()