Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,4 @@ sentry-kafka-schemas==2.1.2
sentry-sdk>=2.36.0
sortedcontainers>=2.4.0
typing-extensions>=4.15.0
zipfile-zstd==0.0.4
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like we get zstd for free in Python 3.14, not sure if you looked into that option: https://docs.python.org/3/library/compression.zstd.html#

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i tried upgrading us to 3.14, but that requires upgrading confluent-kafka to 2.12 which then conflicts with sentry-arroyo which has that package pinned to <2.10.0... so not possible right now

hen we're off of arroyo though.. lol

3 changes: 3 additions & 0 deletions src/launchpad/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1,4 @@
# Monkey patches - import these first to register handlers globally
import zipfile_zstd # noqa: F401 - Registers zstd compression support with zipfile module. Should not be required after upgrading to python 3.14

__version__ = "0.0.1"
7 changes: 5 additions & 2 deletions src/launchpad/artifacts/providers/zip_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,12 +92,15 @@ def extract_to_temp_directory(self) -> Path:
self._temp_dirs.append(temp_dir)

self._safe_extract(str(self.path), str(temp_dir))
logger.debug(f"Extracted zip contents to {temp_dir} using system unzip")
logger.debug(f"Extracted zip contents to {temp_dir}")

return temp_dir

def _safe_extract(self, zip_path: str, extract_path: str):
"""Extract the zip contents to a temporary directory, ensuring that the paths are safe from path traversal attacks."""
"""Extract the zip contents to a temporary directory, ensuring that the paths are safe from path traversal attacks.

Supports both standard compression methods and Zstandard compression.
"""
base_dir = Path(extract_path)
with zipfile.ZipFile(zip_path, "r") as zip_ref:
check_reasonable_zip(zip_ref)
Expand Down
19 changes: 19 additions & 0 deletions tests/unit/artifacts/providers/test_zip_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,3 +125,22 @@ def test_max_file_size(self, hackernews_xcarchive: Path) -> None:
# iOS fixture is ~32MB uncompressed, so limit of 10MB should fail
with pytest.raises(UnreasonableZipError, match="exceeding the limit of 10.0MB"):
check_reasonable_zip(zf, max_uncompressed_size=10 * 1024 * 1024)

def test_extract_zstd_zip(self) -> None:
"""Test that zstd-compressed zips can be extracted."""
with tempfile.NamedTemporaryFile(suffix=".zip") as temp_file:
temp_path = Path(temp_file.name)

# Create a zstd-compressed zip (compression method 93)
with zipfile.ZipFile(temp_path, "w") as zf:
zf.writestr("test.txt", "content", compress_type=93)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can this use zipfile.ZIP_ZSTANDARD instead of magic number 93?

Copy link
Contributor Author

@NicoHinderling NicoHinderling Dec 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"Unfortunately no - the standard library zipfile module doesn't define ZIP_ZSTANDARD. The zipfile_zstd package only adds the decompression handler, not the constant."

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

perhaps this will exist in the python 3.14 world though..


try:
provider = ZipProvider(temp_path)
temp_dir = provider.extract_to_temp_directory()

assert temp_dir.exists()
assert (temp_dir / "test.txt").exists()
assert (temp_dir / "test.txt").read_text() == "content"
finally:
temp_path.unlink(missing_ok=True)
Loading