Reject ZIP parser-differential archives in safe_extract (#780)

A wheel/zip can be crafted so Python's zipfile reads it as empty (e.g. an
End-Of-Central-Directory record with size-of-central-directory set to 0, a
trailing second EOCD with 0 entries, or a cd_offset pointing at an empty
central directory) while installers still unpack the payload from the local
file headers. GuardDog enumerates members from the central directory, so such
a package extracts zero files and scans as clean ("Found 0 potentially
malicious indicators") even though it carries executable code.

safe_extract now walks the local file headers independently of the EOCD and
rejects the archive when more members are visible there than zipfile
enumerates. The walk parses each header's compressed size and seeks over the
data (instead of scanning for the PK signature, which false-positives on
compressed bytes), and stops conservatively on data descriptors / ZIP64 size
markers to avoid false positives.

Tested: flags the cd_size=0 and trailing-EOCD framings; no false positive on a
real DEFLATE wheel (107 members incl. a directory entry).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dvlinuxx-max

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