fix(utils): guard zip extraction against decompression bombs (#4281)

openviking/parse/accessors/git_accessor.py

@@ -42,6 +42,7 @@
     parse_git_http_auth_config,
     raise_git_auth_error,
 )
+from openviking.utils.zip_safe import ZipExtractionLimit
 from openviking_cli.utils.logger import get_logger
 
 from .base import DataAccessor, LocalResource, SourceType
@@ -505,6 +506,9 @@ def _download() -> None:
             req = urllib.request.Request(zip_url, headers=headers)
             context = ParserNetworkContext(ParserNetworkOperation.GIT_ARCHIVE, zip_url)
             with open_parser_url(context, req, timeout=1800) as resp, open(zip_path, "wb") as f:
+                content_length = resp.headers.get("Content-Length")
+                if content_length and int(content_length) > ZipExtractionLimit().max_total_bytes:
+                    raise ValueError(f"GitHub ZIP too large to download: {content_length} bytes")
                 shutil.copyfileobj(resp, f)
 
         try:
@@ -599,6 +603,9 @@ def _download() -> None:
             req = urllib.request.Request(zip_url, headers=headers)
             context = ParserNetworkContext(ParserNetworkOperation.GIT_ARCHIVE, zip_url)
             with open_parser_url(context, req, timeout=1800) as resp, open(zip_path, "wb") as f:
+                content_length = resp.headers.get("Content-Length")
+                if content_length and int(content_length) > ZipExtractionLimit().max_total_bytes:
+                    raise ValueError(f"GitLab ZIP too large to download: {content_length} bytes")
                 shutil.copyfileobj(resp, f)
 
         try:

openviking/parse/parsers/code/code.py

@@ -39,6 +39,7 @@
 from openviking.parse.parsers.upload_utils import upload_directory
 from openviking.utils import is_github_url, parse_code_hosting_url
 from openviking.utils.code_hosting_utils import _domain_matches
+from openviking.utils.zip_safe import ZipExtractionLimit
 from openviking_cli.utils.config import get_openviking_config
 from openviking_cli.utils.logger import get_logger
 
@@ -435,6 +436,9 @@ def _download() -> None:
 
             req = urllib.request.Request(zip_url, headers=headers)
             with urllib.request.urlopen(req, timeout=1800) as resp, open(zip_path, "wb") as f:
+                content_length = resp.headers.get("Content-Length")
+                if content_length and int(content_length) > ZipExtractionLimit().max_total_bytes:
+                    raise ValueError(f"GitHub ZIP too large to download: {content_length} bytes")
                 shutil.copyfileobj(resp, f)
 
         try:

openviking/utils/zip_safe.py

@@ -6,6 +6,7 @@
 import re
 import shutil
 import zipfile
+from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 
 _UTF8_FLAG = 0x800
@@ -81,16 +82,51 @@ def _safe_zip_member_path(filename: str) -> Path:
     return Path(*safe_parts)
 
 
-def safe_extract_zip(zipf: zipfile.ZipFile, dest_dir: Path) -> None:
+@dataclass(frozen=True)
+class ZipExtractionLimit:
+    """Caps that reject decompression bombs before and during extraction."""
+
+    max_total_bytes: int = 2 * 1024**3  # 2 GiB uncompressed aggregate
+    max_per_file_bytes: int = 512 * 1024**2  # 512 MiB per member
+    max_member_count: int = 10_000
+    max_compression_ratio: int = 1_000  # uncompressed : compressed
+
+
+def _check_zip_extraction_limits(zipf: zipfile.ZipFile, limits: ZipExtractionLimit) -> None:
+    """Reject archives whose infolist metadata signals a decompression bomb.
+
+    Inspects member count, aggregate uncompressed size, and overall
+    compression ratio before any byte is written to disk.
+    """
+    infos = zipf.infolist()
+    if len(infos) > limits.max_member_count:
+        raise ValueError(f"Zip member count {len(infos)} exceeds limit {limits.max_member_count}")
+    total_uncompressed = sum(i.file_size for i in infos)
+    if total_uncompressed > limits.max_total_bytes:
+        raise ValueError(
+            f"Zip total uncompressed size {total_uncompressed} exceeds limit "
+            f"{limits.max_total_bytes}"
+        )
+    total_compressed = sum(i.compress_size for i in infos)
+    if total_compressed and total_uncompressed / total_compressed > limits.max_compression_ratio:
+        raise ValueError(f"Zip compression ratio exceeds limit {limits.max_compression_ratio}")
+
+
+def safe_extract_zip(
+    zipf: zipfile.ZipFile, dest_dir: Path, limits: ZipExtractionLimit | None = None
+) -> None:
     """Extract ZIP archive with Zip Slip protection.
 
     Validates every member path stays within dest_dir before extraction.
     Rejects absolute paths, Windows drive prefixes, and parent-directory
     traversal (..). Windows-style backslash separators are normalized to
     forward slash semantics before extraction.
     """
