diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index b4f6d28add..e41e6e203e 100644 --- a/src/specify_cli/_download_security.py +++ b/src/specify_cli/_download_security.py @@ -1,10 +1,19 @@ -"""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 struct +import unicodedata +import zipfile +from collections.abc import Iterator +from contextlib import ExitStack, contextmanager from ipaddress import IPv4Address, IPv6Address, ip_address +from itertools import pairwise +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import NoReturn, TypeVar from urllib.parse import ParseResult, urlparse @@ -12,17 +21,42 @@ 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 +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 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 + +_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( @@ -179,6 +213,41 @@ 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 + + +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, *, @@ -199,20 +268,483 @@ 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(unicodedata.category(character) == "Cc" 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( + 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. + """ + _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!r}", exc) + if info.file_size > max_bytes: + _raise( + error_type, + f"ZIP member {member_label!r} exceeds maximum size of {max_bytes} bytes", + ) + + try: + with zf.open(name, "r") as source: + return _read_limited(source, max_bytes) + except _ReadLimitExceeded: + _raise( + error_type, + 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, + ) + + +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}") + + 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 + # 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!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( + 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((" ", ".")) + 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, + *, + 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.""" + _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) + + 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: + _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]] = [] + validated_paths: dict[tuple[str, ...], tuple[str, bool]] = {} + total_size = 0 + for member in members: + 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) + + 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): + _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)) + + # 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 + # 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 Exception 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/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/__init__.py b/src/specify_cli/extensions/__init__.py index 00a6d92b41..969246e013 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,13 @@ from packaging.specifiers import InvalidSpecifier, SpecifierSet 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, +) 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 +2059,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 +2854,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 +3049,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, @@ -3195,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 @@ -3208,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}" ) @@ -3221,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) @@ -3238,7 +3257,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..2793424706 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -23,6 +23,14 @@ from .._console import console 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, + read_zip_member_limited, +) extension_app = typer.Typer( name="extension", @@ -435,14 +443,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) @@ -525,7 +536,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( @@ -1146,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) @@ -1224,24 +1241,70 @@ 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() - # 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 {} - 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 {} + # 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. + # 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 = normalize_zip_member_name(name) + 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, manifest_path) + ) + 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'") @@ -1262,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 @@ -1328,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/presets/__init__.py b/src/specify_cli/presets/__init__.py index 52494fede7..abe117adfe 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,13 @@ from packaging import version as pkg_version from packaging.specifiers import SpecifierSet, InvalidSpecifier +from .._download_security import ( + MAX_JSON_CATALOG_BYTES, + build_safe_download_path, + is_https_or_localhost_http, + 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 +1862,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 +2446,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 +2615,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, @@ -2774,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 @@ -2786,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) @@ -2814,7 +2834,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/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 f5dffe10de..583b85d99b 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -1,15 +1,23 @@ -"""Tests for bounded HTTP download helpers.""" +"""Tests for bounded download and ZIP extraction helpers.""" 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, + read_zip_member_limited, + safe_extract_zip, ) @@ -112,8 +120,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: @@ -160,6 +166,55 @@ 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"): @@ -171,9 +226,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 +277,572 @@ def test_read_response_limited_rejects_first_byte_at_zero_limit(): max_bytes=0, error_type=_CustomLimitError, ) + + +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, + "delete\x7f", + "csi\x9b[2J", + "\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", + [ + "../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 _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("= 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 = {}