Further lock-down ZIP archive features (#18846)

* Further lock-down ZIP archive features

* Get 100% coverage

* Add unexpected error to assert message

Author: sethmlarson

tests/unit/utils/test_zipfiles.py

@@ -4,6 +4,7 @@
 import os
 import pathlib
 import struct
+import tempfile
 
 import pytest
 
@@ -23,7 +24,18 @@ def zippath(filename: str):
         ("reject/8bitcomment.zip", "Filename not in central directory"),
         ("reject/cd_extra_entry.zip", "Duplicate filename in central directory"),
         ("reject/cd_missing_entry.zip", "Filename not in central directory"),
-        ("reject/data_descriptor_bad_crc_0.zip", "Unknown record signature"),
+        ("reject/data_descriptor.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_bad_crc.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_bad_crc_0.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_bad_csize.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_bad_usize.zip", "ZIP contains a data descriptor"),
+        (
+            "reject/data_descriptor_bad_usize_no_sig.zip",
+            "ZIP contains a data descriptor",
+        ),
+        ("reject/data_descriptor_zip64.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_zip64_csize.zip", "ZIP contains a data descriptor"),
+        ("reject/data_descriptor_zip64_usize.zip", "ZIP contains a data descriptor"),
         ("reject/dupe_eocd.zip", "Truncated central directory"),
         (
             "reject/eocd64_locator_mismatch.zip",
@@ -37,10 +49,10 @@ def zippath(filename: str):
         ("reject/non_ascii_original_name.zip", "Filename not unicode"),
         ("reject/not.zip", "File is not a zip file"),
         ("reject/prefix.zip", "Unknown record signature"),
-        ("reject/second_unicode_extra.zip", "Filename not in central directory"),
+        ("reject/second_unicode_extra.zip", "Invalid duplicate extra in local file"),
         ("reject/shortextra.zip", "Corrupt extra field 7075 (size=9)"),
         ("reject/suffix_not_comment.zip", "Trailing data"),
-        ("reject/unicode_extra_chain.zip", "Filename not in central directory"),
+        ("reject/unicode_extra_chain.zip", "Invalid duplicate extra in local file"),
         ("reject/wheel-1.0-py3-none-any.whl", "Duplicate filename in local headers"),
         ("reject/zip64_eocd_confusion.zip", "Filename not in central directory"),
         ("reject/zip64_eocd_extensible_data.zip", "Bad offset for central directory"),
@@ -70,7 +82,7 @@ def test_bad_zips(filename, error):
 @pytest.mark.parametrize("filename", list(os.listdir(ZIPDATA_DIR / "accept")))
 def test_good_zips(filename):
     result = zipfiles.validate_zipfile(zippath(f"accept/{filename}"))
-    assert result[0] is True
+    assert result[0] is True, f"Expected no error, got {result[1]!r}"
     assert result[1] is None
 
 
@@ -151,3 +163,185 @@ def test_local_file_header_zip64_extra_no_compressed_size_nok_using_deflate():
     fp = io.BytesIO(header + b"a" + extra + b"a")
     with pytest.raises(zipfiles.InvalidZipFileError):
         zipfiles._handle_local_file_header(fp, {"a": 0})
+
+
+def test_local_file_invalid_filename():
+    header = struct.pack("<xxHHxxxxxxxxLxxxxHxx", 0, 0, 0xFFFFFFFF, 1)
+    fp = io.BytesIO(header + b"\x7f")
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_local_file_header(fp, {"a": 0})
+    assert str(e.value) == "Invalid character in filename"
+
+
+def test_local_file_invalid_filename_in_unicode_extra():
+    extra = struct.pack("<HHxxxxxxxx", 0x7075, 8)
+    header = struct.pack("<xxHHxxxxxxxxLxxxxHH", 0, 0, 0, 1, len(extra))
+    fp = io.BytesIO(header + b"a" + extra)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_local_file_header(fp, {"a": 0})
+    assert str(e.value) == "Invalid character in filename"
+
+
+def test_local_file_invalid_filename_utf8():
+    extra = struct.pack("<HHxxxxx", 0x7075, 6) + b"\xf7"
+    header = struct.pack("<xxHHxxxxxxxxLxxxxHH", 0, 0, 0, 1, len(extra))
+    fp = io.BytesIO(header + b"a" + extra)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_local_file_header(fp, {"a": 0})
+    assert str(e.value) == "Filename not valid UTF-8"
+
+
+def test_local_file_multiple_extras():
+    extras = struct.pack("<HH", 0x7075, 1) + b"a" + struct.pack("<HHxxxx", 0x0002, 4)
+    header = struct.pack("<xxHHxxxxxxxxLxxxxHH", 0, 0, 0, 1, len(extras))
+    fp = io.BytesIO(header + b"a" + extras)
+    filename = zipfiles._handle_local_file_header(fp, {"a": 0})
+    assert filename == b"a"
+
+
+def test_cd_with_comment_rejected():
+    data = struct.pack("<xxxxxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 0, 1, 0, 1, 0)
+    fp = io.BytesIO(data)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_central_directory_header(fp)
+    assert str(e.value) == "Comment in central directory"
+
+
+def test_cd_with_invalid_filename():
+    data = struct.pack("<xxxxxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 0, 1, 0, 0, 0) + b"\x00"
+    fp = io.BytesIO(data)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_central_directory_header(fp)
+    assert str(e.value) == "Invalid character in filename"
+
+
+def test_eocd_mismatched_records_on_disk():
+    data = struct.pack("<xxxxHHLLH", 100, 1, 0, 0, 0)
+    fp = io.BytesIO(data)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_eocd(fp)
+    assert str(e.value) == "Malformed zip file"
+
+
+def test_eocd64_mismatched_records_on_disk():
+    data = struct.pack("<QxxxxxxxxxxxxQQQQ", 0, 100, 1, 0, 0)
+    fp = io.BytesIO(data)
+    with pytest.raises(zipfiles.InvalidZipFileError) as e:
+        zipfiles._handle_eocd64(fp)
+    assert str(e.value) == "Malformed zip file"
+
+
+def test_cd_and_eocd_match():
+    data_lf = (
+        zipfiles.RECORD_SIG_LOCAL_FILE
+        + struct.pack("<xxHHHxxxxxxLxxxxHH", 0, 0, 20, 0, 1, 0)
+        + b"a"
+    )
+    data_cd = (
+        zipfiles.RECORD_SIG_CENTRAL_DIRECTORY
+        + struct.pack("<HHxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 20, 20, 0, 1, 0, 0, 0)
+        + b"a"
+    )
+    cd_records = 1
+    cd_offset = len(data_lf)
+    cd_size = len(data_cd)
+    data_eocd = zipfiles.RECORD_SIG_EOCD + struct.pack(
+        "<xxxxHHLLH", cd_records, cd_records, cd_size, cd_offset, 0
+    )
+    data = data_lf + data_cd + data_eocd
+    with tempfile.NamedTemporaryFile(mode="wb") as tmp:
+        tmp.write(data)
+        tmp.flush()
+        assert (True, None) == zipfiles.validate_zipfile(tmp.name)
+
+
+@pytest.mark.parametrize(
+    ("cd_records", "error"),
+    [
+        (0, "Mismatched central directory records"),
+        (2, "Mismatched central directory records"),
+    ],
+)
+def test_cd_and_eocd_mismatch_records(cd_records, error):
+    data_lf = (
+        zipfiles.RECORD_SIG_LOCAL_FILE
+        + struct.pack("<xxHHHxxxxxxLxxxxHH", 0, 0, 20, 0, 1, 0)
+        + b"a"
+    )
+    data_cd = (
+        zipfiles.RECORD_SIG_CENTRAL_DIRECTORY
+        + struct.pack("<HHxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 20, 20, 0, 1, 0, 0, 0)
+        + b"a"
+    )
+    cd_offset = len(data_lf)
+    cd_size = len(data_cd)
+    data_eocd = zipfiles.RECORD_SIG_EOCD + struct.pack(
+        "<xxxxHHLLH", cd_records, cd_records, cd_size, cd_offset, 0
+    )
+    data = data_lf + data_cd + data_eocd
+    with tempfile.NamedTemporaryFile(mode="wb") as tmp:
+        tmp.write(data)
+        tmp.flush()
+        assert (False, error) == zipfiles.validate_zipfile(tmp.name)
+
+
+@pytest.mark.parametrize(
+    ("cd_offset_diff", "error"),
+    [
+        (-1, "Mismatched central directory offset"),
+        (1, "Mismatched central directory offset"),
+    ],
+)
+def test_cd_and_eocd_mismatch_offset(cd_offset_diff, error):
+    data_lf = (
+        zipfiles.RECORD_SIG_LOCAL_FILE
+        + struct.pack("<xxHHHxxxxxxLxxxxHH", 0, 0, 20, 0, 1, 0)
+        + b"a"
+    )
+    data_cd = (
+        zipfiles.RECORD_SIG_CENTRAL_DIRECTORY
+        + struct.pack("<HHxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 20, 20, 0, 1, 0, 0, 0)
+        + b"a"
+    )
+    cd_records = 1
+    cd_offset = len(data_lf) + cd_offset_diff
+    cd_size = len(data_cd)
+    data_eocd = zipfiles.RECORD_SIG_EOCD + struct.pack(
+        "<xxxxHHLLH", cd_records, cd_records, cd_size, cd_offset, 0
+    )
+    data = data_lf + data_cd + data_eocd
+    with tempfile.NamedTemporaryFile(mode="wb") as tmp:
+        tmp.write(data)
+        tmp.flush()
+        assert (False, error) == zipfiles.validate_zipfile(tmp.name)
+
+
+@pytest.mark.parametrize(
+    ("cd_size_diff", "error"),
+    [
+        (-1, "Bad magic number for central directory"),
+        (1, "Bad magic number for central directory"),
+    ],
+)
+def test_cd_and_eocd_mismatch_size(cd_size_diff, error):
+    data_lf = (
+        zipfiles.RECORD_SIG_LOCAL_FILE
+        + struct.pack("<xxHHHxxxxxxLxxxxHH", 0, 0, 20, 0, 1, 0)
+        + b"a"
+    )
+    data_cd = (
+        zipfiles.RECORD_SIG_CENTRAL_DIRECTORY
+        + struct.pack("<HHxxxxxxxxxxxxLxxxxHHHxxxxxxxxL", 20, 20, 0, 1, 0, 0, 0)
+        + b"a"
+    )
+    cd_records = 1
+    cd_offset = len(data_lf)
+    cd_size = len(data_cd) + cd_size_diff
+    data_eocd = zipfiles.RECORD_SIG_EOCD + struct.pack(
+        "<xxxxHHLLH", cd_records, cd_records, cd_size, cd_offset, 0
+    )
+    data = data_lf + data_cd + data_eocd
+    with tempfile.NamedTemporaryFile(mode="wb") as tmp:
+        tmp.write(data)
+        tmp.flush()
+        assert (False, error) == zipfiles.validate_zipfile(tmp.name)