+    if limits is None:
+        limits = ZipExtractionLimit()
     dest_dir = Path(dest_dir).resolve()
     normalize_zip_filenames(zipf)
+    _check_zip_extraction_limits(zipf, limits)
     for member in zipf.infolist():
         safe_rel_path = _safe_zip_member_path(member.filename)
         member_path = (dest_dir / safe_rel_path).resolve()
@@ -101,6 +137,11 @@ def safe_extract_zip(zipf: zipfile.ZipFile, dest_dir: Path) -> None:
             member_path.mkdir(parents=True, exist_ok=True)
             continue
 
+        if member.file_size > limits.max_per_file_bytes:
+            raise ValueError(
+                f"Zip member {member.filename!r} size {member.file_size} exceeds "
+                f"per-file limit {limits.max_per_file_bytes}"
+            )
         member_path.parent.mkdir(parents=True, exist_ok=True)
         with zipf.open(member) as source, member_path.open("wb") as target:
             shutil.copyfileobj(source, target)

tests/misc/test_zip_safe.py

@@ -412,3 +412,62 @@ def test_symlink_entry_does_not_create_symlink_outside_dest(self, tmp_path: Path
                 assert target.is_relative_to(dest.resolve()), (
                     f"Symlink {item} points outside dest: {target}"
                 )
+
+
+class TestSafeExtractZipBombGuard:
+    """Verify decompression-bomb guards reject oversized archives."""
+
+    def test_decompression_bomb_rejected(self, tmp_path: Path) -> None:
+        """A highly-compressed archive is rejected before extraction.
+
+        A 1 MiB payload compressed with bzip2 expands past the default
+        compression-ratio cap (and would, on a real bomb, exhaust disk).
+        On an unguarded build the member extracts to disk; the default guard
+        rejects it from infolist metadata before any byte is written.
+        """
+        dest = tmp_path / "out"
+        dest.mkdir()
+        payload = b"\x00" * (1024 * 1024)  # bzip2 compresses this to ~45 bytes
+        buf = io.BytesIO()
+        with zipfile.ZipFile(buf, "w", zipfile.ZIP_BZIP2) as zf:
+            zf.writestr("bomb.bin", payload)
+        buf.seek(0)
+        with zipfile.ZipFile(buf, "r") as zf:
+            with pytest.raises(ValueError, match="compression ratio"):
+                safe_extract_zip(zf, dest)
+        assert not (dest / "bomb.bin").exists()
+
+    def test_zip_member_count_limit(self, tmp_path: Path) -> None:
+        """Archives with more members than the cap are rejected."""
+        from openviking.utils.zip_safe import ZipExtractionLimit
+
+        dest = tmp_path / "out"
+        dest.mkdir()
+        entries = {f"file_{i}.txt": "x" for i in range(20)}
+        data = _make_zip_bytes(entries)
+        limits = ZipExtractionLimit(max_member_count=10)
+        with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
+            with pytest.raises(ValueError, match="member count"):
+                safe_extract_zip(zf, dest, limits=limits)
+        assert list(dest.iterdir()) == []
+
+    def test_per_file_size_limit(self, tmp_path: Path) -> None:
+        """A single member larger than the per-file cap is rejected mid-loop.
+
+        Aggregate caps (count/total/ratio) are sized to pass so the per-file
+        check inside the extraction loop is what rejects the member.
+        """
+        from openviking.utils.zip_safe import ZipExtractionLimit
+
+        dest = tmp_path / "out"
+        dest.mkdir()
+        data = _make_zip_bytes({"big.txt": "A" * 2048})  # 2 KiB, stored
+        limits = ZipExtractionLimit(
+            max_total_bytes=1024 * 1024,
+            max_per_file_bytes=1024,
+            max_compression_ratio=100_000,
+        )
+        with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
+            with pytest.raises(ValueError, match="per-file"):
+                safe_extract_zip(zf, dest, limits=limits)
+        assert not (dest / "big.txt").exists()