Harden package archive extraction

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

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",

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(

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()