From bdd148e31b9c54dab6967fb303048e4b7015d766 Mon Sep 17 00:00:00 2001 From: supermario_leo Date: Mon, 7 Sep 2026 13:24:40 +0800 Subject: [PATCH] fix(utils): guard zip extraction against decompression bombs (#4281) Add ZipExtractionLimit caps (total bytes, per-file bytes, member count, compression ratio) and reject archives whose infolist metadata signals a decompression bomb before writing to disk. Wire the same cap into the GitHub/GitLab ZIP download paths via a Content-Length check before shutil.copyfileobj. --- openviking/parse/accessors/git_accessor.py | 7 +++ openviking/parse/parsers/code/code.py | 4 ++ openviking/utils/zip_safe.py | 43 +++++++++++++++- tests/misc/test_zip_safe.py | 59 ++++++++++++++++++++++ 4 files changed, 112 insertions(+), 1 deletion(-) diff --git a/openviking/parse/accessors/git_accessor.py b/openviking/parse/accessors/git_accessor.py index cf46230ca3..4e91191b93 100644 --- a/openviking/parse/accessors/git_accessor.py +++ b/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: diff --git a/openviking/parse/parsers/code/code.py b/openviking/parse/parsers/code/code.py index be4c7f307c..f5a2b784f4 100644 --- a/openviking/parse/parsers/code/code.py +++ b/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: diff --git a/openviking/utils/zip_safe.py b/openviking/utils/zip_safe.py index 60ca28615c..f238f6cfda 100644 --- a/openviking/utils/zip_safe.py +++ b/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,7 +82,39 @@ 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. @@ -89,8 +122,11 @@ def safe_extract_zip(zipf: zipfile.ZipFile, dest_dir: Path) -> None: 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) diff --git a/tests/misc/test_zip_safe.py b/tests/misc/test_zip_safe.py index cd273196a3..bd2ef5a46e 100644 --- a/tests/misc/test_zip_safe.py +++ b/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()