diff --git a/docs/reference/catalogs.md b/docs/reference/catalogs.md index 7299e6851d..1555bccce8 100644 --- a/docs/reference/catalogs.md +++ b/docs/reference/catalogs.md @@ -109,6 +109,20 @@ configuration does not mean its sources or components were approved. Run the relevant `catalog list` command and inspect every install-allowed source before installing or running components from an unfamiliar project. +Shared archive validation limits the decompressed tar.gz stream to 54 MiB, +including tar headers, padding, and GNU/PAX metadata as well as file contents. +Each GNU/PAX metadata payload is also capped at 64 KiB before it is read or +parsed, with a cumulative budget of 1 MiB for all metadata payloads per archive. +Global PAX payloads have a tighter cumulative budget of 8 KiB per +archive because their fields are copied into retained members. Metadata chains +are limited to 32 headers, and negative metadata sizes are rejected. +These limits apply during format detection and extraction; ordinary member-size +and total-file-size limits still apply. Packages that exceed these limits must +be reduced before installation. +Old GNU sparse archives and GNU sparse 1.0 archives are rejected before their +separately stored maps are parsed; sparse maps contained in bounded PAX 0.0/0.1 +metadata remain supported. + ## Host a private catalog 1. Review and approve the entries and release artifacts before registering an diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index 5ff460666e..b0b5e099fd 100644 --- a/src/specify_cli/_download_security.py +++ b/src/specify_cli/_download_security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import gzip import io import re import socket @@ -32,6 +33,14 @@ # ``ZipFile`` reads this whole structure into memory. Four MiB leaves roughly # 8 KiB of filename/extra/comment metadata for each of the 512 allowed entries. MAX_ZIP_CENTRAL_DIRECTORY_BYTES = 4 * 1024 * 1024 +# Include tar headers, padding, and GNU/PAX metadata, not just file payloads. +MAX_TAR_STREAM_BYTES = MAX_ZIP_TOTAL_BYTES + 4 * 1024 * 1024 +MAX_TAR_METADATA_BYTES = 64 * 1024 +# Per-member metadata dictionaries are retained during archive validation. +MAX_TAR_TOTAL_METADATA_BYTES = 1 * 1024 * 1024 +# Global fields are copied into every retained member. +MAX_TAR_GLOBAL_PAX_BYTES = 8 * 1024 +MAX_TAR_METADATA_DEPTH = 32 READ_CHUNK_SIZE = 64 * 1024 # Tighter ceilings for responses that are read fully into memory and parsed as @@ -92,6 +101,114 @@ } +class _TarLimitError(tarfile.ReadError): + pass + + +def is_archive_safety_error(exc: Exception) -> bool: + """Identify a tar safety signal retained by shared error translation.""" + return isinstance(exc, _TarLimitError) or isinstance(exc.__cause__, _TarLimitError) + + +class _BoundedTarInfo(tarfile.TarInfo): + """Check every metadata header, including headers parsed recursively.""" + + def _proc_member(self, archive: tarfile.TarFile) -> tarfile.TarInfo: + if self.type == tarfile.GNUTYPE_SPARSE: + raise _TarLimitError("Unsupported GNU sparse metadata in tar.gz archive") + if self.type in ( + tarfile.GNUTYPE_LONGNAME, + tarfile.GNUTYPE_LONGLINK, + tarfile.XHDTYPE, + tarfile.XGLTYPE, + tarfile.SOLARIS_XHDTYPE, + ): + if self.size < 0: + raise _TarLimitError("tar.gz metadata payload has negative size") + if self.size > MAX_TAR_METADATA_BYTES: + raise _TarLimitError( + "tar.gz metadata payload exceeds maximum size " + f"of {MAX_TAR_METADATA_BYTES} bytes" + ) + if not isinstance(archive, _BoundedTarFile): + raise TypeError("Metadata validation requires a bounded tar archive") + with archive._metadata_header(self): + return super()._proc_member(archive) + return super()._proc_member(archive) + + def _proc_gnusparse_10( + self, + next_member: tarfile.TarInfo, + pax_headers: dict[str, str], + archive: tarfile.TarFile, + ) -> NoReturn: + # This map lives in file data, outside the bounded PAX header payload. + raise _TarLimitError("Unsupported GNU sparse metadata in tar.gz archive") + + +class _BoundedTarFile(tarfile.TarFile): + tarinfo = _BoundedTarInfo + _metadata_bytes: int = 0 + _global_pax_bytes: int = 0 + _metadata_depth: int = 0 + + @contextmanager + def _metadata_header(self, member: tarfile.TarInfo) -> Iterator[None]: + if self._metadata_depth >= MAX_TAR_METADATA_DEPTH: + raise _TarLimitError( + "tar.gz metadata chain exceeds maximum depth " + f"of {MAX_TAR_METADATA_DEPTH}" + ) + if self._metadata_bytes + member.size > MAX_TAR_TOTAL_METADATA_BYTES: + raise _TarLimitError( + "tar.gz archive metadata exceeds maximum cumulative size " + f"of {MAX_TAR_TOTAL_METADATA_BYTES} bytes" + ) + if member.type == tarfile.XGLTYPE: + if self._global_pax_bytes + member.size > MAX_TAR_GLOBAL_PAX_BYTES: + raise _TarLimitError( + "tar.gz global PAX metadata exceeds maximum cumulative size " + f"of {MAX_TAR_GLOBAL_PAX_BYTES} bytes" + ) + self._global_pax_bytes += member.size + self._metadata_bytes += member.size + self._metadata_depth += 1 + try: + yield + finally: + self._metadata_depth -= 1 + + +class _BoundedTarGzipFile(gzip.GzipFile): + """Bound decompressed offsets before tarfile reads metadata or skips data.""" + + def _check_position(self, position: int) -> None: + if position > MAX_TAR_STREAM_BYTES: + raise _TarLimitError( + "tar.gz archive exceeds maximum decompressed stream size " + f"of {MAX_TAR_STREAM_BYTES} bytes" + ) + + def read(self, size: int = -1) -> bytes: + position = super().seek(0, io.SEEK_CUR) + if size >= 0: + self._check_position(position + size) + else: + size = MAX_TAR_STREAM_BYTES - position + 1 + data = super().read(size) + self._check_position(position + len(data)) + return data + + def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: + if whence == io.SEEK_CUR: + offset += super().seek(0, io.SEEK_CUR) + elif whence != io.SEEK_SET: + # SEEK_END would decompress the whole stream before checking it. + raise io.UnsupportedOperation("Unbounded tar.gz seek") + self._check_position(offset) + return super().seek(offset) + + def archive_format_from_name(name: str) -> ArchiveFormat | None: """Return the supported archive format declared by a path or URL.""" try: @@ -178,15 +295,22 @@ def detect_archive_format( is_tar_gz = False if is_gzip: try: - with tarfile.open(fileobj=archive_file, mode="r:gz"): + with ( + _BoundedTarGzipFile(fileobj=archive_file, mode="rb") as source, + _BoundedTarFile.open(fileobj=source, mode="r:"), + ): is_tar_gz = True - except _TAR_DECOMPRESSION_ERRORS: + except _TarLimitError: + raise + except (*_TAR_DECOMPRESSION_ERRORS, gzip.BadGzipFile): # A truncated gzip stream raises a bare EOFError here rather # than a TarError, so catching only TarError let it escape # this probe as a raw exception instead of leaving # ``is_tar_gz`` False and reporting the format mismatch. pass archive_file.seek(0) + except _TarLimitError as exc: + _raise_from(error_type, str(exc), exc) except OSError as exc: _raise_from(error_type, f"Invalid archive: {archive_path}", exc) @@ -1087,18 +1211,23 @@ def safe_extract_tar( except OSError as exc: _raise_from(error_type, f"Invalid tar extraction target: {target_dir}", exc) - try: - if archive_file is not None: - archive_file.seek(0) - archive = tarfile.open( - archive_path if archive_file is None else None, - mode="r:gz", - fileobj=archive_file, - ) - except (*_TAR_DECOMPRESSION_ERRORS, OSError) as exc: - _raise_from(error_type, f"Invalid tar.gz archive: {archive_path}", exc) + with ExitStack() as stack: + try: + if archive_file is not None: + archive_file.seek(0) + source = stack.enter_context( + _BoundedTarGzipFile( + archive_path if archive_file is None else None, + mode="rb", + fileobj=archive_file, + ) + ) + archive = stack.enter_context( + _BoundedTarFile.open(fileobj=source, mode="r:") + ) + except (*_TAR_DECOMPRESSION_ERRORS, OSError) as exc: + _raise_from(error_type, f"Invalid tar.gz archive: {archive_path}: {exc}", exc) - with archive: validated: list[tuple[tarfile.TarInfo, str, bool]] = [] validated_paths: dict[tuple[str, ...], tuple[str, bool]] = {} total_size = 0 @@ -1170,7 +1299,7 @@ def safe_extract_tar( except (*_TAR_DECOMPRESSION_ERRORS, OSError) as exc: _raise_from( error_type, - f"Invalid tar.gz archive: {archive_path}", + f"Invalid tar.gz archive: {archive_path}: {exc}", exc, ) diff --git a/src/specify_cli/extensions/_commands.py b/src/specify_cli/extensions/_commands.py index 08fe6747fc..5316adf4fa 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -24,6 +24,7 @@ from .._download_security import ( archive_format_from_name, detect_archive_format, + is_archive_safety_error, is_https_or_localhost_http, read_response_limited, ) @@ -254,6 +255,8 @@ def install_extension_from_url( error_type=ExtensionError, ) except ExtensionError as exc: + if is_archive_safety_error(exc): + raise raise ExtensionError( f"{url} did not return a ZIP archive or tar.gz/tgz archive " f"(got {len(archive_data)} bytes). This usually means the request " diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index c1fb716f7b..19e9a35cf0 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -23,6 +23,7 @@ archive_format_from_name, archive_suffix as archive_suffix, detect_archive_format, + is_archive_safety_error, is_https_or_localhost_http, is_safe_download_redirect, read_response_limited, @@ -296,7 +297,9 @@ def _sniff_workflow_archive_format(data: bytes): Path("workflow-download"), archive_file=BytesIO(data), ) - except ValueError: + except ValueError as exc: + if is_archive_safety_error(exc): + raise return None diff --git a/tests/archive_helpers.py b/tests/archive_helpers.py new file mode 100644 index 0000000000..219cad2ee1 --- /dev/null +++ b/tests/archive_helpers.py @@ -0,0 +1,104 @@ +"""Small compressed archives exercising shared tar safety rejections.""" + +import gzip +import io +import tarfile + + +def tar_metadata_record(metadata_type: bytes, size: int, key: str = "comment") -> bytes: + info = tarfile.TarInfo("metadata") + info.type = metadata_type + info.size = size + if metadata_type in (tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK): + payload = b"file.txt\0"[:size].ljust(size, b"\0") + elif size: + prefix = f"{size} {key}=".encode() + payload = prefix + b"x" * (size - len(prefix) - 1) + b"\n" + else: + payload = b"" + return info.tobuf() + payload + b"\0" * (-size % tarfile.BLOCKSIZE) + + +def tar_global_pax_archive(sizes: list[int], *, interleaved: bool) -> bytes: + records = [] + for index, size in enumerate(sizes): + records.append(tar_metadata_record(tarfile.XGLTYPE, size, f"field{index}")) + if interleaved: + records.append(tarfile.TarInfo(f"file{index}.txt").tobuf()) + if not interleaved: + records.append(tarfile.TarInfo("file.txt").tobuf()) + return gzip.compress(b"".join(records) + b"\0" * tarfile.BLOCKSIZE) + + +def tar_metadata_chain_archive( + count: int, metadata_type: bytes = tarfile.XHDTYPE, *, size: int = 32 +) -> bytes: + plain = tar_metadata_record(metadata_type, size) * count + return gzip.compress( + plain + tarfile.TarInfo("file.txt").tobuf() + b"\0" * tarfile.BLOCKSIZE + ) + + +def tar_negative_metadata_archive(metadata_type: bytes) -> bytes: + info = tarfile.TarInfo("metadata") + info.type = metadata_type + info.size = -512 + return gzip.compress( + info.tobuf(format=tarfile.GNU_FORMAT) + b"x" * 65537 + ) + + +def tar_safety_archive(kind: str) -> bytes: + if kind == "total-metadata": + return tar_metadata_chain_archive(17, size=65536) + if kind == "global-pax": + return tar_global_pax_archive([1024] * 9, interleaved=False) + if kind == "metadata-chain": + return tar_metadata_chain_archive(400) + if kind == "negative-metadata": + return tar_negative_metadata_archive(tarfile.XHDTYPE) + if kind in ("stream", "metadata"): + size = 4096 if kind == "stream" else 65537 + prefix = f"{size} comment=".encode() + payload = prefix + b"x" * (size - len(prefix) - 1) + b"\n" + info = tarfile.TarInfo("metadata") + info.type = tarfile.XHDTYPE + info.size = size + plain = info.tobuf() + payload + b"\0" * (-size % tarfile.BLOCKSIZE) + plain += tarfile.TarInfo("file.txt").tobuf() + elif kind == "gnu-sparse": + info = tarfile.TarInfo("file.txt") + info.type = tarfile.GNUTYPE_SPARSE + header = bytearray(info.tobuf(format=tarfile.GNU_FORMAT)) + header[482] = 1 # The old GNU format stores map extensions after this header. + header[483:495] = tarfile.itn(2, 12) + header[148:156] = b" " * 8 + header[148:156] = f"{sum(header):06o}\0 ".encode() + blocks = [] + for index in range(160): + block = bytearray(tarfile.BLOCKSIZE) + for entry in range(21): + offset = entry * 24 + block[offset:offset + 24] = tarfile.itn(1, 12) * 2 + block[504] = int(index < 159) + blocks.append(bytes(block)) + plain = bytes(header) + b"".join(blocks) + elif kind == "pax-sparse-1.0": + sparse_map = (b"20000\n" + b"1\n1\n" * 20000) + payload = sparse_map.ljust( + (len(sparse_map) + 511) // 512 * 512, b"\0" + ) + info = tarfile.TarInfo("file.txt") + info.size = len(payload) + info.pax_headers = { + "GNU.sparse.major": "1", + "GNU.sparse.minor": "0", + "GNU.sparse.realsize": "2", + } + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w", format=tarfile.PAX_FORMAT) as archive: + archive.addfile(info, io.BytesIO(payload)) + plain = buffer.getvalue() + else: + raise ValueError(f"Unknown test archive kind: {kind}") + return gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE) diff --git a/tests/specify_cli/extensions/test_command_add.py b/tests/specify_cli/extensions/test_command_add.py index f898fa052f..7cdcc9308b 100644 --- a/tests/specify_cli/extensions/test_command_add.py +++ b/tests/specify_cli/extensions/test_command_add.py @@ -27,6 +27,7 @@ ValidationError, ) from tests.conftest import strip_ansi +from tests.archive_helpers import tar_safety_archive from tests.specify_cli.extensions._helpers import ( LISTED_COMMAND_NAME_CASES, MINIMAL_ZIP_BYTES as _MINIMAL_ZIP_BYTES, @@ -40,6 +41,58 @@ class TestExtensionAddCLI: """CLI tests for ``specify extension add``.""" + @pytest.mark.parametrize( + "kind, diagnostic", + [ + ("stream", "maximum decompressed"), + ("metadata", "metadata payload exceeds"), + ("gnu-sparse", "Unsupported GNU sparse"), + ("pax-sparse-1.0", "Unsupported GNU sparse"), + ("global-pax", "global PAX metadata"), + ("metadata-chain", "metadata chain"), + ("negative-metadata", "negative size"), + ("total-metadata", "archive metadata exceeds"), + ], + ) + @pytest.mark.parametrize("suffix", ["ext.tar.gz", "assets/42"]) + def test_add_from_url_preserves_archive_safety_errors( + self, project_dir, monkeypatch, kind, diagnostic, suffix + ): + from specify_cli import _download_security + + if kind == "stream": + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 2048) + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch("typer.confirm", return_value=True), + patch( + "specify_cli.extensions._commands._validate_safe_cache_dir", + side_effect=_validate_safe_cache_dir_test_stand_in, + ), + patch( + "specify_cli.extensions._commands._safe_open_download_zip", + side_effect=_open_test_download_zip, + ), + patch( + "specify_cli.authentication.http.open_url", + return_value=io.BytesIO(tar_safety_archive(kind)), + ), + patch.object(ExtensionManager, "install_from_zip") as install, + ): + result = CliRunner().invoke( + app, + ["extension", "add", "unsafe", "--from", f"https://example.com/{suffix}"], + ) + + output = " ".join(strip_ansi(result.output).split()) + assert result.exit_code == 1, output + assert isinstance(result.exception, SystemExit), repr(result.exception) + assert diagnostic in output + assert "not authenticated" not in output + assert "Traceback" not in output + install.assert_not_called() + assert not (project_dir / ".specify" / "extensions" / "unsafe").exists() + @pytest.mark.parametrize("command", [ pytest.param(["speckit.test-ext.hello"], id="list"), pytest.param({"name": "speckit.test-ext.hello"}, id="mapping"), diff --git a/tests/specify_cli/workflows/test_command_add.py b/tests/specify_cli/workflows/test_command_add.py index 263fe79add..6b82ea48b3 100644 --- a/tests/specify_cli/workflows/test_command_add.py +++ b/tests/specify_cli/workflows/test_command_add.py @@ -18,6 +18,8 @@ from typer.testing import CliRunner from specify_cli import app +from tests.archive_helpers import tar_safety_archive +from tests.conftest import strip_ansi runner = CliRunner() @@ -1382,6 +1384,59 @@ def test_add_from_suffixless_url_sniffs_archive( installed = project_dir / ".specify" / "workflows" / "align-wf" assert (installed / "assets" / "sniffed.txt").read_text() == "sniffed\n" + @pytest.mark.parametrize("source", ["url", "catalog"]) + @pytest.mark.parametrize( + "kind, diagnostic", + [ + ("stream", "maximum decompressed"), + ("metadata", "metadata payload exceeds"), + ("gnu-sparse", "Unsupported GNU sparse"), + ("pax-sparse-1.0", "Unsupported GNU sparse"), + ("global-pax", "global PAX metadata"), + ("metadata-chain", "metadata chain"), + ("negative-metadata", "negative size"), + ("total-metadata", "archive metadata exceeds"), + ], + ) + def test_suffixless_download_preserves_archive_safety_errors( + self, project_dir, monkeypatch, source, kind, diagnostic + ): + from unittest.mock import patch + from specify_cli import _download_security + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry + + monkeypatch.chdir(project_dir) + if kind == "stream": + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 2048) + url = "https://example.com/assets/12345" + data = tar_safety_archive(kind) + info = { + "id": "align-wf", "name": "Align Workflow", "version": "1.0.0", + "url": url, "_install_allowed": True, "_catalog_name": "test", + } + args = ["workflow", "add", "align-wf"] + if source == "url": + args += ["--from", url] + with ( + patch.object(WorkflowCatalog, "get_workflow_info", return_value=info), + patch( + "specify_cli.authentication.http.open_url", + side_effect=lambda *_args, **_kwargs: self._FakeResponse( + data, url, {"Content-Type": "application/octet-stream"}, + ), + ), + ): + result = CliRunner().invoke(app, args, input="y\n") + + output = " ".join(strip_ansi(result.output).split()) + assert result.exit_code == 1, output + assert isinstance(result.exception, SystemExit), repr(result.exception) + assert diagnostic in output + assert "Invalid workflow YAML" not in output + assert "Traceback" not in output + assert not WorkflowRegistry(project_dir).is_installed("align-wf") + assert not (project_dir / ".specify" / "workflows" / "align-wf").exists() + @pytest.mark.parametrize("suffix", [".zip", ".tar.gz", ".tgz"]) def test_add_catalog_installs_complete_archive_package_and_sha( self, project_dir, monkeypatch, suffix diff --git a/tests/test_download_security.py b/tests/test_download_security.py index 6f47b06cb5..1d069852cc 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -1,7 +1,8 @@ -"""Tests for bounded download and ZIP extraction helpers.""" +"""Tests for bounded download and archive extraction helpers.""" from __future__ import annotations +import gzip import io import stat import struct @@ -12,6 +13,7 @@ import pytest +from specify_cli import _download_security from specify_cli._download_security import ( MAX_ZIP_CENTRAL_DIRECTORY_BYTES, archive_format_from_content_type, @@ -26,6 +28,639 @@ safe_extract_tar, safe_extract_zip, ) +from tests.archive_helpers import ( + tar_global_pax_archive, + tar_metadata_chain_archive, + tar_metadata_record, + tar_negative_metadata_archive, + tar_safety_archive, +) + + +@pytest.mark.parametrize("operation", ["detect", "extract", "combined"]) +@pytest.mark.parametrize( + "kind, diagnostic", + [ + ("global-pax", "global PAX.*cumulative size.*8192"), + ("metadata-chain", "metadata chain.*maximum depth.*32"), + ("total-metadata", "archive metadata.*cumulative size.*1048576"), + ], +) +def test_tar_archive_metadata_budgets_reject_before_parsing( + tmp_path, monkeypatch, operation, kind, diagnostic +): + archive_path = tmp_path / "metadata.tar.gz" + archive_path.write_bytes(tar_safety_archive(kind)) + out_dir = tmp_path / "out" + original_parser = tarfile.TarInfo._proc_pax + parsed_global_bytes = 0 + active_depth = 0 + + def bounded_parser(member, archive): + nonlocal parsed_global_bytes, active_depth + if member.type == tarfile.XGLTYPE: + parsed_global_bytes += member.size + assert parsed_global_bytes <= 8192, "Global metadata reached parser" + active_depth += 1 + assert active_depth <= 32, "Long metadata chain reached parser" + try: + return original_parser(member, archive) + finally: + active_depth -= 1 + + monkeypatch.setattr(tarfile.TarInfo, "_proc_pax", bounded_parser) + with archive_path.open("rb") as source: + with pytest.raises(_CustomZipError, match=diagnostic): + if operation == "detect": + detect_archive_format( + archive_path, archive_file=source, error_type=_CustomZipError + ) + else: + extractor = ( + safe_extract_tar if operation == "extract" else safe_extract_archive + ) + extractor( + archive_path, out_dir, archive_file=source, + error_type=_CustomZipError, + ) + assert not source.closed + assert not out_dir.exists() + + +@pytest.mark.parametrize("metadata_type", [tarfile.XHDTYPE, tarfile.SOLARIS_XHDTYPE]) +@pytest.mark.parametrize("over_budget", [False, True]) +def test_per_member_metadata_cumulative_boundary( + tmp_path, monkeypatch, metadata_type, over_budget +): + plain = b"" + for index, size in enumerate([65536] * 15 + [65535]): + plain += tar_metadata_record(metadata_type, size) + plain += tarfile.TarInfo(f"file{index}.txt").tobuf() + plain += tar_metadata_record(tarfile.GNUTYPE_LONGLINK, 2 if over_budget else 1) + plain += tarfile.TarInfo("last.txt").tobuf() + archive_path = tmp_path / "per-member.tar.gz" + archive_path.write_bytes(gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE)) + original_parser = tarfile.TarInfo._proc_member + parsed_bytes = {} + + def bounded_parser(member, archive): + if member.type in ( + tarfile.XHDTYPE, tarfile.SOLARIS_XHDTYPE, tarfile.GNUTYPE_LONGLINK + ): + parsed_bytes[archive] = parsed_bytes.get(archive, 0) + member.size + assert parsed_bytes[archive] <= 1048576, "Excess metadata reached parser" + return original_parser(member, archive) + + monkeypatch.setattr(tarfile.TarInfo, "_proc_member", bounded_parser) + if over_budget: + with pytest.raises(_CustomZipError, match="archive metadata.*cumulative size.*1048576"): + safe_extract_archive( + archive_path, tmp_path / "out", error_type=_CustomZipError + ) + assert not (tmp_path / "out").exists() + else: + for index in range(2): + out_dir = tmp_path / f"out{index}" + assert safe_extract_archive(archive_path, out_dir) == "tar.gz" + assert len(list(out_dir.iterdir())) == 17 + + +@pytest.mark.parametrize( + "metadata_type", + [tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.SOLARIS_XHDTYPE, + tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK], +) +@pytest.mark.parametrize("over_budget", [False, True]) +def test_cumulative_metadata_budget_includes_every_header_type( + tmp_path, monkeypatch, metadata_type, over_budget +): + monkeypatch.setattr( + _download_security, "MAX_TAR_TOTAL_METADATA_BYTES", 64, raising=False + ) + plain = tar_metadata_record(tarfile.XGLTYPE, 32) + plain += tarfile.TarInfo("first.txt").tobuf() + plain += tar_metadata_record(metadata_type, 33 if over_budget else 32) + plain += tarfile.TarInfo("last.txt").tobuf() + archive_path = tmp_path / "mixed.tar.gz" + archive_path.write_bytes(gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE)) + if over_budget: + with pytest.raises(_CustomZipError, match="archive metadata.*cumulative size.*64"): + safe_extract_archive( + archive_path, tmp_path / "out", error_type=_CustomZipError + ) + assert not (tmp_path / "out").exists() + else: + assert safe_extract_archive(archive_path, tmp_path / "out") == "tar.gz" + assert len(list((tmp_path / "out").iterdir())) == 2 + + +@pytest.mark.parametrize("extractor", [safe_extract_tar, safe_extract_archive]) +@pytest.mark.parametrize("over_budget", [False, True]) +def test_interleaved_global_pax_budget_is_cumulative( + tmp_path, monkeypatch, extractor, over_budget +): + archive_path = tmp_path / "global.tar.gz" + sizes = [4096, 4097 if over_budget else 4096] + archive_path.write_bytes(tar_global_pax_archive(sizes, interleaved=True)) + original_parser = tarfile.TarInfo._proc_pax + parsed_bytes = {} + + def bounded_parser(member, archive): + parsed_bytes[archive] = parsed_bytes.get(archive, 0) + member.size + assert parsed_bytes[archive] <= 8192, "Excess global metadata reached parser" + return original_parser(member, archive) + + monkeypatch.setattr(tarfile.TarInfo, "_proc_pax", bounded_parser) + out_dir = tmp_path / "out" + if over_budget: + with pytest.raises(_CustomZipError, match="global PAX.*cumulative size.*8192"): + extractor(archive_path, out_dir, error_type=_CustomZipError) + assert not out_dir.exists() + else: + extractor(archive_path, out_dir) + assert (out_dir / "file0.txt").read_bytes() == b"" + assert (out_dir / "file1.txt").read_bytes() == b"" + + +@pytest.mark.parametrize( + "metadata_type", + [tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.SOLARIS_XHDTYPE, + tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK], +) +@pytest.mark.parametrize("operation", ["detect", "extract", "combined"]) +def test_negative_metadata_size_rejected_before_legacy_unsized_read( + tmp_path, monkeypatch, metadata_type, operation +): + archive_path = tmp_path / "negative.tar.gz" + data = tar_negative_metadata_archive(metadata_type) + assert gzip.decompress(data)[124] == 255 # Negative base-256 representation. + archive_path.write_bytes(data) + original_read = gzip.GzipFile.read + + def bounded_read(stream, size=-1): + assert 0 <= size <= 65536, "Negative metadata size became an unsized read" + return original_read(stream, size) + + # Older supported patch releases rounded negative sizes instead of rejecting them. + monkeypatch.setattr( + tarfile.TarInfo, "_block", + lambda _self, size: ((size + 511) // 512) * 512, + ) + monkeypatch.setattr(gzip.GzipFile, "read", bounded_read) + with pytest.raises(_CustomZipError, match="metadata.*negative size"): + if operation == "detect": + detect_archive_format(archive_path, error_type=_CustomZipError) + else: + extractor = ( + safe_extract_tar if operation == "extract" else safe_extract_archive + ) + extractor(archive_path, tmp_path / "out", error_type=_CustomZipError) + assert not (tmp_path / "out").exists() + + +@pytest.mark.parametrize("metadata_type", [tarfile.XHDTYPE, tarfile.GNUTYPE_LONGNAME]) +@pytest.mark.parametrize("count", [32, 33]) +def test_metadata_chain_depth_boundary(tmp_path, metadata_type, count): + archive_path = tmp_path / "chain.tar.gz" + archive_path.write_bytes(tar_metadata_chain_archive(count, metadata_type)) + if count > 32: + with pytest.raises(_CustomZipError, match="maximum depth.*32"): + safe_extract_archive( + archive_path, tmp_path / "out", error_type=_CustomZipError + ) + else: + assert safe_extract_archive(archive_path, tmp_path / "out") == "tar.gz" + + +def test_metadata_chain_depth_resets_between_members(tmp_path): + chain = tar_metadata_record(tarfile.XHDTYPE, 32) * 32 + plain = chain + tarfile.TarInfo("first.txt").tobuf() + plain += chain + tarfile.TarInfo("second.txt").tobuf() + archive_path = tmp_path / "chains.tar.gz" + archive_path.write_bytes(gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE)) + + assert safe_extract_archive(archive_path, tmp_path / "out") == "tar.gz" + assert (tmp_path / "out" / "first.txt").exists() + assert (tmp_path / "out" / "second.txt").exists() + + +def test_global_pax_budget_is_independent_per_archive(tmp_path): + archive_path = tmp_path / "global.tar.gz" + archive_path.write_bytes(tar_global_pax_archive([8192], interleaved=False)) + for index in range(2): + assert safe_extract_archive(archive_path, tmp_path / f"out{index}") == "tar.gz" + + +@pytest.mark.parametrize("metadata_type", [tarfile.XHDTYPE, tarfile.XGLTYPE]) +def test_zero_length_metadata_remains_supported(tmp_path, metadata_type): + plain = tar_metadata_record(metadata_type, 0) + tarfile.TarInfo("file.txt").tobuf() + archive_path = tmp_path / "empty-metadata.tar.gz" + archive_path.write_bytes(gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE)) + assert safe_extract_archive(archive_path, tmp_path / "out") == "tar.gz" + + +@pytest.mark.parametrize("kind", ["gnu-sparse", "pax-sparse-1.0"]) +@pytest.mark.parametrize( + "operation, position", + [ + ("detect", "first"), ("detect", "nested"), + ("extract", "first"), ("extract", "later"), ("extract", "nested"), + ("combined", "first"), ("combined", "later"), ("combined", "nested"), + ], +) +def test_tar_sparse_maps_rejected_before_parsing( + tmp_path, monkeypatch, kind, operation, position +): + plain = gzip.decompress(tar_safety_archive(kind)) + if position == "later": + plain = tarfile.TarInfo("safe.txt").tobuf() + plain + elif position == "nested": + plain = tarfile.TarInfo.create_pax_global_header({"comment": "normal"}) + plain + archive_path = tmp_path / "sparse.tar.gz" + archive_path.write_bytes(gzip.compress(plain)) + out_dir = tmp_path / "out" + + def reject_parser(*_args, **_kwargs): + pytest.fail("Unbounded GNU sparse map parser was called") + + monkeypatch.setattr( + tarfile.TarInfo, + "_proc_sparse" if kind == "gnu-sparse" else "_proc_gnusparse_10", + reject_parser, + ) + with archive_path.open("rb") as source: + with pytest.raises(_CustomZipError, match="Unsupported GNU sparse"): + if operation == "detect": + detect_archive_format( + archive_path, archive_file=source, error_type=_CustomZipError + ) + else: + extractor = ( + safe_extract_tar if operation == "extract" else safe_extract_archive + ) + extractor( + archive_path, out_dir, archive_file=source, + error_type=_CustomZipError, + ) + assert not source.closed + assert not out_dir.exists() + + +@pytest.mark.parametrize( + "sparse_headers", + [ + {"GNU.sparse.map": "0,1", "GNU.sparse.size": "1"}, + { + "GNU.sparse.offset": "0", "GNU.sparse.numbytes": "1", + "GNU.sparse.size": "1", + }, + ], +) +def test_tar_accepts_sparse_maps_within_bounded_pax_headers( + tmp_path, sparse_headers +): + archive_path = tmp_path / "bounded-sparse.tar.gz" + info = tarfile.TarInfo("file.txt") + info.size = 1 + info.pax_headers = sparse_headers + with tarfile.open(archive_path, "w:gz", format=tarfile.PAX_FORMAT) as archive: + archive.addfile(info, io.BytesIO(b"x")) + + out_dir = tmp_path / "out" + assert safe_extract_archive(archive_path, out_dir) == "tar.gz" + assert (out_dir / "file.txt").read_bytes() == b"x" + + +def _write_tar_metadata_archive(path, metadata_type, size, position="first"): + def metadata_record(record_type, record_size): + info = tarfile.TarInfo("metadata") + info.type = record_type + info.size = record_size + if record_type in (tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK): + payload = b"file.txt\0".ljust(record_size, b"\0") + else: + prefix = f"{record_size} comment=".encode() + payload = prefix + b"x" * (record_size - len(prefix) - 1) + b"\n" + return ( + info.tobuf(format=tarfile.USTAR_FORMAT) + + payload + + b"\0" * (-record_size % tarfile.BLOCKSIZE) + ) + + plain = b"" + if position == "later": + plain += tarfile.TarInfo("safe.txt").tobuf() + elif position == "nested": + plain += metadata_record(tarfile.XHDTYPE, 32) + plain += metadata_record(metadata_type, size) + info = tarfile.TarInfo("file.txt") + info.size = 8 + plain += info.tobuf() + b"contents".ljust(tarfile.BLOCKSIZE, b"\0") + path.write_bytes(gzip.compress(plain + b"\0" * tarfile.BLOCKSIZE)) + + +@pytest.mark.parametrize( + "metadata_type", + [ + tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.SOLARIS_XHDTYPE, + tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK, + ], +) +@pytest.mark.parametrize( + "operation, position", + [ + ("detect", "first"), ("detect", "nested"), + ("extract", "first"), ("extract", "later"), ("extract", "nested"), + ("combined", "first"), ("combined", "later"), ("combined", "nested"), + ], +) +def test_tar_metadata_payload_limit_precedes_read( + tmp_path, monkeypatch, metadata_type, operation, position +): + archive_path = tmp_path / "metadata.tar.gz" + _write_tar_metadata_archive(archive_path, metadata_type, 65537, position) + out_dir = tmp_path / "out" + original_read = gzip.GzipFile.read + + def bounded_read(stream, size=-1): + assert 0 <= size <= 65536, "Oversized metadata read reached gzip" + return original_read(stream, size) + + monkeypatch.setattr(gzip.GzipFile, "read", bounded_read) + with archive_path.open("rb") as source: + with pytest.raises(_CustomZipError, match="metadata.*maximum size.*65536"): + if operation == "detect": + detect_archive_format( + archive_path, archive_file=source, error_type=_CustomZipError + ) + else: + extractor = ( + safe_extract_tar if operation == "extract" else safe_extract_archive + ) + extractor( + archive_path, out_dir, archive_file=source, + error_type=_CustomZipError, + ) + assert not source.closed + assert not out_dir.exists() + + +@pytest.mark.parametrize( + "metadata_type", + [ + tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.SOLARIS_XHDTYPE, + tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK, + ], +) +def test_tar_accepts_metadata_at_effective_payload_limit(tmp_path, metadata_type): + archive_path = tmp_path / "metadata.tar.gz" + size = 8192 if metadata_type == tarfile.XGLTYPE else 65536 + _write_tar_metadata_archive(archive_path, metadata_type, size) + out_dir = tmp_path / "out" + + assert safe_extract_archive(archive_path, out_dir) == "tar.gz" + assert (out_dir / "file.txt").read_bytes() == b"contents" + + +def test_tiny_compressed_pax_archive_rejects_large_metadata(tmp_path): + archive_path = tmp_path / "metadata.tar.gz" + _write_tar_metadata_archive(archive_path, tarfile.XHDTYPE, 2 * 1024 * 1024) + assert archive_path.stat().st_size < 4096 + + with pytest.raises(_CustomZipError, match="metadata.*maximum size.*65536"): + safe_extract_archive( + archive_path, tmp_path / "out", error_type=_CustomZipError + ) + assert not (tmp_path / "out").exists() + + +def test_tar_metadata_limit_does_not_limit_regular_file_payload(tmp_path): + archive_path = tmp_path / "payload.tar.gz" + contents = b"x" * 1048577 + _write_tar_gz(archive_path, [("file.txt", contents)]) + out_dir = tmp_path / "out" + + assert safe_extract_archive(archive_path, out_dir) == "tar.gz" + assert (out_dir / "file.txt").read_bytes() == contents + + +@pytest.mark.parametrize("operation", ["detect", "combined"]) +@pytest.mark.parametrize("limit_kind", ["stream", "metadata"]) +def test_tar_limit_diagnostic_survives_oserror_caller_type( + tmp_path, monkeypatch, operation, limit_kind +): + class ArchiveIOError(OSError): + pass + + archive_path = tmp_path / "metadata.tar.gz" + if limit_kind == "stream": + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 2048) + size = 4096 + diagnostic = "maximum decompressed" + else: + size = 65537 + diagnostic = "metadata.*maximum size" + _write_tar_metadata_archive(archive_path, tarfile.XHDTYPE, size) + + with pytest.raises(ArchiveIOError, match=diagnostic) as exc_info: + if operation == "detect": + detect_archive_format(archive_path, error_type=ArchiveIOError) + else: + safe_extract_archive( + archive_path, tmp_path / "out", error_type=ArchiveIOError + ) + assert "Invalid archive" not in str(exc_info.value) + + +@pytest.mark.parametrize("operation", ["detect", "combined"]) +@pytest.mark.parametrize("use_stream", [False, True]) +def test_malformed_gzip_header_keeps_format_mismatch( + tmp_path, operation, use_stream +): + archive_path = tmp_path / "corrupt.tar.gz" + archive_path.write_bytes(b"\x1f\x8b\x00" + b"\0" * 7) + source = io.BytesIO(archive_path.read_bytes()) if use_stream else None + with pytest.raises(_CustomZipError, match="Archive format mismatch"): + if operation == "detect": + detect_archive_format( + archive_path, archive_file=source, error_type=_CustomZipError + ) + else: + safe_extract_archive( + archive_path, tmp_path / "out", archive_file=source, + error_type=_CustomZipError, + ) + if source is not None: + assert not source.closed + source.close() + + +@pytest.mark.parametrize("operation", ["detect", "combined"]) +@pytest.mark.parametrize("use_stream", [False, True]) +def test_zip_with_malformed_gzip_prefix_remains_supported( + tmp_path, operation, use_stream +): + archive_path = tmp_path / "prefixed.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("file.txt", b"contents") + archive_path.write_bytes(b"\x1f\x8b\x00" + b"\0" * 7 + archive_path.read_bytes()) + source = io.BytesIO(archive_path.read_bytes()) if use_stream else None + out_dir = tmp_path / "out" + + if operation == "detect": + assert detect_archive_format(archive_path, archive_file=source) == "zip" + else: + assert safe_extract_archive( + archive_path, out_dir, archive_file=source + ) == "zip" + assert (out_dir / "file.txt").read_bytes() == b"contents" + if source is not None: + assert not source.closed + source.close() + + +@pytest.mark.parametrize( + "metadata_type", + [tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK], +) +@pytest.mark.parametrize( + "operation, later", + [ + ("detect", False), + ("extract", False), + ("extract", True), + ("combined", False), + ("combined", True), + ], +) +def test_tar_metadata_is_bounded_before_parsing( + tmp_path, monkeypatch, metadata_type, later, operation +): + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 2048) + plain = io.BytesIO() + with tarfile.open(fileobj=plain, mode="w") as archive: + if later: + archive.addfile(tarfile.TarInfo("safe.txt"), io.BytesIO()) + info = tarfile.TarInfo("././@LongLink") + info.type = metadata_type + payload = b"4096 comment=" + b"x" * (4096 - 14) + b"\n" + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + archive.addfile(tarfile.TarInfo("file.txt"), io.BytesIO()) + archive_path = tmp_path / "metadata.tar.gz" + archive_path.write_bytes(gzip.compress(plain.getvalue())) + out_dir = tmp_path / "out" + original_read = gzip.GzipFile.read + + def bounded_read(stream, size=-1): + assert 0 <= size <= 2048 + return original_read(stream, size) + + monkeypatch.setattr(gzip.GzipFile, "read", bounded_read) + + with archive_path.open("rb") as source: + with pytest.raises(_CustomZipError, match="maximum decompressed"): + if operation == "detect": + detect_archive_format( + archive_path, archive_file=source, error_type=_CustomZipError + ) + else: + extractor = ( + safe_extract_tar if operation == "extract" else safe_extract_archive + ) + extractor( + archive_path, out_dir, archive_file=source, + error_type=_CustomZipError, + ) + assert not source.closed + assert not out_dir.exists() + + +@pytest.mark.parametrize("archive_format", [tarfile.PAX_FORMAT, tarfile.GNU_FORMAT]) +def test_tar_accepts_bounded_metadata_and_preserves_caller_stream( + tmp_path, monkeypatch, archive_format +): + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 4096) + archive_path = tmp_path / "metadata.tar.gz" + name = "a" * 150 + "/" + "b" * 150 + with tarfile.open(archive_path, "w:gz", format=archive_format) as archive: + info = tarfile.TarInfo(name) + info.size = 8 + if archive_format == tarfile.PAX_FORMAT: + info.pax_headers = {"comment": "normal metadata"} + archive.addfile(info, io.BytesIO(b"contents")) + second = tarfile.TarInfo("second.txt") + second.size = 6 + archive.addfile(second, io.BytesIO(b"second")) + out_dir = tmp_path / "out" + + with archive_path.open("rb") as source: + assert safe_extract_archive( + archive_path, out_dir, archive_file=source + ) == "tar.gz" + assert not source.closed + source.seek(0) + assert detect_archive_format(archive_path, archive_file=source) == "tar.gz" + assert source.tell() == 0 + assert (out_dir / name).read_bytes() == b"contents" + assert (out_dir / "second.txt").read_bytes() == b"second" + + +@pytest.mark.parametrize("excess_bytes", [0, 1, 513]) +def test_tar_stream_budget_includes_headers_padding_and_skipped_data( + tmp_path, monkeypatch, excess_bytes +): + info = tarfile.TarInfo("file.txt") + info.size = 1 + plain = info.tobuf() + b"x" + b"\0" * 511 + b"\0" * 512 + monkeypatch.setattr( + _download_security, "MAX_TAR_STREAM_BYTES", len(plain) - excess_bytes + ) + archive_path = tmp_path / "budget.tar.gz" + archive_path.write_bytes(gzip.compress(plain)) + out_dir = tmp_path / "out" + + if excess_bytes: + with pytest.raises(_CustomZipError, match="maximum decompressed"): + safe_extract_tar(archive_path, out_dir, error_type=_CustomZipError) + assert not out_dir.exists() + else: + safe_extract_tar(archive_path, out_dir) + assert (out_dir / "file.txt").read_bytes() == b"x" + + +@pytest.mark.parametrize("access", ["read", "seek", "relative_seek", "seek_end"]) +def test_tar_reader_rejects_unbounded_access_before_decompression( + tmp_path, monkeypatch, access +): + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 16) + archive_path = tmp_path / "stream.gz" + archive_path.write_bytes(gzip.compress(b"x" * 32)) + with _download_security._BoundedTarGzipFile(archive_path, "rb") as source: + assert source.read(8) == b"x" * 8 + if access == "seek_end": + with pytest.raises(io.UnsupportedOperation): + source.seek(0, io.SEEK_END) + else: + with pytest.raises(tarfile.ReadError, match="maximum decompressed"): + if access == "read": + source.read(9) + elif access == "seek": + source.seek(17) + else: + source.seek(9, io.SEEK_CUR) + assert source.tell() == 8 + assert source.read(8) == b"x" * 8 + + +def test_tar_reader_bounds_unsized_reads(tmp_path, monkeypatch): + monkeypatch.setattr(_download_security, "MAX_TAR_STREAM_BYTES", 16) + archive_path = tmp_path / "stream.gz" + archive_path.write_bytes(gzip.compress(b"x" * 32)) + with _download_security._BoundedTarGzipFile(archive_path, "rb") as source: + with pytest.raises(tarfile.ReadError, match="maximum decompressed"): + source.read() + assert gzip.GzipFile.seek(source, 0, io.SEEK_CUR) == 17 @pytest.mark.parametrize(