From 54d5f0c35634bab343a95589797e1af8c8af3c4b Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 24 Jun 2026 00:55:21 +0200 Subject: [PATCH 1/3] harden: secure extension and preset archive downloads Adopt the shared download-security primitives from #3140 across extension and preset catalog, package, direct-URL, and ZIP-install flows: - bound catalog, package, and inline manifest reads; - verify catalog SHA-256 values when present; - replace path-only extraction with bounded traversal/symlink-safe extraction; - validate malformed hosts and ports before opening download URLs; - handle normalized trailing-backslash directory entries consistently. Redirect enforcement and checksum verification remain owned by the shared helpers already on main; this commit wires them into extension and preset behavior. Assisted-by: OpenAI Codex (model: GPT-5, autonomous) --- src/specify_cli/_download_security.py | 240 +++++++++++++++++++++++- src/specify_cli/extensions/__init__.py | 46 +++-- src/specify_cli/extensions/_commands.py | 19 +- src/specify_cli/presets/__init__.py | 43 +++-- src/specify_cli/presets/_commands.py | 28 +-- tests/test_download_security.py | 164 +++++++++++++++- tests/test_extensions.py | 38 ++-- tests/test_presets.py | 34 ++-- 8 files changed, 513 insertions(+), 99 deletions(-) diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index b4f6d28add..05bed1be76 100644 --- a/src/specify_cli/_download_security.py +++ b/src/specify_cli/_download_security.py @@ -1,10 +1,14 @@ -"""Helpers for bounded HTTP downloads.""" +"""Helpers for bounded downloads and archive extraction.""" from __future__ import annotations import io +import re import socket +import stat +import zipfile from ipaddress import IPv4Address, IPv6Address, ip_address +from pathlib import Path, PurePosixPath from typing import NoReturn, TypeVar from urllib.parse import ParseResult, urlparse @@ -12,17 +16,24 @@ ErrorT = TypeVar("ErrorT", bound=Exception) MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024 +MAX_ZIP_ENTRIES = 512 +MAX_ZIP_MEMBER_BYTES = 10 * 1024 * 1024 +MAX_ZIP_TOTAL_BYTES = 50 * 1024 * 1024 READ_CHUNK_SIZE = 64 * 1024 -# Tighter ceiling for responses that are read fully into memory and parsed as +# Tighter ceilings for responses that are read fully into memory and parsed as # JSON. The 50 MiB MAX_DOWNLOAD_BYTES default is sized for archive/payload -# downloads; JSON metadata responses are far smaller, so capping them close to -# their real size shrinks the memory-DoS surface and keeps the "too large" -# error reachable (rather than only triggering on tens of MiB). Pass it +# downloads; JSON responses are far smaller, so capping them close to their real +# size shrinks the memory-DoS surface and keeps the "too large" error reachable +# (rather than only triggering on tens of MiB). Pass the matching constant # explicitly at each JSON call site so the intended bound is pinned there. -# METADATA covers fixed-shape single-object responses (an OAuth token, one -# release's metadata): a few KiB in practice, 1 MiB is already generous. +# * METADATA - fixed-shape single-object responses (an OAuth token, one +# release's metadata): a few KiB in practice, 1 MiB is already generous. +# * CATALOG - listings that grow with the number of published items. The +# largest bundled catalog is ~130 KiB today, so 8 MiB leaves ~60x headroom +# for growth while staying well under the download ceiling. MAX_JSON_METADATA_BYTES = 1 * 1024 * 1024 +MAX_JSON_CATALOG_BYTES = 8 * 1024 * 1024 def _ip_address_without_scope( @@ -179,6 +190,10 @@ def _raise(error_type: type[ErrorT], message: str) -> NoReturn: raise error_type(message) +def _raise_from(error_type: type[ErrorT], message: str, exc: Exception) -> NoReturn: + raise error_type(message) from exc + + def read_response_limited( response, *, @@ -216,3 +231,214 @@ def read_response_limited( _raise(error_type, f"{label} exceeds maximum size of {max_bytes} bytes") output.write(chunk) return output.getvalue() + + +def read_zip_member_limited( + zf: zipfile.ZipFile, + name: str, + *, + max_bytes: int = MAX_ZIP_MEMBER_BYTES, + error_type: type[ErrorT] = ValueError, + label: str | None = None, +) -> bytes: + """Read a single ZIP member into memory under a hard size cap. + + Reading a member with ``zf.open(name).read()`` is unbounded: a crafted + archive can declare a tiny ``file_size`` yet decompress to many gigabytes (a + "zip bomb"), exhausting memory before the caller ever inspects the data. + This rejects members whose *declared* size already exceeds *max_bytes* and, + to defend against headers that lie, also reads in bounded chunks and stops + one byte past the limit. + + Use this for any inline manifest/metadata read that happens *before* + :func:`safe_extract_zip` (which already enforces the same per-member bound + during extraction); a raw ``zf.open(...).read()`` bypasses that protection. + """ + member_label = label or name + try: + info = zf.getinfo(name) + except KeyError as exc: + _raise_from(error_type, f"ZIP member not found: {name}", exc) + if info.file_size > max_bytes: + _raise( + error_type, + f"ZIP member {member_label} exceeds maximum size of {max_bytes} bytes", + ) + + chunks: list[bytes] = [] + total = 0 + limit = max_bytes + 1 + try: + with zf.open(name, "r") as source: + while total < limit: + chunk = source.read(min(READ_CHUNK_SIZE, limit - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + except (OSError, zipfile.BadZipFile, RuntimeError) as exc: + _raise_from(error_type, f"Failed to read ZIP member {member_label}: {exc}", exc) + if total > max_bytes: + _raise( + error_type, + f"ZIP member {member_label} exceeds maximum size of {max_bytes} bytes", + ) + return b"".join(chunks) + + +def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: + """Return a normalized ZIP member name or raise on traversal.""" + if "\x00" in name: + _raise(error_type, f"Unsafe path in ZIP archive: {name!r}") + + normalized = name.replace("\\", "/") + path = PurePosixPath(normalized) + raw_parts = normalized.split("/") + # Strip a single trailing empty segment, i.e. the one-slash directory + # marker that legitimate ZIPs use ("mydir/", "mydir/subdir/"). Anything + # else that produces an empty segment - consecutive slashes ("a//b") or a + # second trailing slash - is left in place and rejected below as malformed. + if raw_parts and raw_parts[-1] == "": + raw_parts = raw_parts[:-1] + has_windows_drive = re.match(r"^[A-Za-z]:", normalized) is not None + if ( + not raw_parts + or path.is_absolute() + or has_windows_drive + or any(part in {"", ".", ".."} for part in raw_parts) + ): + _raise( + error_type, + f"Unsafe path in ZIP archive: {name} (potential path traversal)", + ) + return normalized + + +def safe_extract_zip( + zip_path: Path, + target_dir: Path, + *, + error_type: type[ErrorT] = ValueError, + max_entries: int = MAX_ZIP_ENTRIES, + max_member_bytes: int = MAX_ZIP_MEMBER_BYTES, + max_total_bytes: int = MAX_ZIP_TOTAL_BYTES, +) -> None: + """Extract a ZIP archive after path, symlink, and size validation.""" + try: + target_root = target_dir.resolve() + except OSError as exc: + _raise_from(error_type, f"Invalid ZIP extraction target: {target_dir}", exc) + + try: + zf = zipfile.ZipFile(zip_path, "r") + except (OSError, zipfile.BadZipFile) as exc: + _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) + + with zf: + try: + members = zf.infolist() + except zipfile.BadZipFile as exc: + _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) + if len(members) > max_entries: + _raise( + error_type, + f"ZIP archive contains too many entries ({len(members)} > {max_entries})", + ) + + normalized_members: list[tuple[zipfile.ZipInfo, str, bool]] = [] + total_size = 0 + for member in members: + normalized_name = _safe_zip_name(member.filename, error_type=error_type) + is_dir = member.is_dir() or normalized_name.endswith("/") + + mode = member.external_attr >> 16 + if stat.S_ISLNK(mode): + _raise(error_type, f"Unsafe symlink in ZIP archive: {member.filename}") + + member_path = (target_dir / normalized_name).resolve() + try: + member_path.relative_to(target_root) + except ValueError: + _raise( + error_type, + f"Unsafe path in ZIP archive: {member.filename} " + "(potential path traversal)", + ) + + if not is_dir: + if member.file_size > max_member_bytes: + _raise( + error_type, + f"ZIP member {member.filename} exceeds maximum size " + f"of {max_member_bytes} bytes", + ) + total_size += member.file_size + if total_size > max_total_bytes: + _raise( + error_type, + f"ZIP archive exceeds maximum uncompressed size " + f"of {max_total_bytes} bytes", + ) + + normalized_members.append((member, normalized_name, is_dir)) + + # The loop above bounds the *declared* total via member.file_size, but a + # crafted archive can understate those headers. Mirror the per-member + # guard below with a cumulative count of the bytes actually written so + # the total-size bound holds even when the headers lie. + total_written = 0 + for member, normalized_name, is_dir in normalized_members: + member_path = target_dir / normalized_name + if is_dir: + try: + member_path.mkdir(parents=True, exist_ok=True) + except OSError as exc: + _raise_from( + error_type, + f"Failed to create ZIP directory {member.filename}: {exc}", + exc, + ) + continue + + try: + member_path.parent.mkdir(parents=True, exist_ok=True) + except OSError as exc: + _raise_from( + error_type, + f"Failed to create parent directory for ZIP member {member.filename}: {exc}", + exc, + ) + written = 0 + # Raised outside the try below: if error_type subclasses OSError or + # RuntimeError, raising inside would re-wrap the limit error as + # "Failed to extract" and lose the size-bound message. + limit_error: str | None = None + try: + with zf.open(member, "r") as source, member_path.open("wb") as dest: + while True: + chunk = source.read(READ_CHUNK_SIZE) + if not chunk: + break + written += len(chunk) + if written > max_member_bytes: + limit_error = ( + f"ZIP member {member.filename} exceeds maximum size " + f"of {max_member_bytes} bytes" + ) + break + total_written += len(chunk) + if total_written > max_total_bytes: + limit_error = ( + f"ZIP archive exceeds maximum uncompressed size " + f"of {max_total_bytes} bytes" + ) + break + dest.write(chunk) + except (OSError, zipfile.BadZipFile, RuntimeError) as exc: + _raise_from( + error_type, + f"Failed to extract ZIP member {member.filename}: {exc}", + exc, + ) + if limit_error is not None: + _raise(error_type, limit_error) diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 00a6d92b41..2754395747 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -17,7 +17,6 @@ import shutil import stat import tempfile -import zipfile from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -29,6 +28,11 @@ from packaging.specifiers import InvalidSpecifier, SpecifierSet from .._assets import _locate_core_pack, _repo_root +from .._download_security import ( + MAX_JSON_CATALOG_BYTES, + read_response_limited, + safe_extract_zip, +) from .._init_options import is_ai_skills_enabled from .._invocation_style import is_dollar_skills_agent, is_slash_skills_agent from .._utils import dump_frontmatter, relative_extension_path_violation, version_satisfies @@ -2053,21 +2057,7 @@ def install_from_zip( with tempfile.TemporaryDirectory() as tmpdir: temp_path = Path(tmpdir) - # Extract ZIP safely (prevent Zip Slip attack) - with zipfile.ZipFile(zip_path, "r") as zf: - # Validate all paths first before extracting anything - temp_path_resolved = temp_path.resolve() - for member in zf.namelist(): - member_path = (temp_path / member).resolve() - # Use is_relative_to for safe path containment check - try: - member_path.relative_to(temp_path_resolved) - except ValueError: - raise ValidationError( - f"Unsafe path in ZIP archive: {member} (potential path traversal)" - ) - # Only extract after all paths are validated - zf.extractall(temp_path) + safe_extract_zip(zip_path, temp_path, error_type=ValidationError) # Find extension directory (may be nested) extension_dir = temp_path @@ -2862,7 +2852,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != entry.url: self._validate_catalog_url(final_url) - catalog_data = json.loads(response.read()) + catalog_data = json.loads( + read_response_limited( + response, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=ExtensionError, + label=f"extension catalog {entry.url}", + ) + ) self._validate_catalog_payload(catalog_data, entry.url) @@ -3050,7 +3047,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != catalog_url: self._validate_catalog_url(final_url) - catalog_data = json.loads(response.read()) + catalog_data = json.loads( + read_response_limited( + response, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=ExtensionError, + label=f"extension catalog {catalog_url}", + ) + ) # Validate catalog structure. Reuses the same helper as # ``_fetch_single_catalog`` so all three branches (root type, @@ -3238,7 +3242,11 @@ def download_extension( with self._open_url( download_url, timeout=60, extra_headers=extra_headers ) as response: - zip_data = response.read() + zip_data = read_response_limited( + response, + error_type=ExtensionError, + label=f"extension '{extension_id}' download", + ) verify_archive_sha256( zip_data, ext_info.get("sha256"), extension_id, ExtensionError diff --git a/src/specify_cli/extensions/_commands.py b/src/specify_cli/extensions/_commands.py index 7ae4f8e9f0..a5a073d2f1 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -23,6 +23,7 @@ from .._console import console from .._assets import get_speckit_version +from .._download_security import read_zip_member_limited extension_app = typer.Typer( name="extension", @@ -1229,19 +1230,25 @@ def extension_update( manifest_data = None namelist = zf.namelist() + # Read the manifest under a hard size cap: this happens + # before install_from_zip()'s safe_extract_zip(), so a + # raw zf.open().read() here would bypass that bound and + # let a zip-bomb extension.yml exhaust memory. # First try root-level extension.yml if "extension.yml" in namelist: - with zf.open("extension.yml") as f: - parsed_manifest = yaml.safe_load(f) - manifest_data = parsed_manifest if parsed_manifest is not None else {} + parsed_manifest = yaml.safe_load( + read_zip_member_limited(zf, "extension.yml") + ) + manifest_data = parsed_manifest if parsed_manifest is not None else {} else: # Look for extension.yml in a single top-level subdirectory # (e.g., "repo-name-branch/extension.yml") manifest_paths = [n for n in namelist if n.endswith("/extension.yml") and n.count("/") == 1] if len(manifest_paths) == 1: - with zf.open(manifest_paths[0]) as f: - parsed_manifest = yaml.safe_load(f) - manifest_data = parsed_manifest if parsed_manifest is not None else {} + parsed_manifest = yaml.safe_load( + read_zip_member_limited(zf, manifest_paths[0]) + ) + manifest_data = parsed_manifest if parsed_manifest is not None else {} if manifest_data is None: raise ValueError("Downloaded extension archive is missing 'extension.yml'") diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index 52494fede7..58d938df0d 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -12,7 +12,6 @@ import hashlib import os import tempfile -import zipfile import shutil from dataclasses import dataclass from pathlib import Path @@ -27,6 +26,11 @@ from packaging import version as pkg_version from packaging.specifiers import SpecifierSet, InvalidSpecifier +from .._download_security import ( + MAX_JSON_CATALOG_BYTES, + read_response_limited, + safe_extract_zip, +) from ..extensions import REINSTALL_COMMAND, ExtensionRegistry, normalize_priority from .._init_options import is_ai_skills_enabled from ..integrations.base import IntegrationBase @@ -1856,18 +1860,7 @@ def install_from_zip( with tempfile.TemporaryDirectory() as tmpdir: temp_path = Path(tmpdir) - with zipfile.ZipFile(zip_path, 'r') as zf: - temp_path_resolved = temp_path.resolve() - for member in zf.namelist(): - member_path = (temp_path / member).resolve() - try: - member_path.relative_to(temp_path_resolved) - except ValueError: - raise PresetValidationError( - f"Unsafe path in ZIP archive: {member} " - "(potential path traversal)" - ) - zf.extractall(temp_path) + safe_extract_zip(zip_path, temp_path, error_type=PresetValidationError) pack_dir = temp_path manifest_path = pack_dir / "preset.yml" @@ -2451,7 +2444,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != entry.url: self._validate_catalog_url(final_url) - catalog_data = json.loads(response.read()) + catalog_data = json.loads( + read_response_limited( + response, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=PresetError, + label=f"preset catalog {entry.url}", + ) + ) self._validate_catalog_payload(catalog_data, entry.url) @@ -2613,7 +2613,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != catalog_url: self._validate_catalog_url(final_url) - catalog_data = json.loads(response.read()) + catalog_data = json.loads( + read_response_limited( + response, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=PresetError, + label=f"preset catalog {catalog_url}", + ) + ) # Validate catalog structure. Reuses the same helper as # ``_fetch_single_catalog`` so all three branches (root type, @@ -2814,7 +2821,11 @@ def download_pack( try: with self._open_url(download_url, timeout=60, extra_headers=extra_headers) as response: - zip_data = response.read() + zip_data = read_response_limited( + response, + error_type=PresetError, + label=f"preset '{pack_id}' download", + ) verify_archive_sha256( zip_data, pack_info.get("sha256"), pack_id, PresetError diff --git a/src/specify_cli/presets/_commands.py b/src/specify_cli/presets/_commands.py index ba2b85b351..2b279e994f 100644 --- a/src/specify_cli/presets/_commands.py +++ b/src/specify_cli/presets/_commands.py @@ -19,6 +19,7 @@ from .._download_security import ( is_https_or_localhost_http, is_safe_download_redirect, + read_response_limited, ) preset_app = typer.Typer( @@ -126,15 +127,15 @@ def _validate_download_redirect(old_url, new_url): if not is_https_or_localhost_http(from_url): console.print( - "[red]Error:[/red] URL must use HTTPS with a hostname, " - "or HTTP for localhost/loopback." + "[red]Error:[/red] URL must use HTTPS with a hostname and be " + "a valid URL with a host. HTTP is only allowed for localhost, " + "127.0.0.1, and ::1." ) raise typer.Exit(1) console.print(f"Installing preset from [cyan]{_escape_markup(from_url)}[/cyan]...") import urllib.error import tempfile - import shutil with tempfile.TemporaryDirectory() as tmpdir: zip_path = Path(tmpdir) / "preset.zip" @@ -162,16 +163,21 @@ def _validate_download_redirect(old_url, new_url): console.print( "[red]Error:[/red] Preset URL redirected to a disallowed URL: " f"{final_url}. Redirect targets must use HTTPS with a hostname, " - "or HTTP for localhost/loopback." + "or HTTP for localhost (127.0.0.1, ::1)." ) raise typer.Exit(1) - with zip_path.open("wb") as output: - try: - shutil.copyfileobj(response, output) - except TypeError: - output.write(response.read()) - except urllib.error.URLError as e: - console.print(f"[red]Error:[/red] Failed to download: {_escape_markup(str(e))}") + zip_path.write_bytes( + read_response_limited( + response, + error_type=PresetError, + label=f"preset {from_url}", + ) + ) + except (urllib.error.URLError, PresetError) as e: + console.print( + f"[red]Error:[/red] Failed to download: " + f"{_escape_markup(str(e))}" + ) raise typer.Exit(1) manifest = manager.install_from_zip(zip_path, speckit_version, priority) diff --git a/tests/test_download_security.py b/tests/test_download_security.py index f5dffe10de..f47a97ded4 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -1,8 +1,10 @@ -"""Tests for bounded HTTP download helpers.""" +"""Tests for bounded download and ZIP extraction helpers.""" from __future__ import annotations +import stat import weakref +import zipfile import pytest @@ -10,6 +12,8 @@ is_https_or_localhost_http, is_loopback_url, read_response_limited, + read_zip_member_limited, + safe_extract_zip, ) @@ -112,8 +116,6 @@ class _Response: def __init__(self, data: bytes, *, chunk: int | None = None): self.data = data self.pos = 0 - # When set, never return more than *chunk* bytes per call even if more is - # requested - simulates short reads (e.g. chunked transfer encoding). self.chunk = chunk def read(self, size: int = -1) -> bytes: @@ -161,6 +163,10 @@ def read(self, _size: int = -1) -> bytes | _TrackedChunk: return chunk +class _CustomZipError(ValueError): + pass + + def test_read_response_limited_rejects_oversized_download(): with pytest.raises(ValueError, match="exceeds maximum size"): read_response_limited(_Response(b"abcde"), max_bytes=4) @@ -171,9 +177,6 @@ def test_read_response_limited_returns_full_body_within_limit(): def test_read_response_limited_enforces_bound_under_short_reads(): - # A server that streams more than max_bytes total while every read() returns - # fewer bytes than requested (chunked encoding) must still be rejected - a - # single read(max_bytes + 1) could be fooled, the accumulating loop cannot. response = _Response(b"x" * 100, chunk=8) with pytest.raises(ValueError, match="exceeds maximum size"): read_response_limited(response, max_bytes=16) @@ -225,3 +228,152 @@ def test_read_response_limited_rejects_first_byte_at_zero_limit(): max_bytes=0, error_type=_CustomLimitError, ) + + +@pytest.mark.parametrize( + "member_name", + [ + "../evil.txt", + "nested/../../evil.txt", + "nested\\..\\evil.txt", + "C:\\Windows\\evil.txt", + "C:drive-relative.txt", + ], +) +def test_safe_extract_zip_rejects_traversal(tmp_path, member_name): + zip_path = tmp_path / "bad.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(member_name, "nope") + + with pytest.raises(ValueError, match="Unsafe path"): + safe_extract_zip(zip_path, tmp_path / "out") + + +@pytest.mark.parametrize("member_name", [".", "./file.txt", "nested/./file.txt", "nested//file.txt"]) +def test_safe_extract_zip_rejects_dot_path_segments(tmp_path, member_name): + zip_path = tmp_path / "bad.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(member_name, "nope") + + with pytest.raises(_CustomZipError, match="Unsafe path"): + safe_extract_zip(zip_path, tmp_path / "out", error_type=_CustomZipError) + + +def test_safe_extract_zip_rejects_symlinks(tmp_path): + zip_path = tmp_path / "bad.zip" + info = zipfile.ZipInfo("link") + info.external_attr = (stat.S_IFLNK | 0o777) << 16 + + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(info, "target") + + with pytest.raises(ValueError, match="Unsafe symlink"): + safe_extract_zip(zip_path, tmp_path / "out") + + +def test_safe_extract_zip_rejects_symlink_without_partial_extraction(tmp_path): + zip_path = tmp_path / "mixed.zip" + link = zipfile.ZipInfo("evil-link") + link.external_attr = (stat.S_IFLNK | 0o777) << 16 + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("safe/first.txt", "hello") + zf.writestr(link, "target") + zf.writestr("safe/second.txt", "world") + + out_dir = tmp_path / "out" + with pytest.raises(ValueError, match="Unsafe symlink"): + safe_extract_zip(zip_path, out_dir) + + assert not out_dir.exists() or not any(out_dir.rglob("*")) + + +def test_safe_extract_zip_rejects_oversized_member(tmp_path): + zip_path = tmp_path / "bad.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("big.txt", "abcde") + + with pytest.raises(ValueError, match="exceeds maximum size"): + safe_extract_zip(zip_path, tmp_path / "out", max_member_bytes=4) + + +def test_safe_extract_zip_rejects_too_many_entries(tmp_path): + zip_path = tmp_path / "bad.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("one.txt", "1") + zf.writestr("two.txt", "2") + + with pytest.raises(ValueError, match="too many entries"): + safe_extract_zip(zip_path, tmp_path / "out", max_entries=1) + + +def test_safe_extract_zip_rejects_total_uncompressed_size(tmp_path): + zip_path = tmp_path / "bad.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("one.txt", "123") + zf.writestr("two.txt", "456") + + with pytest.raises(ValueError, match="maximum uncompressed size"): + safe_extract_zip(zip_path, tmp_path / "out", max_total_bytes=5) + + +def test_safe_extract_zip_wraps_bad_zip_file(tmp_path): + zip_path = tmp_path / "bad.zip" + zip_path.write_bytes(b"not a zip archive") + + with pytest.raises(_CustomZipError, match="Invalid ZIP archive"): + safe_extract_zip(zip_path, tmp_path / "out", error_type=_CustomZipError) + + +def test_read_zip_member_limited_returns_member_within_limit(tmp_path): + zip_path = tmp_path / "ok.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("extension.yml", "extension:\n id: demo\n") + + with zipfile.ZipFile(zip_path, "r") as zf: + data = read_zip_member_limited(zf, "extension.yml") + + assert data == b"extension:\n id: demo\n" + + +def test_read_zip_member_limited_rejects_oversized_member(tmp_path): + zip_path = tmp_path / "bomb.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("extension.yml", "a" * 5000) + + with zipfile.ZipFile(zip_path, "r") as zf: + with pytest.raises(ValueError, match="exceeds maximum size"): + read_zip_member_limited(zf, "extension.yml", max_bytes=16) + + +def test_read_zip_member_limited_wraps_missing_member(tmp_path): + zip_path = tmp_path / "ok.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("other.txt", "x") + + with zipfile.ZipFile(zip_path, "r") as zf: + with pytest.raises(_CustomZipError, match="ZIP member not found"): + read_zip_member_limited(zf, "extension.yml", error_type=_CustomZipError) + + +def test_safe_extract_zip_extracts_safe_archive(tmp_path): + zip_path = tmp_path / "ok.zip" + out_dir = tmp_path / "out" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("nested/file.txt", "hello") + + safe_extract_zip(zip_path, out_dir) + + assert (out_dir / "nested" / "file.txt").read_text(encoding="utf-8") == "hello" + + +def test_safe_extract_zip_treats_normalized_trailing_backslash_as_directory(tmp_path): + zip_path = tmp_path / "ok.zip" + out_dir = tmp_path / "out" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("nested\\", "") + zf.writestr("nested/file.txt", "hello") + + safe_extract_zip(zip_path, out_dir) + + assert (out_dir / "nested").is_dir() + assert (out_dir / "nested" / "file.txt").read_text(encoding="utf-8") == "hello" diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 61826aad5b..0b65b659e6 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -4423,7 +4423,7 @@ def test_fetch_single_catalog_sends_auth_header(self, temp_dir, monkeypatch): catalog_data = {"schema_version": "1.0", "extensions": {}} mock_response = MagicMock() - mock_response.read.return_value = json.dumps(catalog_data).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(catalog_data).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://raw.githubusercontent.com/org/repo/main/catalog.json" @@ -4567,7 +4567,7 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, temp_dir, payload) catalog = self._make_catalog(temp_dir) mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4636,7 +4636,7 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload( "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4684,7 +4684,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, temp_dir, payload): catalog = self._make_catalog(temp_dir) mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4725,7 +4725,7 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, temp_dir): "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4764,7 +4764,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, temp_dir): "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4839,7 +4839,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, temp_dir, monkeypatch): "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode("utf-8") + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode("utf-8")).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -4890,11 +4890,13 @@ def test_fetch_catalog_survives_unwritable_cache(self, temp_dir, monkeypatch): "schema_version": "1.0", "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}}, } - mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() - mock_response.__enter__ = lambda s: s - mock_response.__exit__ = MagicMock(return_value=False) - mock_response.geturl.return_value = "https://example.com/catalog.json" + def make_response(): + mock_response = MagicMock() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + mock_response.geturl.return_value = "https://example.com/catalog.json" + return mock_response # Simulate an unwritable cache dir: every write_text under the # cache directory raises PermissionError (an OSError subclass). @@ -4907,7 +4909,7 @@ def failing_write_text(self, data, *args, **kwargs): monkeypatch.setattr(_PathCls, "write_text", failing_write_text) - with patch.object(catalog, "_open_url", return_value=mock_response): + with patch.object(catalog, "_open_url", side_effect=lambda *a, **kw: make_response()): # Legacy single-catalog path. assert catalog.fetch_catalog(force_refresh=True) == valid @@ -4943,7 +4945,7 @@ def test_get_merged_extensions_skips_non_mapping_entries(self, temp_dir): }, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -5041,7 +5043,7 @@ def _mock_response(self, data): from unittest.mock import MagicMock resp = MagicMock() - resp.read.return_value = data + resp.read.side_effect = io.BytesIO(data).read # Configure the context-manager protocol explicitly so `with resp` # yields `resp` itself, independent of how the protocol is invoked. resp.__enter__.return_value = resp @@ -5145,7 +5147,7 @@ def test_download_extension_accepts_direct_github_rest_asset_url(self, temp_dir, zip_bytes = zip_buf.getvalue() asset_response = MagicMock() - asset_response.read.return_value = zip_bytes + asset_response.read.side_effect = io.BytesIO(zip_bytes).read asset_response.__enter__ = lambda s: s asset_response.__exit__ = MagicMock(return_value=False) @@ -5516,7 +5518,7 @@ def test_load_catalog_config_defaults_blank_names(self, temp_dir): @pytest.mark.parametrize( ("url", "expected_detail"), [ - ("relative/catalog.json", "HTTPS"), + ("relative/catalog.json", "valid URL with a host"), ("https:///no-host", "valid URL with a host"), ], ) @@ -7046,7 +7048,7 @@ def test_download_extension_allows_bundled_with_url(self, temp_dir): } mock_response = MagicMock() - mock_response.read.return_value = b"fake zip data" + mock_response.read.side_effect = io.BytesIO(b"fake zip data").read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" diff --git a/tests/test_presets.py b/tests/test_presets.py index 193d86b6ae..2a1c8cc7f5 100644 --- a/tests/test_presets.py +++ b/tests/test_presets.py @@ -1740,7 +1740,7 @@ def test_fetch_single_catalog_sends_auth_header(self, project_dir, monkeypatch): catalog_data = {"schema_version": "1.0", "presets": {}} mock_response = MagicMock() - mock_response.read.return_value = json.dumps(catalog_data).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(catalog_data).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://raw.githubusercontent.com/org/repo/main/presets/catalog.json" @@ -1893,7 +1893,7 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, project_dir, paylo catalog = PresetCatalog(project_dir) mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) # A real urllib response reports the final URL (== request URL with no @@ -1965,7 +1965,7 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload( "presets": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = catalog.DEFAULT_CATALOG_URL @@ -2013,7 +2013,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, project_dir, payload): catalog = PresetCatalog(project_dir) mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -2055,7 +2055,7 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, project_dir): "presets": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -2094,7 +2094,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, project_dir): "presets": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -2165,7 +2165,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, project_dir, monkeypatch): "presets": {"foo": {"name": "Foo", "version": "1.0.0"}}, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode("utf-8") + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode("utf-8")).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -2214,11 +2214,13 @@ def test_fetch_catalog_survives_unwritable_cache(self, project_dir, monkeypatch) "schema_version": "1.0", "presets": {"foo": {"name": "Foo", "version": "1.0.0"}}, } - mock_response = MagicMock() - mock_response.read.return_value = json.dumps(valid).encode() - mock_response.__enter__ = lambda s: s - mock_response.__exit__ = MagicMock(return_value=False) - mock_response.geturl.return_value = catalog.DEFAULT_CATALOG_URL + def make_response(): + mock_response = MagicMock() + mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + mock_response.geturl.return_value = catalog.DEFAULT_CATALOG_URL + return mock_response # Simulate an unwritable cache dir: every write_text under the # cache directory raises PermissionError (an OSError subclass). @@ -2231,7 +2233,7 @@ def failing_write_text(self, data, *args, **kwargs): monkeypatch.setattr(_PathCls, "write_text", failing_write_text) - with patch.object(catalog, "_open_url", return_value=mock_response): + with patch.object(catalog, "_open_url", side_effect=lambda *a, **kw: make_response()): # Legacy single-catalog path. assert catalog.fetch_catalog(force_refresh=True) == valid @@ -2268,7 +2270,7 @@ def test_get_merged_packs_skips_non_mapping_entries(self, project_dir): }, } mock_response = MagicMock() - mock_response.read.return_value = json.dumps(payload).encode() + mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read mock_response.__enter__ = lambda s: s mock_response.__exit__ = MagicMock(return_value=False) mock_response.geturl.return_value = "https://example.com/catalog.json" @@ -2361,7 +2363,7 @@ def _pack_zip_and_response(self): zip_bytes = zip_buf.getvalue() resp = MagicMock() - resp.read.return_value = zip_bytes + resp.read.side_effect = io.BytesIO(zip_bytes).read # Configure the context-manager protocol explicitly so `with resp` # yields `resp` itself, independent of how the protocol is invoked. resp.__enter__.return_value = resp @@ -2471,7 +2473,7 @@ def test_download_pack_accepts_direct_github_rest_asset_url(self, project_dir, m zip_bytes = zip_buf.getvalue() asset_response = MagicMock() - asset_response.read.return_value = zip_bytes + asset_response.read.side_effect = io.BytesIO(zip_bytes).read asset_response.__enter__ = lambda s: s asset_response.__exit__ = MagicMock(return_value=False) From b52570e97520948c3e53370c245de7d6dd28714a Mon Sep 17 00:00:00 2001 From: Pascal Date: Thu, 23 Jul 2026 23:47:03 +0200 Subject: [PATCH 2/3] harden: close archive and catalog download edge cases Preflight ZIP central directories before ZipFile allocates them, bound both declared and actual payload sizes, and reject ambiguous or non-portable archive paths before extraction. Keep extension update manifest selection consistent with extraction, reject unsafe catalog-derived output filenames and malformed URL types, and escape untrusted values in download errors. Add regression coverage for parser differentials, collisions, platform-specific filenames, bounded call sites, and failure ordering. Assisted-by: OpenAI Codex (model: GPT-5, autonomous) --- src/specify_cli/_download_security.py | 378 +++++++++++++++++--- src/specify_cli/extensions/__init__.py | 27 +- src/specify_cli/extensions/_commands.py | 91 ++++- src/specify_cli/presets/__init__.py | 29 +- tests/test_download_security.py | 453 ++++++++++++++++++++++++ tests/test_extensions.py | 387 +++++++++++++++++++- tests/test_presets.py | 193 +++++++++- 7 files changed, 1473 insertions(+), 85 deletions(-) diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index 05bed1be76..0aa0a74c03 100644 --- a/src/specify_cli/_download_security.py +++ b/src/specify_cli/_download_security.py @@ -6,9 +6,14 @@ import re import socket import stat +import struct +import unicodedata import zipfile +from collections.abc import Iterator +from contextlib import ExitStack, contextmanager from ipaddress import IPv4Address, IPv6Address, ip_address -from pathlib import Path, PurePosixPath +from itertools import pairwise +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import NoReturn, TypeVar from urllib.parse import ParseResult, urlparse @@ -19,6 +24,11 @@ MAX_ZIP_ENTRIES = 512 MAX_ZIP_MEMBER_BYTES = 10 * 1024 * 1024 MAX_ZIP_TOTAL_BYTES = 50 * 1024 * 1024 +MAX_ZIP_PATH_BYTES = 4096 +MAX_ZIP_COMPONENT_BYTES = 255 +# ``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 READ_CHUNK_SIZE = 64 * 1024 # Tighter ceilings for responses that are read fully into memory and parsed as @@ -35,6 +45,19 @@ MAX_JSON_METADATA_BYTES = 1 * 1024 * 1024 MAX_JSON_CATALOG_BYTES = 8 * 1024 * 1024 +_WINDOWS_INVALID_FILENAME_CHARS = frozenset('<>:"|?*') +_WINDOWS_RESERVED_FILENAME = re.compile( + r"^(?:con|prn|aux|nul|conin\$|conout\$|" + r"com[1-9\u00b9\u00b2\u00b3]|lpt[1-9\u00b9\u00b2\u00b3])$", + re.IGNORECASE, +) +_ZIP_EOCD = struct.Struct("<4s4H2LH") +_ZIP_EOCD_SIGNATURE = b"PK\x05\x06" +_ZIP64_LOCATOR_SIGNATURE = b"PK\x06\x07" +_ZIP_CENTRAL_HEADER_SIZE = 46 +_ZIP_CENTRAL_SIGNATURE = b"PK\x01\x02" +_ZIP_MAX_COMMENT_BYTES = (1 << 16) - 1 + def _ip_address_without_scope( hostname: str, @@ -194,6 +217,37 @@ def _raise_from(error_type: type[ErrorT], message: str, exc: Exception) -> NoRet raise error_type(message) from exc +class _ReadLimitExceeded(Exception): + """Internal signal used to keep domain-specific errors at call sites.""" + + +def _validate_non_negative_int(value: int, name: str) -> None: + if isinstance(value, bool) or not isinstance(value, int): + raise TypeError(f"{name} must be an integer") + if value < 0: + raise ValueError(f"{name} must be non-negative") + + +def _validate_max_bytes(max_bytes: int) -> None: + _validate_non_negative_int(max_bytes, "max_bytes") + + +def _read_limited(response, max_bytes: int) -> bytes: + """Read a stream with bounded requests and without retaining fragments.""" + output = io.BytesIO() + total = 0 + limit = max_bytes + 1 + while total < limit: + chunk = response.read(min(READ_CHUNK_SIZE, limit - total)) + if not chunk: + break + total += len(chunk) + if total > max_bytes: + raise _ReadLimitExceeded + output.write(chunk) + return output.getvalue() + + def read_response_limited( response, *, @@ -214,23 +268,55 @@ def read_response_limited( explicit value so the intended bound is pinned at the call site rather than tracking changes to the shared default. """ - if isinstance(max_bytes, bool) or not isinstance(max_bytes, int): - raise TypeError("max_bytes must be an integer") - if max_bytes < 0: - raise ValueError("max_bytes must be non-negative") + _validate_max_bytes(max_bytes) + try: + return _read_limited(response, max_bytes) + except _ReadLimitExceeded: + _raise(error_type, f"{label!r} exceeds maximum size of {max_bytes} bytes") - output = io.BytesIO() - total = 0 - limit = max_bytes + 1 - while total < limit: - chunk = response.read(min(READ_CHUNK_SIZE, limit - total)) - if not chunk: - break - total += len(chunk) - if total > max_bytes: - _raise(error_type, f"{label} exceeds maximum size of {max_bytes} bytes") - output.write(chunk) - return output.getvalue() + +def build_safe_download_path( + target_dir: Path, + identifier: object, + version: object, + *, + error_type: type[ErrorT] = ValueError, + label: str = "archive", +) -> Path: + """Build a portable single-component archive path inside *target_dir*.""" + if not isinstance(identifier, str) or not isinstance(version, str): + _raise( + error_type, + f"Unsafe {label} download filename derived from " + f"{identifier!r} and {version!r}", + ) + + filename = f"{identifier}-{version}.zip" + try: + filename_too_long = ( + len(filename.encode("utf-8")) > MAX_ZIP_COMPONENT_BYTES + ) + except UnicodeEncodeError: + filename_too_long = True + posix_path = PurePosixPath(filename) + windows_path = PureWindowsPath(filename) + if ( + filename_too_long + or posix_path.name != filename + or windows_path.name != filename + or any(ord(character) < 32 for character in filename) + or any( + character in _WINDOWS_INVALID_FILENAME_CHARS + for character in filename + ) + or filename.endswith((" ", ".")) + ): + _raise( + error_type, + f"Unsafe {label} download filename derived from " + f"{identifier!r} and {version!r}", + ) + return Path(target_dir) / filename def read_zip_member_limited( @@ -254,36 +340,32 @@ def read_zip_member_limited( :func:`safe_extract_zip` (which already enforces the same per-member bound during extraction); a raw ``zf.open(...).read()`` bypasses that protection. """ + _validate_max_bytes(max_bytes) member_label = label or name try: info = zf.getinfo(name) except KeyError as exc: - _raise_from(error_type, f"ZIP member not found: {name}", exc) + _raise_from(error_type, f"ZIP member not found: {name!r}", exc) if info.file_size > max_bytes: _raise( error_type, - f"ZIP member {member_label} exceeds maximum size of {max_bytes} bytes", + f"ZIP member {member_label!r} exceeds maximum size of {max_bytes} bytes", ) - chunks: list[bytes] = [] - total = 0 - limit = max_bytes + 1 try: with zf.open(name, "r") as source: - while total < limit: - chunk = source.read(min(READ_CHUNK_SIZE, limit - total)) - if not chunk: - break - chunks.append(chunk) - total += len(chunk) - except (OSError, zipfile.BadZipFile, RuntimeError) as exc: - _raise_from(error_type, f"Failed to read ZIP member {member_label}: {exc}", exc) - if total > max_bytes: + return _read_limited(source, max_bytes) + except _ReadLimitExceeded: _raise( error_type, - f"ZIP member {member_label} exceeds maximum size of {max_bytes} bytes", + f"ZIP member {member_label!r} exceeds maximum size of {max_bytes} bytes", + ) + except Exception as exc: + _raise_from( + error_type, + f"Failed to read ZIP member {member_label!r}: {exc!r}", + exc, ) - return b"".join(chunks) def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: @@ -292,6 +374,16 @@ def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: _raise(error_type, f"Unsafe path in ZIP archive: {name!r}") normalized = name.replace("\\", "/") + try: + encoded_name = normalized.encode("utf-8") + except UnicodeEncodeError: + _raise(error_type, f"Unsafe path in ZIP archive: {name!r}") + if len(encoded_name) > MAX_ZIP_PATH_BYTES: + _raise( + error_type, + f"Unsafe path in ZIP archive: {name!r} " + "(not portable across supported filesystems)", + ) path = PurePosixPath(normalized) raw_parts = normalized.split("/") # Strip a single trailing empty segment, i.e. the one-slash directory @@ -309,11 +401,185 @@ def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: ): _raise( error_type, - f"Unsafe path in ZIP archive: {name} (potential path traversal)", + f"Unsafe path in ZIP archive: {name!r} (potential path traversal)", ) + for part in raw_parts: + reserved_stem = part.partition(".")[0].partition(":")[0].rstrip(" ") + if ( + len(part.encode("utf-8")) > MAX_ZIP_COMPONENT_BYTES + or any(ord(character) < 32 for character in part) + or any(character in _WINDOWS_INVALID_FILENAME_CHARS for character in part) + or part.startswith(" ") + or part.endswith((" ", ".")) + or _WINDOWS_RESERVED_FILENAME.fullmatch(reserved_stem) + ): + _raise( + error_type, + f"Unsafe path in ZIP archive: {name!r} " + "(not portable across supported filesystems)", + ) return normalized +def portable_zip_path_key(name: str) -> tuple[str, ...]: + """Return a comparison key for filesystems with case/Unicode folding.""" + normalized_name = name.replace("\\", "/") + return tuple( + unicodedata.normalize("NFC", part.casefold()) + for part in normalized_name.removesuffix("/").split("/") + ) + + +def _preflight_zip_central_directory( + archive_file, + zip_path: Path, + *, + error_type: type[ErrorT], + max_entries: int, +) -> None: + """Bound and count the central directory before ``ZipFile`` materializes it.""" + archive_file.seek(0, 2) + file_size = archive_file.tell() + tail_size = min(file_size, _ZIP_EOCD.size + _ZIP_MAX_COMMENT_BYTES) + archive_file.seek(file_size - tail_size) + tail = archive_file.read(tail_size) + + # ZipFile selects the last EOCD signature in the search window. Inspect + # exactly that record too: falling back to an earlier signature would let + # the preflight validate one central directory while ZipFile materializes + # another. + eocd_index = tail.rfind(_ZIP_EOCD_SIGNATURE) + if eocd_index < 0 or eocd_index + _ZIP_EOCD.size > len(tail): + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + eocd = _ZIP_EOCD.unpack_from(tail, eocd_index) + comment_size = eocd[-1] + if eocd_index + _ZIP_EOCD.size + comment_size != len(tail): + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + + eocd_offset = file_size - len(tail) + eocd_index + if eocd_offset >= 20: + archive_file.seek(eocd_offset - 20) + if archive_file.read(4) == _ZIP64_LOCATOR_SIGNATURE: + _raise( + error_type, + "ZIP64 archives are not supported by the bounded extractor", + ) + + ( + _signature, + disk_number, + central_directory_disk, + entries_on_disk, + declared_entries, + central_directory_size, + central_directory_offset, + _comment_size, + ) = eocd + if ( + disk_number != 0 + or central_directory_disk != 0 + or entries_on_disk != declared_entries + ): + _raise(error_type, "Multi-disk ZIP archives are not supported") + if ( + declared_entries == 0xFFFF + or central_directory_size == 0xFFFFFFFF + or central_directory_offset == 0xFFFFFFFF + ): + _raise( + error_type, + "ZIP64 archives are not supported by the bounded extractor", + ) + if declared_entries > max_entries: + _raise( + error_type, + f"ZIP archive contains too many entries " + f"({declared_entries} > {max_entries})", + ) + if central_directory_size > MAX_ZIP_CENTRAL_DIRECTORY_BYTES: + _raise( + error_type, + f"ZIP central directory exceeds maximum size of " + f"{MAX_ZIP_CENTRAL_DIRECTORY_BYTES} bytes", + ) + + central_directory_start = eocd_offset - central_directory_size + if ( + central_directory_start < 0 + or central_directory_offset > central_directory_start + ): + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + + archive_file.seek(central_directory_start) + consumed = 0 + actual_entries = 0 + while consumed < central_directory_size: + remaining = central_directory_size - consumed + if remaining < _ZIP_CENTRAL_HEADER_SIZE: + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + header = archive_file.read(_ZIP_CENTRAL_HEADER_SIZE) + if ( + len(header) != _ZIP_CENTRAL_HEADER_SIZE + or header[:4] != _ZIP_CENTRAL_SIGNATURE + ): + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + disk_number_start = struct.unpack_from(" remaining: + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + archive_file.seek(variable_size, 1) + consumed += record_size + actual_entries += 1 + if actual_entries > max_entries: + _raise( + error_type, + f"ZIP archive contains too many entries " + f"({actual_entries} > {max_entries})", + ) + + if actual_entries != declared_entries: + _raise(error_type, f"Invalid ZIP archive: {zip_path}") + + +@contextmanager +def open_zip_bounded( + zip_path: Path, + *, + error_type: type[ErrorT] = ValueError, + max_entries: int = MAX_ZIP_ENTRIES, +) -> Iterator[zipfile.ZipFile]: + """Open an untrusted ZIP only after an O(1)-memory central-dir preflight.""" + _validate_non_negative_int(max_entries, "max_entries") + zip_path = Path(zip_path) + with ExitStack() as stack: + try: + archive_file = stack.enter_context(zip_path.open("rb")) + except OSError as exc: + _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) + try: + _preflight_zip_central_directory( + archive_file, + zip_path, + error_type=error_type, + max_entries=max_entries, + ) + except OSError as exc: + _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) + try: + archive_file.seek(0) + zf = stack.enter_context(zipfile.ZipFile(archive_file, "r")) + except Exception as exc: + _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) + yield zf + + def safe_extract_zip( zip_path: Path, target_dir: Path, @@ -324,17 +590,18 @@ def safe_extract_zip( max_total_bytes: int = MAX_ZIP_TOTAL_BYTES, ) -> None: """Extract a ZIP archive after path, symlink, and size validation.""" + _validate_non_negative_int(max_member_bytes, "max_member_bytes") + _validate_non_negative_int(max_total_bytes, "max_total_bytes") try: target_root = target_dir.resolve() except OSError as exc: _raise_from(error_type, f"Invalid ZIP extraction target: {target_dir}", exc) - try: - zf = zipfile.ZipFile(zip_path, "r") - except (OSError, zipfile.BadZipFile) as exc: - _raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc) - - with zf: + with open_zip_bounded( + zip_path, + error_type=error_type, + max_entries=max_entries, + ) as zf: try: members = zf.infolist() except zipfile.BadZipFile as exc: @@ -346,10 +613,21 @@ def safe_extract_zip( ) normalized_members: list[tuple[zipfile.ZipInfo, str, bool]] = [] + validated_paths: dict[tuple[str, ...], tuple[str, bool]] = {} total_size = 0 for member in members: normalized_name = _safe_zip_name(member.filename, error_type=error_type) is_dir = member.is_dir() or normalized_name.endswith("/") + path_key = portable_zip_path_key(normalized_name) + + existing = validated_paths.get(path_key) + if existing is not None: + _raise( + error_type, + f"Conflicting path in ZIP archive: {member.filename} conflicts " + f"with {existing[0]}", + ) + validated_paths[path_key] = (member.filename, is_dir) mode = member.external_attr >> 16 if stat.S_ISLNK(mode): @@ -382,6 +660,24 @@ def safe_extract_zip( normalized_members.append((member, normalized_name, is_dir)) + # Tuple sorting places every path immediately before its descendants. + # One adjacent comparison per entry detects file/directory conflicts + # without repeatedly rebuilding every path prefix. + for ( + (path_key, (original, is_dir)), + (next_key, (next_original, _next_is_dir)), + ) in pairwise(sorted(validated_paths.items())): + if ( + not is_dir + and len(next_key) > len(path_key) + and next_key[: len(path_key)] == path_key + ): + _raise( + error_type, + f"Conflicting path in ZIP archive: {original} conflicts " + f"with {next_original}", + ) + # The loop above bounds the *declared* total via member.file_size, but a # crafted archive can understate those headers. Mirror the per-member # guard below with a cumulative count of the bytes actually written so @@ -434,7 +730,7 @@ def safe_extract_zip( ) break dest.write(chunk) - except (OSError, zipfile.BadZipFile, RuntimeError) as exc: + except Exception as exc: _raise_from( error_type, f"Failed to extract ZIP member {member.filename}: {exc}", diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 2754395747..969246e013 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -30,6 +30,8 @@ from .._assets import _locate_core_pack, _repo_root from .._download_security import ( MAX_JSON_CATALOG_BYTES, + build_safe_download_path, + is_https_or_localhost_http, read_response_limited, safe_extract_zip, ) @@ -3199,6 +3201,10 @@ def download_extension( download_url = ext_info.get("download_url") if not download_url: raise ExtensionError(f"Extension '{extension_id}' has no download URL") + if not isinstance(download_url, str): + raise ExtensionError( + f"Extension download URL is malformed: {download_url}" + ) # Validate download URL requires HTTPS (prevent man-in-the-middle attacks) from urllib.parse import urlparse @@ -3212,12 +3218,16 @@ def download_extension( try: parsed = urlparse(download_url) hostname = parsed.hostname + parsed.port except ValueError: raise ExtensionError( f"Extension download URL is malformed: {download_url}" ) from None - is_localhost = hostname in ("localhost", "127.0.0.1", "::1") - if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost): + if not hostname: + raise ExtensionError( + f"Extension download URL is malformed: {download_url}" + ) + if not is_https_or_localhost_http(download_url): raise ExtensionError( f"Extension download URL must use HTTPS: {download_url}" ) @@ -3225,11 +3235,16 @@ def download_extension( # Determine target path if target_dir is None: target_dir = self.cache_dir / "downloads" - target_dir.mkdir(parents=True, exist_ok=True) - + target_dir = Path(target_dir) version = ext_info.get("version", "unknown") - zip_filename = f"{extension_id}-{version}.zip" - zip_path = target_dir / zip_filename + zip_path = build_safe_download_path( + target_dir, + extension_id, + version, + error_type=ExtensionError, + label="extension", + ) + target_dir.mkdir(parents=True, exist_ok=True) extra_headers = None resolved_download_url = self._resolve_github_release_asset_api_url(download_url) diff --git a/src/specify_cli/extensions/_commands.py b/src/specify_cli/extensions/_commands.py index a5a073d2f1..e3410cca3b 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -23,7 +23,13 @@ from .._console import console from .._assets import get_speckit_version -from .._download_security import read_zip_member_limited +from .._download_security import ( + is_https_or_localhost_http, + open_zip_bounded, + portable_zip_path_key, + read_response_limited, + read_zip_member_limited, +) extension_app = typer.Typer( name="extension", @@ -436,14 +442,17 @@ def extension_add( # "Invalid URL" message instead of leaking a raw traceback past the # CLI. Reuse the value below. hostname = parsed.hostname + parsed.port except ValueError: console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(from_url)}") raise typer.Exit(1) - is_localhost = hostname in ("localhost", "127.0.0.1", "::1") + if not hostname: + console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(from_url)}") + raise typer.Exit(1) - if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost): + if not is_https_or_localhost_http(from_url): console.print("[red]Error:[/red] URL must use HTTPS for security.") - console.print("HTTP is only allowed for localhost URLs.") + console.print("HTTP is only allowed for loopback URLs.") raise typer.Exit(1) safe_url = _escape_markup(from_url) @@ -526,7 +535,11 @@ def extension_add( with dl_catalog._open_url( download_url, timeout=60, extra_headers=extra_headers ) as response: - zip_data = response.read() + zip_data = read_response_limited( + response, + error_type=ExtensionError, + label=f"extension {from_url}", + ) if not zipfile.is_zipfile(io.BytesIO(zip_data)): console.print( @@ -1225,7 +1238,7 @@ def extension_update( try: # 6. Validate extension ID from ZIP BEFORE modifying installation # Handle both root-level and nested extension.yml (GitHub auto-generated ZIPs) - with zipfile.ZipFile(zip_path, "r") as zf: + with open_zip_bounded(zip_path) as zf: import yaml manifest_data = None namelist = zf.namelist() @@ -1234,21 +1247,61 @@ def extension_update( # before install_from_zip()'s safe_extract_zip(), so a # raw zf.open().read() here would bypass that bound and # let a zip-bomb extension.yml exhaust memory. - # First try root-level extension.yml - if "extension.yml" in namelist: + # Normalize separators before choosing the manifest so + # this pre-scan cannot approve one entry while extraction + # later overwrites it with a backslash alias. + manifest_candidates = [] + for name in namelist: + normalized_name = name.replace("\\", "/") + parts = normalized_name.split("/") + path_key = portable_zip_path_key(normalized_name) + if ( + len(parts) in {1, 2} + and path_key[-1] == "extension.yml" + ): + manifest_candidates.append( + (name, normalized_name, path_key) + ) + + seen_manifest_keys = {} + for name, _normalized_name, path_key in manifest_candidates: + previous = seen_manifest_keys.get(path_key) + if previous is not None: + raise ValueError( + "Downloaded extension archive contains multiple " + "extension.yml manifests" + ) + seen_manifest_keys[path_key] = name + + root_manifest = next( + ( + name + for name, normalized_name, _path_key + in manifest_candidates + if normalized_name == "extension.yml" + ), + None, + ) + nested_manifests = [ + name + for name, normalized_name, _path_key + in manifest_candidates + if normalized_name.endswith("/extension.yml") + and normalized_name.count("/") == 1 + ] + manifest_path = root_manifest + if manifest_path is None and len(nested_manifests) == 1: + manifest_path = nested_manifests[0] + + if manifest_path is not None: parsed_manifest = yaml.safe_load( - read_zip_member_limited(zf, "extension.yml") + read_zip_member_limited(zf, manifest_path) + ) + manifest_data = ( + parsed_manifest + if parsed_manifest is not None + else {} ) - manifest_data = parsed_manifest if parsed_manifest is not None else {} - else: - # Look for extension.yml in a single top-level subdirectory - # (e.g., "repo-name-branch/extension.yml") - manifest_paths = [n for n in namelist if n.endswith("/extension.yml") and n.count("/") == 1] - if len(manifest_paths) == 1: - parsed_manifest = yaml.safe_load( - read_zip_member_limited(zf, manifest_paths[0]) - ) - manifest_data = parsed_manifest if parsed_manifest is not None else {} if manifest_data is None: raise ValueError("Downloaded extension archive is missing 'extension.yml'") diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index 58d938df0d..abe117adfe 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -28,6 +28,8 @@ from .._download_security import ( MAX_JSON_CATALOG_BYTES, + build_safe_download_path, + is_https_or_localhost_http, read_response_limited, safe_extract_zip, ) @@ -2781,6 +2783,10 @@ def download_pack( raise PresetError( f"Preset '{pack_id}' has no download URL" ) + if not isinstance(download_url, str): + raise PresetError( + f"Preset download URL is malformed: {download_url}" + ) from urllib.parse import urlparse @@ -2793,25 +2799,32 @@ def download_pack( try: parsed = urlparse(download_url) hostname = parsed.hostname + parsed.port except ValueError: raise PresetError( f"Preset download URL is malformed: {download_url}" ) from None - is_localhost = hostname in ("localhost", "127.0.0.1", "::1") - if parsed.scheme != "https" and not ( - parsed.scheme == "http" and is_localhost - ): + if not hostname: + raise PresetError( + f"Preset download URL is malformed: {download_url}" + ) + if not is_https_or_localhost_http(download_url): raise PresetError( f"Preset download URL must use HTTPS: {download_url}" ) if target_dir is None: target_dir = self.cache_dir / "downloads" - target_dir.mkdir(parents=True, exist_ok=True) - + target_dir = Path(target_dir) version = pack_info.get("version", "unknown") - zip_filename = f"{pack_id}-{version}.zip" - zip_path = target_dir / zip_filename + zip_path = build_safe_download_path( + target_dir, + pack_id, + version, + error_type=PresetError, + label="preset", + ) + target_dir.mkdir(parents=True, exist_ok=True) extra_headers = None resolved_download_url = self._resolve_github_release_asset_api_url(download_url) diff --git a/tests/test_download_security.py b/tests/test_download_security.py index f47a97ded4..f93f3680ec 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -3,12 +3,16 @@ from __future__ import annotations import stat +import struct import weakref import zipfile +import zlib import pytest from specify_cli._download_security import ( + MAX_ZIP_CENTRAL_DIRECTORY_BYTES, + build_safe_download_path, is_https_or_localhost_http, is_loopback_url, read_response_limited, @@ -162,11 +166,56 @@ def read(self, _size: int = -1) -> bytes | _TrackedChunk: ) return chunk + def __enter__(self): + return self + + def __exit__(self, _exc_type, _exc, _tb): + return False + class _CustomZipError(ValueError): pass +class _ExplodingResponse: + def read(self, _size: int = -1) -> bytes: + raise zlib.error("corrupt compressed data") + + def __enter__(self): + return self + + def __exit__(self, _exc_type, _exc, _tb): + return False + + +class _FakeZipArchive: + def __init__( + self, + response, + *, + filename: str = "extension.yml", + file_size: int = 0, + ): + self.response = response + self.info = zipfile.ZipInfo(filename) + self.info.file_size = file_size + + def __enter__(self): + return self + + def __exit__(self, _exc_type, _exc, _tb): + return False + + def getinfo(self, _name): + return self.info + + def infolist(self): + return [self.info] + + def open(self, _member, _mode="r"): + return self.response + + def test_read_response_limited_rejects_oversized_download(): with pytest.raises(ValueError, match="exceeds maximum size"): read_response_limited(_Response(b"abcde"), max_bytes=4) @@ -230,6 +279,38 @@ def test_read_response_limited_rejects_first_byte_at_zero_limit(): ) +def test_read_response_limited_escapes_control_characters_in_label(): + with pytest.raises(ValueError) as exc_info: + read_response_limited( + _Response(b"x"), + max_bytes=0, + label="bad\x1b[2J download", + ) + + assert "\x1b" not in str(exc_info.value) + assert "\\x1b" in str(exc_info.value) + + +@pytest.mark.parametrize( + "identifier", + [ + "../outside", + "..\\outside", + "a" * 256, + "\ud800", + ], +) +def test_build_safe_download_path_rejects_nonportable_identifiers( + tmp_path, identifier +): + with pytest.raises(ValueError, match="Unsafe archive download filename"): + build_safe_download_path( + tmp_path, + identifier, + "1.0.0", + ) + + @pytest.mark.parametrize( "member_name", [ @@ -306,6 +387,170 @@ def test_safe_extract_zip_rejects_too_many_entries(tmp_path): safe_extract_zip(zip_path, tmp_path / "out", max_entries=1) +def _legacy_zip_eocd( + *, + entries: int, + central_directory_size: int, + central_directory_offset: int = 0, + comment_size: int = 0, +) -> bytes: + return struct.pack( + "<4s4H2LH", + b"PK\x05\x06", + 0, + 0, + entries, + entries, + central_directory_size, + central_directory_offset, + comment_size, + ) + + +def test_safe_extract_zip_preflights_declared_entry_count(tmp_path, monkeypatch): + zip_path = tmp_path / "too-many.zip" + zip_path.write_bytes( + _legacy_zip_eocd(entries=513, central_directory_size=0) + ) + monkeypatch.setattr( + zipfile, + "ZipFile", + lambda *_args, **_kwargs: pytest.fail("ZipFile constructor was called"), + ) + + with pytest.raises(ValueError, match="too many entries"): + safe_extract_zip(zip_path, tmp_path / "out") + + +def test_safe_extract_zip_preflights_actual_entry_count_when_eocd_lies( + tmp_path, monkeypatch +): + central_header = b"PK\x01\x02" + b"\x00" * 42 + central_directory = central_header * 513 + zip_path = tmp_path / "lying-count.zip" + zip_path.write_bytes( + central_directory + + _legacy_zip_eocd( + entries=1, + central_directory_size=len(central_directory), + ) + ) + monkeypatch.setattr( + zipfile, + "ZipFile", + lambda *_args, **_kwargs: pytest.fail("ZipFile constructor was called"), + ) + + with pytest.raises(ValueError, match="too many entries"): + safe_extract_zip(zip_path, tmp_path / "out") + + +def test_safe_extract_zip_rejects_truncated_last_eocd_comment( + tmp_path, monkeypatch +): + trailing_eocd = _legacy_zip_eocd( + entries=0, + central_directory_size=0, + comment_size=1, + ) + zip_path = tmp_path / "ambiguous-eocd.zip" + zip_path.write_bytes( + _legacy_zip_eocd( + entries=0, + central_directory_size=0, + comment_size=len(trailing_eocd), + ) + + trailing_eocd + ) + monkeypatch.setattr( + zipfile, + "ZipFile", + lambda *_args, **_kwargs: pytest.fail("ZipFile constructor was called"), + ) + + with pytest.raises(ValueError, match="Invalid ZIP archive"): + safe_extract_zip(zip_path, tmp_path / "out") + + +def test_safe_extract_zip_rejects_zip64_before_zipfile_construction( + tmp_path, monkeypatch +): + zip64_eocd = struct.pack( + "<4sQ2H2L4Q", + b"PK\x06\x06", + 44, + 45, + 45, + 0, + 0, + 0, + 0, + 0, + 0, + ) + zip64_locator = struct.pack( + "<4sLQL", + b"PK\x06\x07", + 0, + 0, + 1, + ) + zip_path = tmp_path / "zip64.zip" + zip_path.write_bytes( + zip64_eocd + + zip64_locator + + _legacy_zip_eocd( + entries=0xFFFF, + central_directory_size=0xFFFFFFFF, + central_directory_offset=0xFFFFFFFF, + ) + ) + with zipfile.ZipFile(zip_path) as zf: + assert zf.namelist() == [] + monkeypatch.setattr( + zipfile, + "ZipFile", + lambda *_args, **_kwargs: pytest.fail("ZipFile constructor was called"), + ) + + with pytest.raises(ValueError, match="ZIP64"): + safe_extract_zip(zip_path, tmp_path / "out") + + +def test_safe_extract_zip_rejects_central_entry_from_another_disk(tmp_path): + zip_path = tmp_path / "multi-disk-entry.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("file.txt", "contents") + + archive = bytearray(zip_path.read_bytes()) + central_header = archive.index(b"PK\x01\x02") + struct.pack_into(" Date: Fri, 24 Jul 2026 18:45:51 +0200 Subject: [PATCH 3/3] harden: address download security review feedback Assisted-by: OpenAI Codex (model: GPT-5, autonomous) --- src/specify_cli/_download_security.py | 20 +- .../bundler/commands_impl/catalog_config.py | 2 + src/specify_cli/bundler/models/catalog.py | 7 + src/specify_cli/bundler/services/adapters.py | 20 +- src/specify_cli/commands/bundle/__init__.py | 91 ++++- src/specify_cli/extensions/_commands.py | 18 +- src/specify_cli/workflows/_commands.py | 96 ++++- src/specify_cli/workflows/catalog.py | 32 +- tests/contract/test_catalog_schema.py | 19 + .../integration/test_bundler_local_install.py | 61 +++ tests/test_download_security.py | 24 +- tests/test_extensions.py | 71 ++++ tests/test_workflows.py | 377 ++++++++++++++++++ tests/unit/test_bundle_download_url.py | 120 ++++++ tests/unit/test_bundler_adapters.py | 37 +- 15 files changed, 958 insertions(+), 37 deletions(-) diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index 0aa0a74c03..e41e6e203e 100644 --- a/src/specify_cli/_download_security.py +++ b/src/specify_cli/_download_security.py @@ -304,7 +304,7 @@ def build_safe_download_path( filename_too_long or posix_path.name != filename or windows_path.name != filename - or any(ord(character) < 32 for character in filename) + or any(unicodedata.category(character) == "Cc" for character in filename) or any( character in _WINDOWS_INVALID_FILENAME_CHARS for character in filename @@ -368,8 +368,12 @@ def read_zip_member_limited( ) -def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: - """Return a normalized ZIP member name or raise on traversal.""" +def normalize_zip_member_name( + name: str, + *, + error_type: type[ErrorT] = ValueError, +) -> str: + """Return a normalized, portable ZIP member name or raise if unsafe.""" if "\x00" in name: _raise(error_type, f"Unsafe path in ZIP archive: {name!r}") @@ -407,7 +411,10 @@ def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str: reserved_stem = part.partition(".")[0].partition(":")[0].rstrip(" ") if ( len(part.encode("utf-8")) > MAX_ZIP_COMPONENT_BYTES - or any(ord(character) < 32 for character in part) + or any( + unicodedata.category(character) == "Cc" + for character in part + ) or any(character in _WINDOWS_INVALID_FILENAME_CHARS for character in part) or part.startswith(" ") or part.endswith((" ", ".")) @@ -616,7 +623,10 @@ def safe_extract_zip( validated_paths: dict[tuple[str, ...], tuple[str, bool]] = {} total_size = 0 for member in members: - normalized_name = _safe_zip_name(member.filename, error_type=error_type) + normalized_name = normalize_zip_member_name( + member.filename, + error_type=error_type, + ) is_dir = member.is_dir() or normalized_name.endswith("/") path_key = portable_zip_path_key(normalized_name) diff --git a/src/specify_cli/bundler/commands_impl/catalog_config.py b/src/specify_cli/bundler/commands_impl/catalog_config.py index 37e45bf990..e9f6d4005a 100644 --- a/src/specify_cli/bundler/commands_impl/catalog_config.py +++ b/src/specify_cli/bundler/commands_impl/catalog_config.py @@ -153,6 +153,8 @@ def add_source( # keeps that ValueError inside the guard instead of leaking a raw # traceback past the CLI's `except BundlerError`. Reuse the value below. hostname = parsed.hostname + # Accessing ``port`` performs urllib's syntax/range validation. + _ = parsed.port except ValueError as exc: raise BundlerError(f"Invalid catalog url: '{url}'.") from exc if not (parsed.scheme or parsed.path): diff --git a/src/specify_cli/bundler/models/catalog.py b/src/specify_cli/bundler/models/catalog.py index 9621b5a334..a5c6499cae 100644 --- a/src/specify_cli/bundler/models/catalog.py +++ b/src/specify_cli/bundler/models/catalog.py @@ -139,6 +139,7 @@ class CatalogEntry: license: str download_url: str requires_speckit_version: str + sha256: str | None = None provides: dict[str, int] = field(default_factory=dict) repository: str | None = None tags: tuple[str, ...] = () @@ -181,6 +182,11 @@ def from_dict(cls, data: Any) -> "CatalogEntry": license=str(data.get("license", "")).strip(), download_url=str(data.get("download_url", "")).strip(), requires_speckit_version=str(requires.get("speckit_version", "")).strip(), + sha256=( + None + if data.get("sha256") is None + else str(data["sha256"]).strip() + ), provides=dict(provides_raw), repository=(str(data["repository"]) if data.get("repository") else None), tags=_parse_tags(data.get("tags"), entry_id), @@ -193,6 +199,7 @@ def with_provenance(self, source: CatalogSource) -> "CatalogEntry": description=self.description, author=self.author, license=self.license, download_url=self.download_url, requires_speckit_version=self.requires_speckit_version, + sha256=self.sha256, provides=self.provides, repository=self.repository, tags=self.tags, verified=self.verified, source_id=source.id, source_policy=source.install_policy, diff --git a/src/specify_cli/bundler/services/adapters.py b/src/specify_cli/bundler/services/adapters.py index 610627b908..403232a7f0 100644 --- a/src/specify_cli/bundler/services/adapters.py +++ b/src/specify_cli/bundler/services/adapters.py @@ -16,6 +16,7 @@ from urllib.request import url2pathname from ..._assets import _locate_core_pack, _repo_root +from ..._download_security import MAX_JSON_CATALOG_BYTES, read_response_limited from .. import BundlerError from ..lib.yamlio import loads_json from ..models.catalog import CatalogSource @@ -76,6 +77,8 @@ def _validate_remote_url(source_id: str, url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname + # Accessing ``port`` performs urllib's syntax/range validation. + _ = parsed.port except ValueError: raise BundlerError( f"Catalog '{source_id}' URL is malformed: {url}" @@ -117,7 +120,15 @@ def make_catalog_fetcher(*, allow_network: bool = True): def fetch(source: CatalogSource) -> dict: url = source.url - parsed = urlparse(url) + try: + parsed = urlparse(url) + # Keep malformed authorities and ports inside the BundlerError + # contract even when a config file was edited by hand. + _ = parsed.port + except ValueError: + raise BundlerError( + f"Catalog {source.id!r} URL is malformed: {url!r}" + ) from None scheme = parsed.scheme.lower() if scheme == "builtin": @@ -180,7 +191,12 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: ) as response: final_url = response.geturl() _validate_remote_url(source_id, final_url) - raw = response.read().decode("utf-8") + raw = read_response_limited( + response, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=BundlerError, + label=f"bundle catalog '{source_id}'", + ).decode("utf-8") except BundlerError: raise except Exception as exc: # noqa: BLE001 diff --git a/src/specify_cli/commands/bundle/__init__.py b/src/specify_cli/commands/bundle/__init__.py index d8fb22e511..04ac0b0289 100644 --- a/src/specify_cli/commands/bundle/__init__.py +++ b/src/specify_cli/commands/bundle/__init__.py @@ -14,6 +14,7 @@ import typer from ..._console import console, err_console +from ..._download_security import MAX_DOWNLOAD_BYTES, read_response_limited from ...bundler import BundlerError from ...bundler.lib.project import ( active_integration, @@ -337,6 +338,10 @@ def bundle_install( local_manifest = _local_manifest_source(bundle_id) if local_manifest is not None: manifest = local_manifest + _validate_manifest_structure( + manifest, + source=f"Local bundle source {bundle_id!r}", + ) else: stack = _build_stack(project_root or Path.cwd(), offline=offline) resolved = stack.resolve(bundle_id) @@ -350,6 +355,16 @@ def bundle_install( if project_root is None: init_integration = _resolve_init_integration(integration, manifest) + # Resolve all hard compatibility gates before ``specify init``. + # Otherwise an incompatible but structurally valid bundle would + # initialize a project and only then fail its version/integration + # checks, leaving state behind after a failed install. + resolve_install_plan( + manifest, + speckit_version=_speckit_version(), + active_integration=init_integration, + integration_explicit=True, + ) console.print( f"[cyan]No Spec Kit project here; initializing with integration " f"'{init_integration}'…[/cyan]" @@ -711,17 +726,24 @@ def _local_manifest_source(arg: str): if candidate.suffix == ".zip": import io - import zipfile import yaml as _yaml - with zipfile.ZipFile(candidate) as archive: + from ..._download_security import open_zip_bounded, read_zip_member_limited + + with open_zip_bounded(candidate, error_type=BundlerError) as archive: try: - raw = archive.read("bundle.yml") + archive.getinfo("bundle.yml") except KeyError as exc: raise BundlerError( f"Artifact '{candidate}' does not contain a bundle.yml." ) from exc + raw = read_zip_member_limited( + archive, + "bundle.yml", + error_type=BundlerError, + label="bundle manifest", + ) data = _yaml.safe_load(io.BytesIO(raw)) return BundleManifest.from_dict(data) @@ -805,7 +827,13 @@ def _download_manifest(resolved, *, offline: bool): f"Network access disabled; cannot download bundle '{resolved.entry.id}' " f"from {url}." ) - return _download_remote_manifest(resolved.entry.id, url) + manifest = _download_remote_manifest( + resolved.entry.id, + url, + expected_sha256=getattr(resolved.entry, "sha256", None), + ) + _validate_catalog_manifest(resolved.entry, manifest) + return manifest def _require_https(label: str, url: str) -> None: @@ -817,6 +845,8 @@ def _require_https(label: str, url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname + # Accessing ``port`` performs urllib's syntax/range validation. + _ = parsed.port except ValueError: raise BundlerError( f"Refusing to download {label}: URL is malformed: {url}" @@ -830,7 +860,12 @@ def _require_https(label: str, url: str) -> None: raise BundlerError(f"Refusing to download {label} from URL with no host: {url}") -def _download_remote_manifest(entry_id: str, url: str): +def _download_remote_manifest( + entry_id: str, + url: str, + *, + expected_sha256: str | None = None, +): """Fetch a remote bundle artifact over HTTPS and extract its manifest.""" import io import tempfile @@ -842,6 +877,7 @@ def _download_remote_manifest(entry_id: str, url: str): from ...authentication.http import github_provider_hosts, open_url from ..._github_http import resolve_github_release_asset_api_url from ...bundler.models.manifest import BundleManifest + from ...shared_infra import verify_archive_sha256 def _validate_redirect(old_url: str, new_url: str) -> None: _require_https(f"bundle '{entry_id}'", new_url) @@ -879,7 +915,18 @@ def _validate_redirect(old_url: str, new_url: str) -> None: extra_headers=extra_headers, ) as resp: _require_https(f"bundle '{entry_id}'", resp.geturl()) - raw = resp.read() + raw = read_response_limited( + resp, + max_bytes=MAX_DOWNLOAD_BYTES, + error_type=BundlerError, + label=f"bundle '{entry_id}' download", + ) + verify_archive_sha256( + raw, + expected_sha256, + entry_id, + BundlerError, + ) except BundlerError: raise except Exception as exc: # noqa: BLE001 @@ -940,6 +987,38 @@ def _validate_redirect(old_url: str, new_url: str) -> None: ) from exc +def _validate_manifest_structure(manifest, *, source: str) -> None: + """Reject a malformed manifest before any project mutation can occur.""" + from ...bundler.services.validator import validate_manifest + + report = validate_manifest(manifest) + if report.ok: + return + raise BundlerError( + f"{source} contains an invalid bundle manifest:\n - " + + "\n - ".join(report.errors) + ) + + +def _validate_catalog_manifest(entry, manifest) -> None: + """Bind a downloaded manifest to the catalog identity that selected it.""" + if manifest.bundle.id != entry.id: + raise BundlerError( + f"Downloaded bundle id mismatch: catalog entry {entry.id!r} points to " + f"a manifest for {manifest.bundle.id!r}." + ) + if manifest.bundle.version != entry.version: + raise BundlerError( + f"Downloaded bundle version mismatch for {entry.id!r}: catalog declares " + f"{entry.version!r}, but the manifest declares " + f"{manifest.bundle.version!r}." + ) + _validate_manifest_structure( + manifest, + source=f"Downloaded bundle {entry.id!r}", + ) + + def register(app: typer.Typer) -> None: """Attach the bundle command group to the root Typer app.""" app.add_typer(bundle_app, name="bundle") diff --git a/src/specify_cli/extensions/_commands.py b/src/specify_cli/extensions/_commands.py index e3410cca3b..2793424706 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -25,6 +25,7 @@ from .._assets import get_speckit_version from .._download_security import ( is_https_or_localhost_http, + normalize_zip_member_name, open_zip_bounded, portable_zip_path_key, read_response_limited, @@ -1160,6 +1161,8 @@ def extension_update( backup_installed = UNSET # Original installed list from extensions.yml backup_hooks = None # None means backup step 4 not yet reached; {} or {...} means backup was captured backed_up_command_files = {} + # Validation failures must not rewrite an untouched installation. + installation_modified = False try: # 1. Backup registry entry (always, even if extension dir doesn't exist) @@ -1252,7 +1255,7 @@ def extension_update( # later overwrites it with a backslash alias. manifest_candidates = [] for name in namelist: - normalized_name = name.replace("\\", "/") + normalized_name = normalize_zip_member_name(name) parts = normalized_name.split("/") path_key = portable_zip_path_key(normalized_name) if ( @@ -1322,6 +1325,7 @@ def extension_update( ) # 7. Remove old extension (handles command file cleanup and registry removal) + installation_modified = True manager.remove(extension_id, keep_config=True) # 8. Install new version @@ -1388,6 +1392,18 @@ def extension_update( console.print(f" [red]✗[/red] Failed: {_escape_markup(str(e))}") failed_updates.append((ext_name, str(e))) + if not installation_modified: + if backup_base.exists(): + try: + shutil.rmtree(backup_base) + except OSError as cleanup_error: + console.print( + " [yellow]Warning:[/yellow] Could not remove " + "untouched-update backup: " + f"{_escape_markup(str(cleanup_error))}" + ) + continue + # Rollback on failure console.print(f" [yellow]↩[/yellow] Rolling back {safe_ext_name}...") diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index f088519934..c6aec659a0 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -12,7 +12,7 @@ import os import re import sys -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import Any import typer @@ -401,6 +401,12 @@ def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None: # a ceiling any legitimate workflow definition should ever approach. _MAX_WORKFLOW_YAML_BYTES = 5 * 1024 * 1024 # 5 MiB _DOWNLOAD_CHUNK_SIZE = 65536 +# Custom step packages contain executable Python, metadata, and optional helper +# files downloaded one-by-one rather than as an archive. Mirror the archive +# ceilings so a catalog cannot turn individually valid files into an unbounded +# aggregate download. +_MAX_STEP_PACKAGE_FILES = 512 +_MAX_STEP_PACKAGE_BYTES = 50 * 1024 * 1024 # 50 MiB def _read_response_within_limit(response, max_bytes: int | None = None) -> bytes: @@ -2636,14 +2642,39 @@ def workflow_step_add( ) raise typer.Exit(1) - step_yml_url = info.get("step_yml_url") or info.get("url") - if not step_yml_url: + declared_step_yml_url = info.get("step_yml_url") + if declared_step_yml_url is not None and not isinstance( + declared_step_yml_url, str + ): + console.print( + f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " + "step.yml URL; expected a non-empty string" + ) + raise typer.Exit(1) + step_yml_url = declared_step_yml_url or info.get("url") + if step_yml_url is None or ( + isinstance(step_yml_url, str) and not step_yml_url.strip() + ): console.print(f"[red]Error:[/red] Catalog entry for '{step_id}' has no URL") raise typer.Exit(1) + if not isinstance(step_yml_url, str): + console.print( + f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " + "step.yml URL; expected a non-empty string" + ) + raise typer.Exit(1) # Derive __init__.py URL: replace trailing step.yml with __init__.py # or use explicit init_url if provided. init_url = info.get("init_url") + if init_url is not None and ( + not isinstance(init_url, str) or not init_url.strip() + ): + console.print( + f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " + "__init__.py URL; expected a non-empty string" + ) + raise typer.Exit(1) if not init_url: if step_yml_url.endswith("step.yml"): init_url = step_yml_url[: -len("step.yml")] + "__init__.py" @@ -2654,6 +2685,41 @@ def workflow_step_add( ) raise typer.Exit(1) + # Preflight the declared file count before creating a staging directory or + # issuing any request. The two required files are always part of the package; + # duplicate declarations for them in extra_files are ignored below and do + # not count twice. + extra_files = info.get("extra_files") + if extra_files is not None and not isinstance(extra_files, dict): + console.print( + "[yellow]Warning:[/yellow] Catalog entry 'extra_files' is not a mapping; " + "additional package files will not be downloaded." + ) + extra_files = {} + + def _is_required_package_file(rel_path: object) -> bool: + """Match portable path/case aliases of the two required package files.""" + if not isinstance(rel_path, str): + return False + parts = PurePosixPath(rel_path.replace("\\", "/")).parts + return len(parts) == 1 and parts[0].casefold() in { + "step.yml", + "__init__.py", + } + + declared_extra_count = sum( + 1 + for rel_path in (extra_files or {}) + if not _is_required_package_file(rel_path) + ) + package_file_count = 2 + declared_extra_count + if package_file_count > _MAX_STEP_PACKAGE_FILES: + console.print( + f"[red]Error:[/red] Step package declares {package_file_count} files, " + f"exceeding the {_MAX_STEP_PACKAGE_FILES}-file limit" + ) + raise typer.Exit(1) + from specify_cli.authentication.http import open_url as _open_url def _safe_fetch(url: str) -> bytes: @@ -2710,6 +2776,14 @@ def _safe_fetch(url: str) -> bytes: console.print(f"[red]Error:[/red] Failed to download step files: {exc}") raise typer.Exit(1) + package_bytes = len(step_yml_content) + len(init_py_content) + if package_bytes > _MAX_STEP_PACKAGE_BYTES: + console.print( + f"[red]Error:[/red] Step package exceeds the " + f"{_MAX_STEP_PACKAGE_BYTES}-byte total size limit" + ) + raise typer.Exit(1) + # Validate step.yml try: import yaml as _yaml @@ -2754,13 +2828,6 @@ def _safe_fetch(url: str) -> bytes: # relative-path → URL. step.yml and __init__.py are ignored here (already # written). Paths are validated to stay within the step package directory to # prevent path-traversal attacks. - extra_files = info.get("extra_files") - if extra_files is not None and not isinstance(extra_files, dict): - console.print( - "[yellow]Warning:[/yellow] Catalog entry 'extra_files' is not a mapping; " - "additional package files will not be downloaded." - ) - extra_files = {} for rel_path, file_url in (extra_files or {}).items(): if not isinstance(rel_path, str) or not rel_path.strip(): console.print( @@ -2768,7 +2835,7 @@ def _safe_fetch(url: str) -> bytes: "empty or non-string path key" ) raise typer.Exit(1) - if rel_path in ("step.yml", "__init__.py"): + if _is_required_package_file(rel_path): continue # already written above # Reject dot-path segments ('', '.', '..') that would refer to the # package directory itself (IsADirectoryError) or escape it. @@ -2804,6 +2871,13 @@ def _safe_fetch(url: str) -> bytes: f"[red]Error:[/red] Failed to download extra file '{rel_path}': {exc}" ) raise typer.Exit(1) + package_bytes += len(file_content) + if package_bytes > _MAX_STEP_PACKAGE_BYTES: + console.print( + f"[red]Error:[/red] Step package exceeds the " + f"{_MAX_STEP_PACKAGE_BYTES}-byte total size limit" + ) + raise typer.Exit(1) try: dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(file_content) diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 63ddc76395..46b4e81706 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -22,6 +22,8 @@ import yaml +from .._download_security import MAX_JSON_CATALOG_BYTES, read_response_limited + # --------------------------------------------------------------------------- # Errors @@ -308,7 +310,8 @@ def _validate_catalog_url(self, url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname - except ValueError: + _ = parsed.port + except (TypeError, ValueError): raise WorkflowValidationError( f"Catalog URL is malformed: {url}" ) from None @@ -505,7 +508,8 @@ def _validate_catalog_url(url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname - except ValueError: + _ = parsed.port + except (TypeError, ValueError): raise WorkflowCatalogError( f"Refusing to fetch catalog from malformed URL: {url}" ) from None @@ -538,7 +542,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: entry.url, timeout=30, redirect_validator=_validate_redirect ) as resp: _validate_catalog_url(resp.geturl()) - data = json.loads(resp.read().decode("utf-8")) + data = json.loads( + read_response_limited( + resp, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=WorkflowCatalogError, + label="workflow catalog", + ).decode("utf-8") + ) except Exception as exc: # Fall back to cache if available if cache_file.exists(): @@ -982,7 +993,8 @@ def _validate_catalog_url(self, url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname - except ValueError: + _ = parsed.port + except (TypeError, ValueError): raise StepValidationError( f"Catalog URL is malformed: {url}" ) from None @@ -1178,7 +1190,8 @@ def _validate_url(url: str) -> None: try: parsed = urlparse(url) hostname = parsed.hostname - except ValueError: + _ = parsed.port + except (TypeError, ValueError): raise StepCatalogError( f"Refusing to fetch catalog from malformed URL: {url}" ) from None @@ -1211,7 +1224,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: entry.url, timeout=30, redirect_validator=_validate_redirect ) as resp: _validate_url(resp.geturl()) - data = json.loads(resp.read().decode("utf-8")) + data = json.loads( + read_response_limited( + resp, + max_bytes=MAX_JSON_CATALOG_BYTES, + error_type=StepCatalogError, + label="step catalog", + ).decode("utf-8") + ) except Exception as exc: if cache_safe and cache_file.exists(): try: diff --git a/tests/contract/test_catalog_schema.py b/tests/contract/test_catalog_schema.py index 97600d21ef..4a2a9efe75 100644 --- a/tests/contract/test_catalog_schema.py +++ b/tests/contract/test_catalog_schema.py @@ -209,6 +209,25 @@ def test_catalog_entry_rejects_non_boolean_verified(): CatalogEntry.from_dict(data) +def test_catalog_entry_preserves_sha256_through_provenance(): + digest = "a" * 64 + payload = catalog_payload( + {"demo": catalog_entry_dict("demo", sha256=f"sha256:{digest}")} + ) + + entry = load_catalog_payload(payload)["demo"] + source = CatalogSource( + id="team", + url="https://example.com/catalog.json", + priority=10, + install_policy=InstallPolicy.INSTALL_ALLOWED, + scope=Scope.PROJECT, + ) + + assert entry.sha256 == f"sha256:{digest}" + assert entry.with_provenance(source).sha256 == f"sha256:{digest}" + + def test_load_payload_rejects_id_key_mismatch(): # The enclosing key is authoritative; an entry whose own id disagrees with # the key must be rejected so a catalog can't list a spoofed/unresolvable id. diff --git a/tests/integration/test_bundler_local_install.py b/tests/integration/test_bundler_local_install.py index 32972ac684..164de57006 100644 --- a/tests/integration/test_bundler_local_install.py +++ b/tests/integration/test_bundler_local_install.py @@ -7,7 +7,9 @@ from __future__ import annotations import os +import zipfile from pathlib import Path +from unittest.mock import patch import pytest import yaml @@ -171,3 +173,62 @@ def test_download_manifest_rejects_non_https_url_even_offline(tmp_path: Path): ) with pytest.raises(BundlerError, match="HTTPS"): _download_manifest(resolved, offline=True) + + +def test_local_zip_uses_bounded_archive_open(tmp_path: Path): + artifact = tmp_path / "too-many-entries.zip" + with zipfile.ZipFile(artifact, "w") as archive: + archive.writestr("bundle.yml", yaml.safe_dump(valid_manifest_dict())) + for index in range(512): + archive.writestr(f"assets/{index}.txt", "") + + with pytest.raises(BundlerError, match="too many entries"): + _local_manifest_source(str(artifact)) + + +def test_invalid_local_manifest_is_rejected_before_project_init( + tmp_path: Path, + monkeypatch, +): + bundle_dir = tmp_path / "invalid-bundle" + data = valid_manifest_dict() + data["bundle"]["author"] = "" + write_manifest(bundle_dir, data) + empty_cwd = tmp_path / "empty" + empty_cwd.mkdir() + monkeypatch.chdir(empty_cwd) + + runner = CliRunner() + with patch("specify_cli.commands.bundle._run_init") as run_init: + result = runner.invoke( + app, + ["bundle", "install", str(bundle_dir), "--offline"], + ) + + assert result.exit_code == 1 + assert "Missing required field: bundle.author" in result.output + run_init.assert_not_called() + + +def test_incompatible_local_manifest_is_rejected_before_project_init( + tmp_path: Path, + monkeypatch, +): + bundle_dir = tmp_path / "incompatible-bundle" + data = valid_manifest_dict() + data["requires"]["speckit_version"] = ">=999.0.0" + write_manifest(bundle_dir, data) + empty_cwd = tmp_path / "empty" + empty_cwd.mkdir() + monkeypatch.chdir(empty_cwd) + + runner = CliRunner() + with patch("specify_cli.commands.bundle._run_init") as run_init: + result = runner.invoke( + app, + ["bundle", "install", str(bundle_dir), "--offline"], + ) + + assert result.exit_code == 1 + assert "requires Spec Kit >=999.0.0" in result.output + run_init.assert_not_called() diff --git a/tests/test_download_security.py b/tests/test_download_security.py index f93f3680ec..583b85d99b 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -297,6 +297,8 @@ def test_read_response_limited_escapes_control_characters_in_label(): "../outside", "..\\outside", "a" * 256, + "delete\x7f", + "csi\x9b[2J", "\ud800", ], ) @@ -719,6 +721,8 @@ def test_safe_extract_zip_rejects_conflicting_paths_before_writing( "CONOUT$.log", "nested/name?.txt", "nested/control\u0001.txt", + "nested/delete\u007f.txt", + "nested/csi\u009b[2J.txt", ], ) def test_safe_extract_zip_rejects_nonportable_member_names(tmp_path, member_name): @@ -752,16 +756,28 @@ def test_safe_extract_zip_rejects_excessively_long_paths(tmp_path, member_name): assert not out_dir.exists() or not any(out_dir.rglob("*")) -def test_safe_extract_zip_escapes_control_characters_in_errors(tmp_path): +@pytest.mark.parametrize( + ("control_character", "escaped_character"), + [ + ("\x1b", "\\x1b"), + ("\x7f", "\\x7f"), + ("\x9b", "\\x9b"), + ], +) +def test_safe_extract_zip_escapes_unicode_control_characters_in_errors( + tmp_path, + control_character, + escaped_character, +): zip_path = tmp_path / "terminal-control.zip" with zipfile.ZipFile(zip_path, "w") as zf: - zf.writestr("bad\x1b[2J.txt", "contents") + zf.writestr(f"bad{control_character}[2J.txt", "contents") with pytest.raises(ValueError) as exc_info: safe_extract_zip(zip_path, tmp_path / "out") - assert "\x1b" not in str(exc_info.value) - assert "\\x1b" in str(exc_info.value) + assert control_character not in str(exc_info.value) + assert escaped_character in str(exc_info.value) def test_safe_extract_zip_accepts_single_decomposed_unicode_name(tmp_path): diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 36118d5507..acbdf110e3 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -7358,6 +7358,77 @@ def _create_catalog_zip( if extra_manifest_path is not None: zf.writestr(extra_manifest_path, manifest_text) + @pytest.mark.parametrize( + "manifest_path", + [ + "../extension.yml", + "/extension.yml", + "./extension.yml", + "C:/extension.yml", + ], + ) + def test_update_rejects_unsafe_manifest_path_before_removal( + self, tmp_path, manifest_path + ): + """Unsafe manifest paths fail before the installed extension is removed.""" + from typer.testing import CliRunner + from unittest.mock import patch + from specify_cli import app + + project_dir = tmp_path / "project" + project_dir.mkdir() + (project_dir / ".specify").mkdir() + (project_dir / ".claude" / "skills").mkdir(parents=True) + + manager = ExtensionManager(project_dir) + v1_dir = self._create_extension_source(tmp_path, "1.0.0") + manager.install_from_directory(v1_dir, "0.1.0") + installed_extension_dir = manager.extensions_dir / "test-ext" + removed_paths = [] + real_rmtree = shutil.rmtree + + def track_rmtree(path, *args, **kwargs): + removed_paths.append(Path(path).resolve()) + return real_rmtree(path, *args, **kwargs) + + zip_path = tmp_path / "unsafe-manifest.zip" + self._create_catalog_zip( + zip_path, + "2.0.0", + manifest_path=manifest_path, + ) + + runner = CliRunner() + with patch.object(Path, "cwd", return_value=project_dir), \ + patch.object(ExtensionCatalog, "get_extension_info", return_value={ + "id": "test-ext", + "name": "Test Extension", + "version": "2.0.0", + "_install_allowed": True, + }), \ + patch.object( + ExtensionCatalog, + "download_extension", + return_value=zip_path, + ), \ + patch.object(shutil, "rmtree", side_effect=track_rmtree), \ + patch.object(ExtensionManager, "remove") as remove, \ + patch.object(ExtensionManager, "install_from_zip") as install: + result = runner.invoke( + app, + ["extension", "update", "test-ext"], + input="y\n", + catch_exceptions=True, + ) + + assert result.exit_code == 1 + assert "Unsafe path in ZIP archive" in result.output + remove.assert_not_called() + install.assert_not_called() + assert installed_extension_dir.resolve() not in removed_paths + assert not (manager.extensions_dir / ".backup" / "test-ext-update").exists() + assert ExtensionManager(project_dir).registry.get("test-ext")["version"] == "1.0.0" + @pytest.mark.parametrize( ("first_path", "second_path"), [ diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 1c29ab56e6..921c7a73b9 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -6356,6 +6356,7 @@ def test_validate_url_localhost_http_allowed(self, project_dir): [ "https://[::1", # unterminated IPv6 bracket "https://[not-an-ip]/x", # bracketed non-IP host + "https://example.com:notaport/catalog.json", ], ) def test_validate_url_malformed_raises_validation_error(self, project_dir, url): @@ -6465,6 +6466,62 @@ def fake_open(url, timeout=30, redirect_validator=None): catalog._fetch_single_catalog(entry, force_refresh=True) assert captured["rv"] is not None + def test_fetch_rejects_oversized_catalog_response( + self, project_dir, monkeypatch + ): + from specify_cli.authentication import http as auth_http + from specify_cli.workflows import catalog as catalog_module + from specify_cli.workflows.catalog import ( + WorkflowCatalog, + WorkflowCatalogEntry, + WorkflowCatalogError, + ) + + monkeypatch.setattr(catalog_module, "MAX_JSON_CATALOG_BYTES", 32) + requested_sizes: list[int] = [] + + class _FakeResponse: + def __init__(self): + self.body = b"x" * 64 + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def geturl(self): + return "https://example.com/catalog.json" + + def read(self, size=-1): + requested_sizes.append(size) + assert size >= 0 + chunk_size = min(size, 7) + chunk = self.body[self.offset : self.offset + chunk_size] + self.offset += len(chunk) + return chunk + + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(), + ) + + catalog = WorkflowCatalog(project_dir) + entry = WorkflowCatalogEntry( + url="https://example.com/catalog.json", + name="test", + priority=1, + install_allowed=True, + ) + + with pytest.raises(WorkflowCatalogError, match="exceeds maximum size"): + catalog._fetch_single_catalog(entry, force_refresh=True) + + assert requested_sizes + assert not catalog.cache_dir.exists() + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import WorkflowCatalog @@ -6960,6 +7017,7 @@ def test_validate_url_localhost_http_allowed(self, project_dir): [ "https://[::1", # unterminated IPv6 bracket "https://[not-an-ip]/x", # bracketed non-IP host + "https://example.com:notaport/steps.json", ], ) def test_validate_url_malformed_raises_validation_error(self, project_dir, url): @@ -7061,6 +7119,62 @@ def fake_open(url, timeout=30, redirect_validator=None): catalog._fetch_single_catalog(entry, force_refresh=True) assert captured["rv"] is not None + def test_fetch_rejects_oversized_catalog_response( + self, project_dir, monkeypatch + ): + from specify_cli.authentication import http as auth_http + from specify_cli.workflows import catalog as catalog_module + from specify_cli.workflows.catalog import ( + StepCatalog, + StepCatalogEntry, + StepCatalogError, + ) + + monkeypatch.setattr(catalog_module, "MAX_JSON_CATALOG_BYTES", 32) + requested_sizes: list[int] = [] + + class _FakeResponse: + def __init__(self): + self.body = b"x" * 64 + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def geturl(self): + return "https://example.com/steps.json" + + def read(self, size=-1): + requested_sizes.append(size) + assert size >= 0 + chunk_size = min(size, 7) + chunk = self.body[self.offset : self.offset + chunk_size] + self.offset += len(chunk) + return chunk + + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(), + ) + + catalog = StepCatalog(project_dir) + entry = StepCatalogEntry( + url="https://example.com/steps.json", + name="test", + priority=1, + install_allowed=True, + ) + + with pytest.raises(StepCatalogError, match="exceeds maximum size"): + catalog._fetch_single_catalog(entry, force_refresh=True) + + assert requested_sizes + assert not catalog.cache_dir.exists() + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import StepCatalog @@ -8267,6 +8381,269 @@ def read(self, size=-1): project_dir / ".specify" / "workflows" / "steps" / "my-step" ).exists() + @pytest.mark.parametrize( + ("catalog_fields", "expected"), + [ + ({"url": 123}, "malformed step.yml URL"), + ( + { + "step_yml_url": [], + "url": "https://example.com/step.yml", + }, + "malformed step.yml URL", + ), + ( + { + "url": "https://example.com/step.yml", + "init_url": 123, + }, + "malformed __init__.py URL", + ), + ], + ) + def test_add_rejects_non_string_required_urls_before_network( + self, project_dir, monkeypatch, catalog_fields, expected + ): + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + from specify_cli.workflows.catalog import StepCatalog + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + StepCatalog, + "get_step_info", + lambda self, step_id: { + "id": step_id, + "name": "Test Step", + "_install_allowed": True, + **catalog_fields, + }, + ) + monkeypatch.setattr( + auth_http, + "open_url", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("download should not start") + ), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step"] + ) + + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert expected in result.output + assert not ( + project_dir / ".specify" / "workflows" / "steps" / "my-step" + ).exists() + + @pytest.mark.parametrize( + ("alias", "protected_name"), + [ + ("./step.yml", "step.yml"), + ("step.yml/", "step.yml"), + ("STEP.YML", "step.yml"), + (".\\step.yml", "step.yml"), + ("./__init__.py", "__init__.py"), + ("__init__.py/", "__init__.py"), + ("__INIT__.PY", "__init__.py"), + (".\\__init__.py", "__init__.py"), + ], + ) + def test_add_does_not_overwrite_required_files_through_path_aliases( + self, project_dir, monkeypatch, alias, protected_name + ): + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + from specify_cli.workflows.catalog import StepCatalog + + monkeypatch.chdir(project_dir) + alias_url = "https://example.com/overwrite" + monkeypatch.setattr( + StepCatalog, + "get_step_info", + lambda self, step_id: { + "id": step_id, + "name": "Test Step", + "url": "https://example.com/step.yml", + "init_url": "https://example.com/__init__.py", + "_install_allowed": True, + "extra_files": {alias: alias_url}, + }, + ) + bodies = { + "https://example.com/step.yml": b"step:\n type_key: my-step\n", + "https://example.com/__init__.py": b"# trusted init\n", + } + requested_urls: list[str] = [] + + class _FakeResponse: + def __init__(self, url): + self.url = url + self.body = bodies[url] + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def geturl(self): + return self.url + + def read(self, size=-1): + if size < 0: + size = len(self.body) - self.offset + chunk = self.body[self.offset : self.offset + size] + self.offset += len(chunk) + return chunk + + def fake_open_url(url, timeout=30, redirect_validator=None): + requested_urls.append(url) + return _FakeResponse(url) + + monkeypatch.setattr(auth_http, "open_url", fake_open_url) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step"] + ) + + assert result.exit_code == 0, result.output + assert alias_url not in requested_urls + installed_dir = ( + project_dir / ".specify" / "workflows" / "steps" / "my-step" + ) + assert (installed_dir / protected_name).read_bytes() == bodies[ + f"https://example.com/{protected_name}" + ] + + def test_add_rejects_too_many_package_files_before_network( + self, project_dir, monkeypatch + ): + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + from specify_cli.workflows import _commands as workflow_commands + from specify_cli.workflows.catalog import StepCatalog + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(workflow_commands, "_MAX_STEP_PACKAGE_FILES", 3) + monkeypatch.setattr( + StepCatalog, + "get_step_info", + lambda self, step_id: { + "id": step_id, + "name": "Test Step", + "url": "https://example.com/step.yml", + "init_url": "https://example.com/__init__.py", + "_install_allowed": True, + "extra_files": { + "one.py": "https://example.com/one.py", + "two.py": "https://example.com/two.py", + }, + }, + ) + monkeypatch.setattr( + auth_http, + "open_url", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("download should not start") + ), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step"] + ) + + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert "exceeding the 3-file limit" in result.output + steps_dir = project_dir / ".specify" / "workflows" / "steps" + assert not (steps_dir / "my-step").exists() + assert list(steps_dir.glob("speckit_step_tmp_*")) == [] + + def test_add_rejects_package_over_cumulative_size_and_cleans_staging( + self, project_dir, monkeypatch + ): + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + from specify_cli.workflows import _commands as workflow_commands + from specify_cli.workflows.catalog import StepCatalog + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(workflow_commands, "_MAX_STEP_PACKAGE_BYTES", 40) + monkeypatch.setattr( + StepCatalog, + "get_step_info", + lambda self, step_id: { + "id": step_id, + "name": "Test Step", + "url": "https://example.com/step.yml", + "init_url": "https://example.com/__init__.py", + "_install_allowed": True, + "extra_files": { + "helper.py": "https://example.com/helper.py", + }, + }, + ) + + bodies = { + "https://example.com/step.yml": b"step:\n type_key: my-step\n", + "https://example.com/__init__.py": b"# init\n", + "https://example.com/helper.py": b"0123456789", + } + + class _FakeResponse: + def __init__(self, url): + self.url = url + self.body = bodies[url] + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def getheader(self, name): + return None + + def geturl(self): + return self.url + + def read(self, size=-1): + if size < 0: + size = len(self.body) - self.offset + chunk = self.body[self.offset : self.offset + size] + self.offset += len(chunk) + return chunk + + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(url), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step"] + ) + + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert "40-byte total size limit" in result.output + steps_dir = project_dir / ".specify" / "workflows" / "steps" + assert not (steps_dir / "my-step").exists() + assert list(steps_dir.glob("speckit_step_tmp_*")) == [] + def test_add_rejects_non_string_extra_files_key(self, project_dir, monkeypatch): from typer.testing import CliRunner from specify_cli import app diff --git a/tests/unit/test_bundle_download_url.py b/tests/unit/test_bundle_download_url.py index 6a53b9368b..6a29423b9d 100644 --- a/tests/unit/test_bundle_download_url.py +++ b/tests/unit/test_bundle_download_url.py @@ -1,19 +1,59 @@ """Unit tests for malformed download-URL handling in bundle manifest resolution.""" from __future__ import annotations +import hashlib +import io from types import SimpleNamespace import pytest +import yaml from specify_cli.bundler import BundlerError +from specify_cli.bundler.models.catalog import CatalogEntry +from specify_cli.commands import bundle as bundle_commands from specify_cli.commands.bundle import _download_manifest, _require_https +from tests.bundler_helpers import catalog_entry_dict, valid_manifest_dict _MALFORMED_URLS = [ "https://[::1", # unclosed IPv6 bracket "https://[not-an-ip]/bundle.yml", + "https://example.com:notaport/bundle.yml", + "https://example.com:70000/bundle.yml", ] +class _Response(io.BytesIO): + def __init__(self, body: bytes, url: str) -> None: + super().__init__(body) + self._url = url + + def geturl(self) -> str: + return self._url + + +def _resolved_entry(**overrides) -> SimpleNamespace: + entry = CatalogEntry.from_dict( + catalog_entry_dict( + "demo-bundle", + download_url="https://example.com/demo-bundle.yml", + **overrides, + ) + ) + return SimpleNamespace(entry=entry) + + +def _patch_download(monkeypatch, body: bytes) -> None: + def fake_open_url( + url, + timeout=10, + extra_headers=None, + redirect_validator=None, + ): + return _Response(body, url) + + monkeypatch.setattr("specify_cli.authentication.http.open_url", fake_open_url) + + @pytest.mark.parametrize("url", _MALFORMED_URLS) def test_download_manifest_rejects_malformed_url_cleanly(url): """A malformed download_url must raise BundlerError, not a raw ValueError. @@ -40,3 +80,83 @@ def test_require_https_rejects_malformed_url_cleanly(url): """ with pytest.raises(BundlerError): _require_https("bundle 'x'", url) + + +def test_download_manifest_bounds_remote_artifact(monkeypatch): + body = yaml.safe_dump(valid_manifest_dict()).encode() + _patch_download(monkeypatch, body) + monkeypatch.setattr(bundle_commands, "MAX_DOWNLOAD_BYTES", len(body) - 1) + + with pytest.raises(BundlerError, match="exceeds maximum size"): + _download_manifest(_resolved_entry(), offline=False) + + +def test_download_manifest_accepts_matching_sha256(monkeypatch): + body = yaml.safe_dump(valid_manifest_dict()).encode() + digest = hashlib.sha256(body).hexdigest() + _patch_download(monkeypatch, body) + + manifest = _download_manifest( + _resolved_entry(sha256=f"sha256:{digest}"), + offline=False, + ) + + assert manifest.bundle.id == "demo-bundle" + + +def test_download_manifest_accepts_legacy_entry_without_sha256(monkeypatch): + body = yaml.safe_dump(valid_manifest_dict()).encode() + _patch_download(monkeypatch, body) + resolved = SimpleNamespace( + entry=SimpleNamespace( + id="demo-bundle", + version="1.2.0", + download_url="https://example.com/demo-bundle.yml", + ) + ) + + manifest = _download_manifest(resolved, offline=False) + + assert manifest.bundle.version == "1.2.0" + + +@pytest.mark.parametrize("declared", ["0" * 64, "not-a-sha256"]) +def test_download_manifest_rejects_bad_sha256(monkeypatch, declared): + body = yaml.safe_dump(valid_manifest_dict()).encode() + _patch_download(monkeypatch, body) + + with pytest.raises(BundlerError, match="sha256|Integrity check"): + _download_manifest( + _resolved_entry(sha256=declared), + offline=False, + ) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("id", "other-bundle", "id mismatch"), + ("version", "9.9.9", "version mismatch"), + ], +) +def test_download_manifest_rejects_catalog_identity_mismatch( + monkeypatch, + field, + value, + message, +): + data = valid_manifest_dict() + data["bundle"][field] = value + _patch_download(monkeypatch, yaml.safe_dump(data).encode()) + + with pytest.raises(BundlerError, match=message): + _download_manifest(_resolved_entry(), offline=False) + + +def test_download_manifest_rejects_invalid_structure(monkeypatch): + data = valid_manifest_dict() + data["bundle"]["author"] = "" + _patch_download(monkeypatch, yaml.safe_dump(data).encode()) + + with pytest.raises(BundlerError, match="invalid bundle manifest"): + _download_manifest(_resolved_entry(), offline=False) diff --git a/tests/unit/test_bundler_adapters.py b/tests/unit/test_bundler_adapters.py index 0212e850db..5ce9e10a12 100644 --- a/tests/unit/test_bundler_adapters.py +++ b/tests/unit/test_bundler_adapters.py @@ -20,6 +20,7 @@ def _source(url: str) -> CatalogSource: class _FakeResponse: def __init__(self, body: bytes, final_url: str) -> None: self._body = body + self._offset = 0 self._final_url = final_url def __enter__(self) -> "_FakeResponse": @@ -31,8 +32,12 @@ def __exit__(self, *exc) -> bool: def geturl(self) -> str: return self._final_url - def read(self) -> bytes: - return self._body + def read(self, size: int = -1) -> bytes: + if size < 0: + size = len(self._body) - self._offset + start = self._offset + self._offset = min(len(self._body), self._offset + size) + return self._body[start:self._offset] def test_http_fetch_uses_shared_client_and_rejects_redirect_downgrade(monkeypatch): @@ -71,6 +76,34 @@ def fake_open_url(url, timeout=10, extra_headers=None, redirect_validator=None): fetcher(_source("https://example.com/c.json")) +def test_http_fetch_bounds_catalog_response(monkeypatch): + body = b'{"schema_version":"1.0","bundles":{}}' + + def fake_open_url(url, timeout=10, extra_headers=None, redirect_validator=None): + return _FakeResponse(body, url) + + monkeypatch.setattr("specify_cli.authentication.http.open_url", fake_open_url) + monkeypatch.setattr(adapters, "MAX_JSON_CATALOG_BYTES", len(body) - 1) + + fetcher = adapters.make_catalog_fetcher(allow_network=True) + with pytest.raises(BundlerError, match="exceeds maximum size"): + fetcher(_source("https://example.com/c.json")) + + +@pytest.mark.parametrize( + "url", + [ + "https://[::1", + "https://example.com:notaport/catalog.json", + "https://example.com:70000/catalog.json", + ], +) +def test_fetch_rejects_malformed_source_url_cleanly(url): + fetcher = adapters.make_catalog_fetcher(allow_network=True) + with pytest.raises(BundlerError, match="URL is malformed"): + fetcher(_source(url)) + + def test_builtin_community_catalog_fetches_repository_catalog_online(monkeypatch): captured: dict = {}