From 6cb8a4d43073dbf6cfb7e5d8382c0435f3210f6a Mon Sep 17 00:00:00 2001 From: Shurong Cao <170531907+CAOShurong@users.noreply.github.com> Date: Sun, 9 Aug 2026 10:38:32 +0800 Subject: [PATCH] Harden package archive extraction --- CHANGELOG.md | 11 +++++ src/willitbreak/__init__.py | 2 +- src/willitbreak/fetch.py | 53 +++++++++++++++++++----- tests/test_fetch.py | 81 +++++++++++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 12 deletions(-) create mode 100644 tests/test_fetch.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9fb38a1..96fa545 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.1.1] - 2026-08-09 + +### Security + +- Require every downloaded ZIP or source-distribution member to remain inside + the exact cache destination by path component, preventing a crafted archive + from writing into a same-prefix sibling directory. +- Reject non-file and non-directory tar members, including links, devices, and + FIFOs, before extraction. + ## [0.1.0] - 2026-08-03 First release. @@ -25,4 +35,5 @@ First release. and colour handling that honours `NO_COLOR`. - Exit code 2 for a breaking upgrade, 1 for the tool itself failing. +[0.1.1]: https://github.com/CAOShurong/willitbreak/compare/v0.1.0...v0.1.1 [0.1.0]: https://github.com/CAOShurong/willitbreak/releases/tag/v0.1.0 diff --git a/src/willitbreak/__init__.py b/src/willitbreak/__init__.py index 4f3b1a1..d3337b7 100644 --- a/src/willitbreak/__init__.py +++ b/src/willitbreak/__init__.py @@ -24,7 +24,7 @@ from .surface import Surface, Symbol, read_surface from .usage import Reference, scan_paths, scan_source -__version__ = "0.1.0" +__version__ = "0.1.1" __all__ = [ "Change", diff --git a/src/willitbreak/fetch.py b/src/willitbreak/fetch.py index f58e5f4..029541a 100644 --- a/src/willitbreak/fetch.py +++ b/src/willitbreak/fetch.py @@ -143,26 +143,57 @@ def _download(url: str) -> bytes: raise FetchError(f"download failed: {exc}") from exc +def _archive_target(destination: pathlib.Path, member: str) -> pathlib.Path: + """Resolve one archive member inside the exact destination directory. + + A string-prefix check is not a directory-boundary check: a sibling named + ``package-escape`` starts with the path to ``package``. ``relative_to`` + compares path components instead, so absolute paths, ``..`` traversal, + and same-prefix siblings are all rejected. + """ + root = destination.resolve() + target = (root / member).resolve() + try: + target.relative_to(root) + except ValueError as exc: + raise FetchError(f"archive entry escapes its directory: {member}") from exc + return target + + def _safe_extract_zip(data: bytes, destination: pathlib.Path) -> None: with zipfile.ZipFile(io.BytesIO(data)) as archive: - for member in archive.namelist(): - target = (destination / member).resolve() + for member in archive.infolist(): # An archive entry that escapes the destination is either malice # or corruption; either way it does not get written. - if not str(target).startswith(str(destination.resolve())): - raise FetchError(f"archive entry escapes its directory: {member}") - archive.extractall(destination) + target = _archive_target(destination, member.filename) + if member.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + with archive.open(member) as source, open(target, "wb") as output: + shutil.copyfileobj(source, output) def _safe_extract_tar(data: bytes, destination: pathlib.Path) -> None: with tarfile.open(fileobj=io.BytesIO(data), mode="r:*") as archive: for member in archive.getmembers(): - target = (destination / member.name).resolve() - if not str(target).startswith(str(destination.resolve())): - raise FetchError(f"archive entry escapes its directory: {member.name}") - if member.issym() or member.islnk(): - raise FetchError(f"archive contains a link: {member.name}") - archive.extractall(destination) + target = _archive_target(destination, member.name) + # PyPI packages only need regular files and directories. Links, + # devices, and FIFOs can redirect writes or create host objects + # that have no place in a source archive. + if not (member.isfile() or member.isdir()): + raise FetchError( + f"archive contains an unsupported entry: {member.name}" + ) + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + continue + source = archive.extractfile(member) + if source is None: # Defensive: every regular file should open. + raise FetchError(f"could not read archive entry: {member.name}") + target.parent.mkdir(parents=True, exist_ok=True) + with source, open(target, "wb") as output: + shutil.copyfileobj(source, output) def _find_import_root( diff --git a/tests/test_fetch.py b/tests/test_fetch.py new file mode 100644 index 0000000..08700ca --- /dev/null +++ b/tests/test_fetch.py @@ -0,0 +1,81 @@ +"""Archive fetching and extraction security tests.""" + +from __future__ import annotations + +import io +import pathlib +import tarfile +import tempfile +import unittest +import zipfile + +from willitbreak.fetch import FetchError, _safe_extract_tar, _safe_extract_zip + + +class SafeExtractionTests(unittest.TestCase): + def setUp(self) -> None: + self._temp = tempfile.TemporaryDirectory() + self.addCleanup(self._temp.cleanup) + self.root = pathlib.Path(self._temp.name) + self.destination = self.root / "package" + self.destination.mkdir() + + def test_zip_rejects_same_prefix_sibling_escape(self) -> None: + data = io.BytesIO() + with zipfile.ZipFile(data, "w") as archive: + archive.writestr("../package-escape/proof.txt", "escaped") + + with self.assertRaisesRegex(FetchError, "escapes its directory"): + _safe_extract_zip(data.getvalue(), self.destination) + + self.assertFalse((self.root / "package-escape" / "proof.txt").exists()) + + def test_tar_rejects_same_prefix_sibling_escape(self) -> None: + data = io.BytesIO() + with tarfile.open(fileobj=data, mode="w:gz") as archive: + payload = b"escaped" + member = tarfile.TarInfo("../package-escape/proof.txt") + member.size = len(payload) + archive.addfile(member, io.BytesIO(payload)) + + with self.assertRaisesRegex(FetchError, "escapes its directory"): + _safe_extract_tar(data.getvalue(), self.destination) + + self.assertFalse((self.root / "package-escape" / "proof.txt").exists()) + + def test_tar_rejects_non_file_members(self) -> None: + data = io.BytesIO() + with tarfile.open(fileobj=data, mode="w:gz") as archive: + member = tarfile.TarInfo("device") + member.type = tarfile.FIFOTYPE + archive.addfile(member) + + with self.assertRaisesRegex(FetchError, "unsupported entry"): + _safe_extract_tar(data.getvalue(), self.destination) + + def test_regular_zip_and_tar_members_are_extracted(self) -> None: + zip_data = io.BytesIO() + with zipfile.ZipFile(zip_data, "w") as archive: + archive.writestr("zip/source.py", "ZIP = True\n") + _safe_extract_zip(zip_data.getvalue(), self.destination) + + tar_data = io.BytesIO() + with tarfile.open(fileobj=tar_data, mode="w:gz") as archive: + payload = b"TAR = True\n" + member = tarfile.TarInfo("tar/source.py") + member.size = len(payload) + archive.addfile(member, io.BytesIO(payload)) + _safe_extract_tar(tar_data.getvalue(), self.destination) + + self.assertEqual( + (self.destination / "zip" / "source.py").read_text(encoding="utf-8"), + "ZIP = True\n", + ) + self.assertEqual( + (self.destination / "tar" / "source.py").read_text(encoding="utf-8"), + "TAR = True\n", + ) + + +if __name__ == "__main__": + unittest.main()