diff --git a/modules/python-package/.dagger/src/python_package/distribution_probe.py b/modules/python-package/.dagger/src/python_package/distribution_probe.py index c6943c0..f68f6b6 100644 --- a/modules/python-package/.dagger/src/python_package/distribution_probe.py +++ b/modules/python-package/.dagger/src/python_package/distribution_probe.py @@ -360,7 +360,16 @@ def _zip_end_record(path: Path) -> list[object]: raw: object = zipfile._EndRecData(stream) # type: ignore[attr-defined] if not isinstance(raw, list): raise ProbeError("wheel end record differs") - return cast(list[object], raw) + return _with_classic_end_location(path, cast(list[object], raw)) + + +def _with_classic_end_location(path: Path, record: list[object]) -> list[object]: + # For ZIP64 archives zipfile reports the classic end record's offset on CPython <= 3.13.5 + # but the ZIP64 end record's offset on 3.13.14. The classic record ends the file, so + # derive its offset from EOF; the physical-EOF check then verifies its signature there. + comment_size = _zip_field(record, ZIP_COMMENT_SIZE_INDEX) + location = path.stat().st_size - ZIP_END_RECORD_BYTES - comment_size + return [*record[:ZIP_END_LOCATION_INDEX], location, *record[ZIP_END_LOCATION_INDEX + 1 :]] def _zip_field(record: list[object], index: int) -> int: diff --git a/modules/python-package/.dagger/tests/test_distribution_probe.py b/modules/python-package/.dagger/tests/test_distribution_probe.py index 3d139e1..e83f79e 100644 --- a/modules/python-package/.dagger/tests/test_distribution_probe.py +++ b/modules/python-package/.dagger/tests/test_distribution_probe.py @@ -253,6 +253,21 @@ def test_should_observe_bounded_forced_zip64_wheel( assert observed[0].member_count == 4 +def test_should_reject_bytes_after_forced_zip64_wheel_end_record( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given a valid forced-ZIP64 wheel followed by bytes outside its framing + monkeypatch.setattr(zipfile, "ZIP64_LIMIT", 1) + _write_forced_zip64_wheel(tmp_path) + _write_sdist(tmp_path, member_count=4) + target = tmp_path / "probe_package-1.2.3-py3-none-any.whl" + target.write_bytes(target.read_bytes() + b"MZ" + bytes(62)) + + # When / Then trailing bytes still fail closed + with pytest.raises(ProbeError, match=r"physical EOF|end record"): + inspect_directory(tmp_path) + + def test_should_observe_bounded_data_descriptor_wheel(tmp_path: Path) -> None: # Given a valid wheel written to an unseekable stream with data descriptors _write_data_descriptor_wheel(tmp_path)