Reject crafted ZIPs that scan empty but unpack code (#780)

guarddog/utils/archives.py

@@ -2,6 +2,7 @@
 import os
 import pathlib
 import stat
+import struct
 import zipfile
 
 import tarsafe  # type: ignore
@@ -39,6 +40,59 @@ def is_zip_archive(path: str) -> bool:
     return is_tar_archive(path) or is_zip_archive(path)
 
 
+_ZIP_LOCAL_FILE_HEADER = b"PK\x03\x04"
+
+
+def _count_local_file_headers(path: str) -> int:
+    """
+    Count members by walking the ZIP local file headers, independent of the
+    End-Of-Central-Directory record and central directory that zipfile trusts.
+
+    Each header is parsed for its compressed size so we can seek over the data
+    rather than scanning for the next signature (which would false-positive on
+    compressed bytes). Returns the number of headers walked. The walk stops
+    conservatively (undercounting) when a member cannot be followed cheaply: a
+    data descriptor with no inline sizes (general-purpose bit 3) or a ZIP64 size
+    marker. Undercounting is safe here; it only avoids false positives.
+    """
+    count = 0
+    with open(path, "rb") as f:
+        while True:
+            header = f.read(30)
+            if len(header) < 30 or header[:4] != _ZIP_LOCAL_FILE_HEADER:
+                break
+            flags = struct.unpack("<H", header[6:8])[0]
+            compressed_size = struct.unpack("<I", header[18:22])[0]
+            name_len = struct.unpack("<H", header[26:28])[0]
+            extra_len = struct.unpack("<H", header[28:30])[0]
+            if (flags & 0x08 and compressed_size == 0) or compressed_size == 0xFFFFFFFF:
+                break
+            f.seek(name_len + extra_len + compressed_size, os.SEEK_CUR)
+            count += 1
+    return count
+
+
+def _assert_zip_fully_enumerated(source_archive: str, enumerated: int) -> None:
+    """
+    Guard against ZIP parser-differential evasion (e.g. an End-Of-Central-Directory
+    record with size-of-central-directory set to 0, which makes zipfile read an
+    empty archive while installers still unpack the payload from the local file
+    headers). Raises if the local file headers expose more members than zipfile
+    enumerated from the central directory.
+
+    See https://github.com/DataDog/guarddog/issues/780 and the ZIP
+    parser-differential class described in USENIX Security 2025, "My ZIP isn't
+    your ZIP".
+    """
+    walked = _count_local_file_headers(source_archive)
+    if walked > enumerated:
+        raise ValueError(
+            f"archive parser anomaly: {walked} ZIP local file headers but zipfile "
+            f"enumerates {enumerated} members from the central directory "
+            f"(possible scan evasion via EOCD size/offset differential)"
+        )
+
+
 def safe_extract(
     source_archive: str,
     target_directory: str,
@@ -181,8 +235,15 @@ def recurse_add_perms(path):
 
     elif zipfile.is_zipfile(source_archive):
         with zipfile.ZipFile(source_archive, "r") as zip_file:
+            members = zip_file.infolist()
+
+            # Reject archives where the central directory zipfile reads hides
+            # members that are still present as local file headers (and unpacked
+            # by installers). Otherwise such an archive scans as if it were empty.
+            _assert_zip_fully_enumerated(source_archive, len(members))
+
             # Check uncompressed size for zip archives
-            files = [info for info in zip_file.infolist() if not info.is_dir()]
+            files = [info for info in members if not info.is_dir()]
             file_count = len(files)
             total_size = sum(info.file_size for info in files)
 

tests/core/test_archives.py

@@ -1,4 +1,6 @@
+import binascii
 import os
+import struct
 
 import pytest
 
@@ -10,6 +12,67 @@
 ZIP_PASSWORD = b"hunter2"
 
 
+def _build_zip(members: dict[str, bytes], cd_size: int | None = None) -> bytes:
+    """
+    Build a stored (uncompressed) ZIP by hand so the End-Of-Central-Directory
+    size-of-central-directory field can be overridden. With ``cd_size=0`` the
+    archive reproduces the parser-differential from issue #780: zipfile reads it
+    as empty while the local file headers still carry every member.
+    """
+    body = bytearray()
+    central = bytearray()
+    offsets = []
+    for name, data in members.items():
+        raw = name.encode()
+        crc = binascii.crc32(data) & 0xFFFFFFFF
+        offsets.append(len(body))
+        body += b"PK\x03\x04" + struct.pack(
+            "<HHHHHIIIHH", 20, 0, 0, 0, 0x21, crc, len(data), len(data), len(raw), 0
+        )
+        body += raw + data
+    real_cd_size = 0
+    for (name, data), offset in zip(members.items(), offsets):
+        raw = name.encode()
+        crc = binascii.crc32(data) & 0xFFFFFFFF
+        record = (
+            b"PK\x01\x02"
+            + struct.pack(
+                "<HHHHHHIIIHHHHHII",
+                20,
+                20,
+                0,
+                0,
+                0,
+                0x21,
+                crc,
+                len(data),
+                len(data),
+                len(raw),
+                0,
+                0,
+                0,
+                0,
+                (0o100644) << 16,
+                offset,
+            )
+            + raw
+        )
+        central += record
+        real_cd_size += len(record)
+    count = len(members)
+    eocd_cd_size = real_cd_size if cd_size is None else cd_size
+    eocd = b"PK\x05\x06" + struct.pack(
+        "<HHHHIIH", 0, 0, count, count, eocd_cd_size, len(body), 0
+    )
+    return bytes(body) + bytes(central) + eocd
+
+
+_WHL_MEMBERS = {
+    "pkg/__init__.py": b"print('hello')\n",
+    "pkg-1.0.dist-info/METADATA": b"Metadata-Version: 2.1\nName: pkg\nVersion: 1.0\n",
+}
+
+
 def test_encrypted_zip_extracts_with_correct_password(tmp_path):
     safe_extract(ENCRYPTED_ZIP, str(tmp_path), zip_password=ZIP_PASSWORD)
     assert (tmp_path / "index.js").read_text() == 'console.log("hi")\n'
@@ -29,3 +92,26 @@ def test_encrypted_zip_wrong_password_raises(tmp_path):
 def test_tar_archive_rejects_password(tmp_path):
     with pytest.raises(ValueError, match="only supported for ZIP"):
         safe_extract(PLAIN_TARGZ, str(tmp_path), zip_password=ZIP_PASSWORD)
+
+
+def test_well_formed_zip_extracts(tmp_path):
+    archive = tmp_path / "pkg-1.0-py3-none-any.whl"
+    archive.write_bytes(_build_zip(_WHL_MEMBERS))
+    out = tmp_path / "out"
+    out.mkdir()
+    safe_extract(str(archive), str(out))
+    assert (out / "pkg" / "__init__.py").read_bytes() == _WHL_MEMBERS["pkg/__init__.py"]
+
+
+def test_cd_size_zero_eocd_differential_rejected(tmp_path):
+    # zipfile reads this as empty (namelist() == []) but the payload is still
+    # present in the local file headers; safe_extract must refuse it (issue #780).
+    archive = tmp_path / "crafted-1.0-py3-none-any.whl"
+    archive.write_bytes(_build_zip(_WHL_MEMBERS, cd_size=0))
+
+    import zipfile
+
+    assert zipfile.ZipFile(str(archive)).namelist() == []
+
+    with pytest.raises(ValueError, match="parser anomaly"):
+        safe_extract(str(archive), str(tmp_path / "out"))