diff --git a/docs/reference/bundles.md b/docs/reference/bundles.md index 139028a377..52e8b3e957 100644 --- a/docs/reference/bundles.md +++ b/docs/reference/bundles.md @@ -97,6 +97,8 @@ Re-resolves a bundle and **refreshes** its components through each primitive's u > **Pin enforcement is install-time only.** Idempotency checks are id-based, not version-aware: a component that is already present is skipped during `install` without comparing its on-disk version to the manifest pin. Version pins are therefore guaranteed to be applied only when the bundler actually installs a component for the first time or refreshes it. Run `specify bundle update ` for catalog bundles or `specify bundle install --refresh` for local sources to re-apply owned components at their pinned versions. +**Pinned releases are retrieved by pin, not by the catalog's current listing.** When a component's catalog entry has moved to a newer version, extensions and presets are not installed at the newer advertised release: the bundler derives the pinned release's download URL from the catalog entry's own `download_url` (same host, HTTPS-validated) and retrieves that pinned release. The catalog's SHA-256 digest covers only the advertised release, so a retrieved pinned release is verified against no digest; instead, before it is installed, the bundler verifies the retrieved archive's own manifest declares the pinned component (id and version), and removes the archive if it does not. When the pinned release cannot be identified from the catalog's download URL, cannot be retrieved, or fails that verification, the install fails with an error that names both the pinned and the advertised versions. + ## Remove a Bundle ```bash diff --git a/src/specify_cli/bundles/primitives.py b/src/specify_cli/bundles/primitives.py index b32342e68d..16c50b1d9f 100644 --- a/src/specify_cli/bundles/primitives.py +++ b/src/specify_cli/bundles/primitives.py @@ -22,7 +22,7 @@ import contextlib import os from pathlib import Path -from typing import Protocol +from typing import Callable, Protocol from . import BundlerError from .manifest import ComponentRef @@ -30,6 +30,26 @@ DEFAULT_PRIORITY = 10 +def _pinned_release_matches(pinned: str | None, advertised: object) -> bool: + """Return whether a manifest pin is satisfied by an advertised version. + + Mirrors the normalization of :func:`_assert_pinned_version`: a missing + pin, or a source that advertises no version, cannot be checked, so both + count as matching. + """ + if not pinned or advertised is None: + return True + actual = str(advertised).strip() + if not actual: + return True + from .versioning import parse_version + + try: + return parse_version(actual) == parse_version(pinned) + except BundlerError: + return actual == str(pinned).strip() + + def _assert_pinned_version( kind: str, component_id: str, pinned: str | None, advertised: object ) -> None: @@ -41,23 +61,270 @@ def _assert_pinned_version( enforce the pin, so installation proceeds (the source, not the bundler, owns that gap). """ - if not pinned or advertised is None: + if _pinned_release_matches(pinned, advertised): return - actual = str(advertised).strip() - if not actual: - return - from .versioning import parse_version + actual = str(advertised).strip() if advertised is not None else "" + raise BundlerError( + f"{kind} '{component_id}' is pinned to version {pinned} in the bundle " + f"manifest, but the resolved version is {actual}. Update the bundle's " + "pinned version or the source before installing." + ) + + +def _replace_version_token(segment: str, token: str, prefix: str, pinned: str) -> str | None: + """Substitute *token* with *pinned* when it appears as a bounded token. + + Returns the rewritten path segment, or ``None`` when no bounded + occurrence exists. A token is bounded when it is not embedded in a + longer dotted/dashed run: a preceding ``.`` is accepted only when it is + not itself preceded by a digit, and a following ``.`` only when it is + not followed by one -- so an advertised ``0.5.1`` never matches inside + ``1.0.5.1``, ``0.5.10`` or ``10.5.1``. + """ + start = 0 + while True: + pos = segment.find(token, start) + if pos < 0: + return None + before = segment[pos - 1] if pos > 0 else "" + end = pos + len(token) + after = segment[end] if end < len(segment) else "" + before_ok = ( + before == "" + or before in "-_/" + or (before == "." and (pos < 2 or not segment[pos - 2].isdigit())) + ) + after_ok = ( + after == "" + or after in "-_/" + or ( + after == "." + and (end + 1 >= len(segment) or not segment[end + 1].isdigit()) + ) + ) + if before_ok and after_ok: + return segment[:pos] + prefix + pinned + segment[end:] + start = pos + 1 + + +def _pinned_release_url( + download_url: object, advertised: object, pinned: str | None +) -> str | None: + """Derive the pinned release's URL from the advertised release's URL. + + Catalog entries advertise a single (version, download_url) pair, so when + a catalog moves to a newer release a bundle's pinned release is no longer + advertised -- although its artifact usually remains reachable at the same + location with the version token substituted (e.g. a GitHub release + download URL pinned to a tag). Returns such a derived URL, or ``None`` + when the advertised version token does not appear as a distinct token in + the URL path and no derivation is possible. + + Only the URL path is rewritten (query strings are left untouched), so + the derivation stays conservative: the same host, scheme, and any + authentication the advertised URL carries are preserved. + + Version tokens are compared in their bare form: both the advertised + version and the pin may carry an optional v/V prefix (bundle manifest + validation accepts it), and a URL token keeps its own prefix -- a pin of + ``v0.4.12`` must derive ``v0.4.12``, never ``vv0.4.12``. + + Substitution is restricted to recognized version positions: the final + segment (an asset filename such as ``xt-0.5.1.zip``) and a segment that + is *exactly* a version token (a release tag or a versioned directory, + e.g. ``releases/download/v0.5.1/`` or ``/v0.5.1/xt.zip``). Any other + segment is static path that merely *contains* a version-looking run -- + a repository named ``tool-0.5.1``, for example -- and is left + untouched, so the derivation never moves the component's home + repository. + """ + if not isinstance(download_url, str) or not download_url: + return None + if not pinned or advertised is None: + return None + advertised = str(advertised).strip() + pinned = str(pinned).strip() + if not advertised or not pinned: + return None + if _pinned_release_matches(pinned, advertised): + return None + bare_pinned = pinned[1:] if pinned[:1] in ("v", "V") else pinned + + from urllib.parse import urlunparse, urlparse try: - matches = parse_version(actual) == parse_version(pinned) - except BundlerError: - matches = actual == str(pinned).strip() - if not matches: - raise BundlerError( - f"{kind} '{component_id}' is pinned to version {pinned} in the bundle " - f"manifest, but the resolved version is {actual}. Update the bundle's " - "pinned version or the source before installing." + parts = urlparse(download_url) + except ValueError: + return None + if not parts.path: + return None + + if advertised[:1] in ("v", "V"): + # The catalog spells the version v/V-prefixed. Prefer the prefixed + # token (a tag segment such as "v0.5.1"), then the bare form, which + # is how a versioned asset filename spells it ("asset-0.5.1.zip"). + candidates: list[tuple[str, str]] = [ + (advertised, advertised[:1]), + (advertised[1:], ""), + ] + else: + candidates = [ + (f"V{advertised}", "V"), + (f"v{advertised}", "v"), + (advertised, ""), + ] + + candidate_tokens = {token for token, _ in candidates} + segments = parts.path.split("/") + last_index = len(segments) - 1 + changed = False + for index, segment in enumerate(segments): + # Only recognized version positions are rewritten: the asset + # filename (the final segment) and a segment that is exactly a + # version token (a release tag or a versioned directory). Any other + # segment is static path that merely contains a version-looking run + # -- a repository named "tool-0.5.1", for example -- and is left + # untouched so the derivation never moves the component's home repo. + if index != last_index and segment not in candidate_tokens: + continue + for token, prefix in candidates: + replaced = _replace_version_token(segment, token, prefix, bare_pinned) + if replaced is not None: + segments[index] = replaced + changed = True + break + if not changed: + return None + return urlunparse(parts._replace(path="/".join(segments))) + + +def _verify_pinned_release_archive( + archive_path: Path, kind: str, component_id: str, pinned: str +) -> None: + """Verify that a retrieved pinned-release archive declares the pin. + + A successful request to a *derived* URL does not prove that the served + artifact is the pinned release: a redirect, a fallback response, or a + mislabeled historical asset can serve a different component or version. + The catalog's SHA-256 covers only the advertised release, and the + primitive installers trust the archive's manifest, so the bundler checks + the extracted manifest's ID and normalized version against the + ``ComponentRef`` before anything is installed. The lookup mirrors the + installers' own: the manifest at the archive root or inside a single + top-level directory. + """ + import tempfile + + import yaml + + from .._download_security import safe_extract_archive + + manifest_name = "extension.yml" if kind == "Extension" else "preset.yml" + root_key = "extension" if kind == "Extension" else "preset" + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + safe_extract_archive(archive_path, root, error_type=BundlerError) + + manifest_path = root / manifest_name + if not manifest_path.exists(): + subdirs = [d for d in root.iterdir() if d.is_dir()] + if len(subdirs) == 1: + manifest_path = subdirs[0] / manifest_name + if not manifest_path.exists(): + raise BundlerError(f"no {manifest_name} in the retrieved archive") + try: + data = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, yaml.YAMLError) as exc: + raise BundlerError( + f"unreadable {manifest_name} in the retrieved archive: {exc}" + ) from exc + + section = data.get(root_key) if isinstance(data, dict) else None + if not isinstance(section, dict): + raise BundlerError(f"no {root_key} section in the retrieved archive") + actual_id = section.get("id") + if not isinstance(actual_id, str) or actual_id != component_id: + raise BundlerError( + f"the retrieved archive declares {kind.lower()} id " + f"'{actual_id}' instead of '{component_id}'" + ) + actual_version = section.get("version") + if not isinstance(actual_version, str) or not actual_version.strip(): + raise BundlerError( + "the retrieved archive declares no version in its manifest" + ) + if not _pinned_release_matches(pinned, actual_version): + raise BundlerError( + f"the retrieved archive declares version '{actual_version}' " + f"instead of the pinned version {pinned}" + ) + + +def _download_catalog_component( + download_by_id: Callable[..., Path], + download_by_url: Callable[..., Path], + kind: str, + component: ComponentRef, + info: dict, + *, + error_types: tuple[type[Exception], ...], +) -> Path: + """Download a catalog component, fetching the pinned release on demand. + + *download_by_id* is the catalog's standard ID-based download; + *download_by_url* is its explicit-URL counterpart. When the bundle's pin + differs from the version the catalog currently advertises, the pinned + release's URL is derived from the catalog's own ``download_url`` (same + host, re-validated as HTTPS by the catalog's download path) and + retrieved without the catalog's SHA-256, which only covers the + advertised release. The retrieved archive's manifest is then verified + to declare the pinned component (ID and version) before it is installed. + When no derivation is possible, the retrieval fails, or the verification + fails, the error names the pin and the advertised version so the failure + reports the pin mismatch rather than a bare network error. + """ + pinned = component.version + if pinned and not _pinned_release_matches(pinned, info.get("version")): + advertised = str(info.get("version")).strip() + derived = _pinned_release_url( + info.get("download_url"), info.get("version"), pinned ) + if derived is None: + raise BundlerError( + f"{kind} '{component.id}' is pinned to version {pinned} in the " + f"bundle manifest, but the catalog now advertises {advertised} " + "and its download URL does not identify that version, so the " + "pinned release cannot be located. Update the bundle's pinned " + "version to match the catalog, or restore the pinned release, " + "before installing." + ) + try: + archive_path = download_by_url(derived, component.id, pinned) + except error_types as exc: + raise BundlerError( + f"{kind} '{component.id}' is pinned to version {pinned} in the " + f"bundle manifest, but the catalog now advertises {advertised}. " + f"Retrieving the pinned release from {derived} failed: {exc} " + "Update the bundle's pinned version or the catalog before " + "installing." + ) from exc + try: + _verify_pinned_release_archive(archive_path, kind, component.id, pinned) + except BundlerError as exc: + # The catalog digest does not cover this release; an unverified + # artifact must not be left behind for a later install to reuse. + with contextlib.suppress(Exception): + if archive_path.exists(): + archive_path.unlink() + raise BundlerError( + f"{kind} '{component.id}' is pinned to version {pinned} in the " + f"bundle manifest, but the catalog now advertises {advertised}: " + f"{exc}. Update the bundle's pinned version or the catalog " + "before installing." + ) from exc + return archive_path + return download_by_id(component.id) def _bundled_manifest_version(manifest_path: Path, root_key: str) -> str | None: @@ -192,7 +459,7 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: "network access; re-run without --offline." ) - from ..presets import PresetCatalog + from ..presets import PresetCatalog, PresetError catalog = PresetCatalog(self._root) info = catalog.get_pack_info(component.id) @@ -203,10 +470,14 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: f"Preset '{component.id}' is from a discovery-only catalog; " "installation is not allowed." ) - _assert_pinned_version( - "Preset", component.id, component.version, info.get("version") + zip_path = _download_catalog_component( + catalog.download_pack, + catalog.download_pack_url, + "Preset", + component, + info, + error_types=(PresetError,), ) - zip_path = catalog.download_pack(component.id) try: self._manager.install_from_zip( zip_path, @@ -280,7 +551,7 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: "network access; re-run without --offline." ) - from ..extensions import ExtensionCatalog + from ..extensions import ExtensionCatalog, ExtensionError catalog = ExtensionCatalog(self._root) info = catalog.get_extension_info(component.id) @@ -293,10 +564,14 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: f"Extension '{component.id}' is from a discovery-only catalog; " "installation is not allowed." ) - _assert_pinned_version( - "Extension", component.id, component.version, info.get("version") + zip_path = _download_catalog_component( + catalog.download_extension, + catalog.download_extension_url, + "Extension", + component, + info, + error_types=(ExtensionError,), ) - zip_path = catalog.download_extension(component.id) try: manifest = self._manager.install_from_zip( zip_path, diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index e4b9e7de9d..56df1c9fc2 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -4369,8 +4369,6 @@ def download_extension( Raises: ExtensionError: If extension not found or download fails """ - import urllib.error - # Get extension info from catalog ext_info = self.get_extension_info(extension_id) if not ext_info: @@ -4392,14 +4390,64 @@ def download_extension( f"Extension download URL is malformed: {download_url}" ) + return self.download_extension_url( + download_url, + extension_id, + ext_info.get("version"), + sha256=ext_info.get("sha256"), + target_dir=target_dir, + ) + + def download_extension_url( + self, + download_url: str, + extension_id: str, + version: Optional[str] = None, + *, + sha256: Optional[str] = None, + target_dir: Optional[Path] = None, + ) -> Path: + """Download an extension archive from an explicit URL. + + The same pipeline ``download_extension`` applies to a catalog entry's + ``download_url`` (HTTPS validation, size-limited fetch, optional + SHA-256 verification, archive-format detection, safe cache path), + without a catalog lookup, so callers can retrieve a specific release + by URL -- e.g. a bundle pin the catalog no longer advertises. + ``sha256`` defaults to ``None`` (no digest check): a release that is + not the catalog's advertised one has no catalog-declared digest to + verify against. + + Args: + download_url: HTTPS URL of the archive to download + extension_id: ID used to name the cached archive + version: Version used to name the cached archive ("unknown" + when None) + sha256: Expected SHA-256 hex digest, or None to skip the check + target_dir: Directory to save the archive + + Returns: + Path to the downloaded archive + + Raises: + ExtensionError: If the URL is invalid or the download fails + """ + import urllib.error + # Validate download URL requires HTTPS (prevent man-in-the-middle attacks) from urllib.parse import urlparse - # A malformed authority (e.g. an unterminated IPv6 bracket + if not isinstance(download_url, str): + raise ExtensionError( + f"Extension download URL is malformed: {download_url}" + ) + + # A malformed authority (e.g., an unterminated IPv6 bracket # "https://[::1") makes urlparse / hostname access raise ValueError. - # The download_url comes from catalog payload data, so surface a clean - # ExtensionError rather than leaking a raw ValueError past the command - # handler (which only catches ExtensionError). Mirrors catalogs (#3435) + # The URL comes from caller data (a catalog payload field or a + # derived pinned-release URL), so surface a clean ExtensionError + # rather than leaking a raw ValueError past the command handler + # (which only catches ExtensionError). Mirrors catalogs (#3435) # and workflows/catalog.py (#3484). try: parsed = urlparse(download_url) @@ -4422,7 +4470,7 @@ def download_extension( if target_dir is None: target_dir = self.cache_dir / "downloads" target_dir = Path(target_dir) - version = ext_info.get("version", "unknown") + version = "unknown" if version is None else version declared_format = archive_format_from_name(download_url) build_safe_download_path( target_dir, @@ -4463,7 +4511,7 @@ def download_extension( ) verify_archive_sha256( - archive_data, ext_info.get("sha256"), extension_id, ExtensionError + archive_data, sha256, extension_id, ExtensionError ) with tempfile.NamedTemporaryFile( diff --git a/src/specify_cli/presets/_catalog.py b/src/specify_cli/presets/_catalog.py index a4768cc22d..0791a6d82e 100644 --- a/src/specify_cli/presets/_catalog.py +++ b/src/specify_cli/presets/_catalog.py @@ -771,10 +771,6 @@ def download_pack( Raises: PresetError: If pack not found or download fails """ - import urllib.error - - from . import read_response_limited, verify_archive_sha256 - pack_info = self.get_pack_info(pack_id) if not pack_info: raise PresetError( @@ -808,14 +804,66 @@ def download_pack( f"Preset download URL is malformed: {download_url}" ) + return self.download_pack_url( + download_url, + pack_id, + pack_info.get("version"), + sha256=pack_info.get("sha256"), + target_dir=target_dir, + ) + + def download_pack_url( + self, + download_url: str, + pack_id: str, + version: Optional[str] = None, + *, + sha256: Optional[str] = None, + target_dir: Optional[Path] = None, + ) -> Path: + """Download a preset archive from an explicit URL. + + The same pipeline ``download_pack`` applies to a catalog entry's + ``download_url`` (HTTPS validation, size-limited fetch, optional + SHA-256 verification, archive-format detection, safe cache path), + without a catalog lookup, so callers can retrieve a specific release + by URL -- e.g. a bundle pin the catalog no longer advertises. + ``sha256`` defaults to ``None`` (no digest check): a release that is + not the catalog's advertised one has no catalog-declared digest to + verify against. + + Args: + download_url: HTTPS URL of the archive to download + pack_id: ID used to name the cached archive + version: Version used to name the cached archive ("unknown" + when None) + sha256: Expected SHA-256 hex digest, or None to skip the check + target_dir: Directory to save the archive + + Returns: + Path to the downloaded archive + + Raises: + PresetError: If the URL is invalid or the download fails + """ + import urllib.error + + from . import read_response_limited, verify_archive_sha256 + + if not isinstance(download_url, str): + raise PresetError( + f"Preset download URL is malformed: {download_url}" + ) + from urllib.parse import urlparse - # A malformed authority (e.g. an unterminated IPv6 bracket + # A malformed authority (e.g., an unterminated IPv6 bracket # "https://[::1") makes urlparse / hostname access raise ValueError. - # The download_url comes from catalog payload data, so surface a clean - # PresetError rather than leaking a raw ValueError past the command - # handler (which only catches PresetError). Mirrors catalogs (#3435) - # and workflows/catalog.py (#3484). + # The URL comes from caller data (a catalog payload field or a + # derived pinned-release URL), so surface a clean PresetError rather + # than leaking a raw ValueError past the command handler (which only + # catches PresetError). Mirrors catalogs (#3435) and + # workflows/catalog.py (#3484). try: parsed = urlparse(download_url) hostname = parsed.hostname @@ -836,7 +884,7 @@ def download_pack( if target_dir is None: target_dir = self.cache_dir / "downloads" target_dir = Path(target_dir) - version = pack_info.get("version", "unknown") + version = "unknown" if version is None else version declared_format = archive_format_from_name(download_url) build_safe_download_path( target_dir, @@ -875,7 +923,7 @@ def download_pack( ) verify_archive_sha256( - archive_data, pack_info.get("sha256"), pack_id, PresetError + archive_data, sha256, pack_id, PresetError ) with tempfile.NamedTemporaryFile( diff --git a/tests/specify_cli/bundles/test_primitives.py b/tests/specify_cli/bundles/test_primitives.py index a3b4d83f45..ca686aa54e 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -125,6 +125,564 @@ def test_assert_pinned_version_mismatch_raises(): _assert_pinned_version("Preset", "preset-a", "2.0.0", "3.1.0") +def test_pinned_release_matches_normalizes_like_assert(): + from specify_cli.bundles.primitives import _pinned_release_matches + + # Equal (including v-prefix/normalization) matches; missing pin or + # unadvertised version cannot be checked and matches (proceed). + assert _pinned_release_matches("2.0.0", "2.0.0") + assert _pinned_release_matches("2.0.0", "v2.0.0") + assert _pinned_release_matches(None, "9.9.9") + assert _pinned_release_matches("2.0.0", None) + assert not _pinned_release_matches("0.4.12", "0.5.1") + assert not _pinned_release_matches("2.0.0", "3.1.0") + + +def test_pinned_release_url_derives_versioned_download_url(): + from specify_cli.bundles.primitives import _pinned_release_url + + base = "https://github.com/acme/xt/releases/download" + # A GitHub release download URL: both the tag segment and a versioned + # asset filename are rewritten. + assert _pinned_release_url(f"{base}/v0.5.1/xt-0.5.1.zip", "0.5.1", "0.4.12") == ( + f"{base}/v0.4.12/xt-0.4.12.zip" + ) + # A versioned filename alone (no tag segment) is derivable too. + assert _pinned_release_url( + "https://example.com/xt-0.5.1.zip", "0.5.1", "0.4.12" + ) == "https://example.com/xt-0.4.12.zip" + # An archive tag URL with the version glued to the suffix. + assert _pinned_release_url( + "https://github.com/acme/xt/archive/refs/tags/v0.5.1.zip", "0.5.1", "0.4.12" + ) == "https://github.com/acme/xt/archive/refs/tags/v0.4.12.zip" + + +def test_pinned_release_url_tolerates_v_prefixed_advertised_version(): + from specify_cli.bundles.primitives import _pinned_release_url + + # The catalog advertises the v-prefixed form; the bundle pin is bare semver. + assert _pinned_release_url( + "https://example.com/v0.5.1/xt.zip", "v0.5.1", "0.4.12" + ) == "https://example.com/v0.4.12/xt.zip" + + +def test_pinned_release_url_normalizes_v_prefixed_pin(): + from specify_cli.bundles.primitives import _pinned_release_url + + # The pin itself may be v-prefixed (manifest validation accepts it); the + # URL token keeps its own prefix, so the result must not be "vv0.4.12". + assert _pinned_release_url( + "https://github.com/acme/xt/releases/download/v0.5.1/xt-0.5.1.zip", + "0.5.1", + "v0.4.12", + ) == "https://github.com/acme/xt/releases/download/v0.4.12/xt-0.4.12.zip" + + +def test_pinned_release_url_rewrites_bare_tokens_for_v_prefixed_advertised(): + from specify_cli.bundles.primitives import _pinned_release_url + + # A v-prefixed advertised version also appears bare in a versioned asset + # filename; both the tag segment and the filename must be rewritten. + assert _pinned_release_url( + "https://github.com/acme/xt/releases/download/v0.5.1/xt-0.5.1.zip", + "v0.5.1", + "0.4.12", + ) == "https://github.com/acme/xt/releases/download/v0.4.12/xt-0.4.12.zip" + + +def test_pinned_release_url_keeps_version_like_static_path_components(): + from specify_cli.bundles.primitives import _pinned_release_url + + # A repository named "tool-0.5.1" is a static component: the rewrite is + # restricted to recognized version positions (the release tag and the + # asset filename), so the component's home repository is preserved. + assert _pinned_release_url( + "https://github.com/acme/tool-0.5.1/releases/download/v0.5.1/tool.zip", + "0.5.1", + "0.4.12", + ) == "https://github.com/acme/tool-0.5.1/releases/download/v0.4.12/tool.zip" + assert _pinned_release_url( + "https://github.com/acme/tool-0.5.1/archive/refs/tags/v0.5.1.zip", + "0.5.1", + "0.4.12", + ) == "https://github.com/acme/tool-0.5.1/archive/refs/tags/v0.4.12.zip" + # The same holds for a version-looking static component on a non-GitHub + # host: only the final (filename) segment is rewritten. + assert _pinned_release_url( + "https://example.com/tools/tool-0.5.1/dist-0.5.1.zip", "0.5.1", "0.4.12" + ) == "https://example.com/tools/tool-0.5.1/dist-0.4.12.zip" + # A segment that is exactly a version token is still a version position + # (a versioned directory), even in a non-final position. + assert _pinned_release_url( + "https://example.com/v0.5.1/xt.zip", "v0.5.1", "0.4.12" + ) == "https://example.com/v0.4.12/xt.zip" + + +def test_pinned_release_url_refuses_ambiguous_or_missing_tokens(): + from specify_cli.bundles.primitives import _pinned_release_url + + # No version token in the path: nothing to derive. + assert ( + _pinned_release_url("https://example.com/xt.zip", "0.5.1", "0.4.12") is None + ) + # A query-string-only version is not a path token. + assert ( + _pinned_release_url( + "https://example.com/xt.zip?tag=0.5.1", "0.5.1", "0.4.12" + ) + is None + ) + # Embedded runs must not partial-match: 0.5.1 inside 10.5.1 / 0.5.10. + assert ( + _pinned_release_url("https://example.com/10.5.1/xt.zip", "0.5.1", "0.4.12") + is None + ) + assert ( + _pinned_release_url("https://example.com/0.5.10/xt.zip", "0.5.1", "0.4.12") + is None + ) + # No advertised version, or no pin: nothing to derive. + assert _pinned_release_url("https://example.com/v0.5.1/xt.zip", None, "0.4.12") is None + assert _pinned_release_url("https://example.com/v0.5.1/xt.zip", "0.5.1", None) is None + # Advertised and pinned already agree: not a mismatch case. + assert ( + _pinned_release_url("https://example.com/v0.5.1/xt.zip", "0.5.1", "0.5.1") + is None + ) + # Non-string URLs are refused outright. + assert _pinned_release_url(None, "0.5.1", "0.4.12") is None + assert _pinned_release_url(123, "0.5.1", "0.4.12") is None + + +def _extension_zip( + tmp_path: Path, ext_id: str = "my-ext", version: str = "1.0.0" +) -> Path: + """Zip the minimal real extension from ``_write_extension_with_config``.""" + import zipfile + + source = tmp_path / f"{ext_id}-source" + _write_extension_with_config(source, ext_id=ext_id, version=version) + zip_path = tmp_path / f"{ext_id}.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + for f in source.rglob("*"): + if f.is_file(): + zf.write(f, f.relative_to(source)) + return zip_path + + +def _preset_zip( + tmp_path: Path, preset_id: str = "p", version: str = "0.4.12" +) -> Path: + """Zip a minimal preset pack (manifest only; the manager is faked).""" + import zipfile + + import yaml + + source = tmp_path / f"{preset_id}-source" + source.mkdir(parents=True, exist_ok=True) + (source / "preset.yml").write_text( + yaml.dump( + { + "schema_version": "1.0", + "preset": { + "id": preset_id, + "name": preset_id, + "version": version, + "description": "Test preset", + }, + "requires": {"speckit_version": ">=0.1.0"}, + } + ), + encoding="utf-8", + ) + zip_path = tmp_path / f"{preset_id}.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.write(source / "preset.yml", "preset.yml") + return zip_path + + +def test_catalog_extension_pin_mismatch_fetches_pinned_release( + tmp_path: Path, monkeypatch +): + """Regression (#4712): when the catalog no longer advertises the bundle's + pinned version, the pinned release must be retrieved from a URL derived + from the catalog's own download_url instead of failing the install.""" + import specify_cli._assets as assets + from specify_cli.extensions import ExtensionCatalog + + project = tmp_path / "project" + project.mkdir() + # The archive's manifest must declare the pinned component (the bundle + # layer verifies a retrieved pinned release before installing it). + zip_path = _extension_zip(tmp_path, ext_id="xt", version="0.4.12") + downloads = [] + standard_calls = [] + + def _standard(*args, **kwargs): + standard_calls.append(args) + + monkeypatch.setattr(assets, "_locate_bundled_extension", lambda cid: None) + monkeypatch.setattr( + ExtensionCatalog, + "get_extension_info", + lambda self, eid: { + "id": eid, + "version": "0.5.1", + "download_url": ( + "https://github.com/acme/xt/releases/download/v0.5.1/xt-0.5.1.zip" + ), + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, + ) + monkeypatch.setattr( + ExtensionCatalog, + "download_extension", + lambda self, eid: standard_calls.append(eid), + ) + monkeypatch.setattr( + ExtensionCatalog, + "download_extension_url", + lambda self, url, eid, version, **kw: ( + downloads.append((url, eid, version)) or zip_path + ), + ) + + manager = primitive_manager("extensions", project, allow_network=True) + manager.install(ComponentRef(kind="extensions", id="xt", version="0.4.12")) + + assert downloads == [ + ( + "https://github.com/acme/xt/releases/download/v0.4.12/xt-0.4.12.zip", + "xt", + "0.4.12", + ) + ] + assert standard_calls == [] + + +def test_catalog_extension_pin_mismatch_unresolvable_reports_pin( + tmp_path: Path, monkeypatch +): + """When the catalog's download_url carries no version token, a stale pin + fails with a pin-aware error naming the pin and the advertised version — + not a silent substitute, and not a bare pin mismatch (issue #4712).""" + import specify_cli._assets as assets + from specify_cli.extensions import ExtensionCatalog + + calls = [] + monkeypatch.setattr(assets, "_locate_bundled_extension", lambda cid: None) + monkeypatch.setattr( + ExtensionCatalog, + "get_extension_info", + lambda self, eid: { + "id": eid, + "version": "0.5.1", + "download_url": "https://example.com/xt/latest.zip", + "_install_allowed": True, + }, + ) + monkeypatch.setattr( + ExtensionCatalog, "download_extension", lambda self, eid: calls.append(eid) + ) + monkeypatch.setattr( + ExtensionCatalog, + "download_extension_url", + lambda *a, **k: calls.append((a, k)), + ) + + manager = primitive_manager("extensions", tmp_path, allow_network=True) + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="extensions", id="xt", version="0.4.12")) + + message = str(exc.value) + assert "xt" in message + assert "pinned to version 0.4.12" in message + assert "0.5.1" in message + assert "cannot be located" in message + assert calls == [] + + +def test_catalog_extension_pin_mismatch_pinned_fetch_failure_reports_pin( + tmp_path: Path, monkeypatch +): + """A failed pinned-release retrieval (e.g. the release was deleted) + reports the pin and the derived URL, not just a bare network error.""" + import specify_cli._assets as assets + from specify_cli.extensions import ExtensionCatalog, ExtensionError + + monkeypatch.setattr(assets, "_locate_bundled_extension", lambda cid: None) + monkeypatch.setattr( + ExtensionCatalog, + "get_extension_info", + lambda self, eid: { + "id": eid, + "version": "0.5.1", + "download_url": ( + "https://github.com/acme/xt/releases/download/v0.5.1/xt-0.5.1.zip" + ), + "_install_allowed": True, + }, + ) + + def _boom(*args, **kwargs): + raise ExtensionError("HTTP Error 404: Not Found") + + monkeypatch.setattr( + ExtensionCatalog, + "download_extension", + lambda self, eid: (_ for _ in ()).throw(AssertionError("unused")), + ) + monkeypatch.setattr(ExtensionCatalog, "download_extension_url", _boom) + + manager = primitive_manager("extensions", tmp_path, allow_network=True) + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="extensions", id="xt", version="0.4.12")) + + message = str(exc.value) + assert "pinned to version 0.4.12" in message + assert "0.5.1" in message + assert ( + "https://github.com/acme/xt/releases/download/v0.4.12/xt-0.4.12.zip" + in message + ) + assert "HTTP Error 404" in message + + +def _pin_mismatch_extension_env(tmp_path: Path, monkeypatch): + """Monkeypatch a catalog that advertises 0.5.1 for 'xt' and wire the + pinned-release download to the caller's archive; return the project.""" + import specify_cli._assets as assets + from specify_cli.extensions import ExtensionCatalog + + project = tmp_path / "project" + project.mkdir() + monkeypatch.setattr(assets, "_locate_bundled_extension", lambda cid: None) + monkeypatch.setattr( + ExtensionCatalog, + "get_extension_info", + lambda self, eid: { + "id": eid, + "version": "0.5.1", + "download_url": ( + "https://github.com/acme/xt/releases/download/v0.5.1/xt-0.5.1.zip" + ), + "_install_allowed": True, + }, + ) + monkeypatch.setattr( + ExtensionCatalog, + "download_extension", + lambda self, eid: pytest.fail("standard download must not be used"), + ) + return project + + +def test_catalog_extension_pin_mismatch_rejects_release_declaring_other_version( + tmp_path: Path, monkeypatch +): + """A derived URL can serve a mislabeled artifact (a redirect, a fallback + response, a renamed historical asset). When the retrieved archive's + manifest declares a version other than the pinned one, the install is + refused and the unverified artifact is removed — the catalog digest only + covers the advertised release, so nothing else verifies the content.""" + import specify_cli.extensions as extensions + + project = _pin_mismatch_extension_env(tmp_path, monkeypatch) + retrieved = tmp_path / "retrieved.zip" + retrieved.write_bytes( + _extension_zip(tmp_path, ext_id="xt", version="0.5.1").read_bytes() + ) + monkeypatch.setattr( + extensions.ExtensionCatalog, + "download_extension_url", + lambda self, url, eid, version, **kw: retrieved, + ) + + manager = primitive_manager("extensions", project, allow_network=True) + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="extensions", id="xt", version="0.4.12")) + + message = str(exc.value) + assert "pinned to version 0.4.12" in message + assert "0.5.1" in message + assert "declares version '0.5.1'" in message + # The unverified artifact must not linger for a later install to reuse. + assert not retrieved.exists() + assert not manager.is_installed(ComponentRef(kind="extensions", id="xt")) + + +def test_catalog_extension_pin_mismatch_rejects_release_declaring_other_id( + tmp_path: Path, monkeypatch +): + """A retrieved archive must name the pinned component: an artifact whose + manifest declares a different extension id is a different component and + is refused, not installed under the pin.""" + import specify_cli.extensions as extensions + + project = _pin_mismatch_extension_env(tmp_path, monkeypatch) + retrieved = tmp_path / "retrieved.zip" + retrieved.write_bytes( + _extension_zip(tmp_path, ext_id="other-ext", version="0.4.12").read_bytes() + ) + monkeypatch.setattr( + extensions.ExtensionCatalog, + "download_extension_url", + lambda self, url, eid, version, **kw: retrieved, + ) + + manager = primitive_manager("extensions", project, allow_network=True) + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="extensions", id="xt", version="0.4.12")) + + message = str(exc.value) + assert "declares extension id 'other-ext' instead of 'xt'" in message + assert not retrieved.exists() + assert not manager.is_installed(ComponentRef(kind="extensions", id="xt")) + + +def test_catalog_preset_pin_mismatch_fetches_pinned_release(tmp_path: Path, monkeypatch): + """Presets follow the same pinned-release retrieval as extensions + (issue #4712).""" + import specify_cli._assets as assets + from specify_cli.presets import PresetCatalog + + # The archive's manifest must declare the pinned pack (the bundle layer + # verifies a retrieved pinned release before installing it). + archive = _preset_zip(tmp_path, preset_id="p", version="0.4.12") + downloads = [] + + class _FakeManager: + def install_from_zip(self, *args, **kwargs): + pass + + monkeypatch.setattr(assets, "_locate_bundled_preset", lambda _id: None) + monkeypatch.setattr( + PresetCatalog, + "get_pack_info", + lambda _self, _id: { + "version": "0.5.1", + "download_url": ( + "https://github.com/acme/preset/releases/download/v0.5.1/preset-0.5.1.zip" + ), + "_install_allowed": True, + "_catalog_name": "bundle-preset-catalog", + }, + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack", + lambda _self, _id: pytest.fail("standard download must not be used"), + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack_url", + lambda _self, url, pid, version, **kw: ( + downloads.append((url, pid, version)) or archive + ), + ) + + manager = primitive_manager("presets", tmp_path, allow_network=True) + manager._manager = _FakeManager() + manager.install(ComponentRef(kind="presets", id="p", version="0.4.12")) + + assert downloads == [ + ( + "https://github.com/acme/preset/releases/download/v0.4.12/preset-0.4.12.zip", + "p", + "0.4.12", + ) + ] + + +def test_catalog_preset_pin_mismatch_unresolvable_reports_pin( + tmp_path: Path, monkeypatch +): + import specify_cli._assets as assets + from specify_cli.presets import PresetCatalog + + monkeypatch.setattr(assets, "_locate_bundled_preset", lambda _id: None) + monkeypatch.setattr( + PresetCatalog, + "get_pack_info", + lambda _self, _id: { + "version": "0.5.1", + "download_url": "https://example.com/preset/latest.zip", + "_install_allowed": True, + }, + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack", + lambda _self, _id: pytest.fail("standard download must not be used"), + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack_url", + lambda *a, **k: pytest.fail("pinned retrieval must not be attempted"), + ) + + manager = primitive_manager("presets", tmp_path, allow_network=True) + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="presets", id="p", version="0.4.12")) + + message = str(exc.value) + assert "pinned to version 0.4.12" in message + assert "0.5.1" in message + assert "cannot be located" in message + + +def test_catalog_preset_pin_mismatch_rejects_release_declaring_other_version( + tmp_path: Path, monkeypatch +): + """Presets get the same verification: a retrieved archive whose manifest + declares a version other than the pin is refused before install.""" + import specify_cli._assets as assets + from specify_cli.presets import PresetCatalog + + retrieved = tmp_path / "retrieved.zip" + retrieved.write_bytes( + _preset_zip(tmp_path, preset_id="p", version="0.5.1").read_bytes() + ) + + class _FakeManager: + def install_from_zip(self, *args, **kwargs): + pytest.fail("install must not proceed for an unverified release") + + monkeypatch.setattr(assets, "_locate_bundled_preset", lambda _id: None) + monkeypatch.setattr( + PresetCatalog, + "get_pack_info", + lambda _self, _id: { + "version": "0.5.1", + "download_url": ( + "https://github.com/acme/preset/releases/download/v0.5.1/preset-0.5.1.zip" + ), + "_install_allowed": True, + }, + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack", + lambda _self, _id: pytest.fail("standard download must not be used"), + ) + monkeypatch.setattr( + PresetCatalog, + "download_pack_url", + lambda _self, url, pid, version, **kw: retrieved, + ) + + manager = primitive_manager("presets", tmp_path, allow_network=True) + manager._manager = _FakeManager() + with pytest.raises(BundlerError) as exc: + manager.install(ComponentRef(kind="presets", id="p", version="0.4.12")) + + message = str(exc.value) + assert "pinned to version 0.4.12" in message + assert "declares version '0.5.1'" in message + assert not retrieved.exists() + + def test_workflow_version_mismatch_refuses(tmp_path: Path, monkeypatch): from specify_cli.workflows.catalog import WorkflowCatalog @@ -295,7 +853,9 @@ def _fake_install(self, *a, **k): assert len(called) == 2 -def _write_extension_with_config(ext_dir: Path) -> None: +def _write_extension_with_config( + ext_dir: Path, ext_id: str = "my-ext", version: str = "1.0.0" +) -> None: """A minimal, real (unmocked) extension source with a provides.config entry.""" import yaml @@ -303,18 +863,21 @@ def _write_extension_with_config(ext_dir: Path) -> None: manifest = { "schema_version": "1.0", "extension": { - "id": "my-ext", + "id": ext_id, "name": "My Extension", - "version": "1.0.0", + "version": version, "description": "Test extension", }, "requires": {"speckit_version": ">=0.1.0"}, "provides": { "commands": [ - {"name": "speckit.my-ext.hello", "file": "commands/hello.md"}, + {"name": f"speckit.{ext_id}.hello", "file": "commands/hello.md"}, ], "config": [ - {"name": "my-ext-config.yml", "template": "config-template.yml"}, + { + "name": f"{ext_id}-config.yml", + "template": "config-template.yml", + }, ], }, } diff --git a/tests/specify_cli/presets/test_catalog.py b/tests/specify_cli/presets/test_catalog.py index 2ffbddfed9..5b4e99bcfa 100644 --- a/tests/specify_cli/presets/test_catalog.py +++ b/tests/specify_cli/presets/test_catalog.py @@ -1239,6 +1239,87 @@ def test_download_pack_preserves_tar_archive_format( assert archive_path.name == "test-pack-1.0.0.tar.gz" assert archive_path.read_bytes() == archive_bytes + def test_download_pack_url_downloads_without_catalog_lookup(self, project_dir): + """download_pack_url retrieves an explicit URL without consulting the + catalog, naming the archive from the caller-supplied id/version (the + #4712 pinned-release retrieval path).""" + from unittest.mock import patch + + catalog = PresetCatalog(project_dir) + zip_bytes, resp = self._pack_zip_and_response() + with patch.object( + catalog, + "get_pack_info", + side_effect=AssertionError( + "no catalog lookup expected for an explicit-URL download" + ), + ), patch.object(catalog, "_open_url", return_value=resp): + zip_path = catalog.download_pack_url( + "https://example.com/releases/download/v0.4.12/test-pack-0.4.12.zip", + "test-pack", + "0.4.12", + target_dir=project_dir, + ) + + assert zip_path.name == "test-pack-0.4.12.zip" + assert zip_path.read_bytes() == zip_bytes + + def test_download_pack_url_verifies_sha256_when_provided(self, project_dir): + """A supplied digest is enforced; omission (None) skips verification.""" + import hashlib + from unittest.mock import patch + + catalog = PresetCatalog(project_dir) + zip_bytes, resp = self._pack_zip_and_response() + + with patch.object(catalog, "_open_url", return_value=resp): + zip_path = catalog.download_pack_url( + "https://example.com/test-pack.zip", + "test-pack", + "1.0.0", + sha256=hashlib.sha256(zip_bytes).hexdigest(), + target_dir=project_dir, + ) + assert zip_path.read_bytes() == zip_bytes + + with patch.object(catalog, "_open_url", return_value=resp): + with pytest.raises(PresetError, match="Integrity check failed"): + catalog.download_pack_url( + "https://example.com/test-pack.zip", + "test-pack", + "1.0.0", + sha256="0" * 64, + target_dir=project_dir, + ) + + def test_download_pack_url_rejects_non_https(self, project_dir): + """Explicit-URL downloads enforce the same HTTPS rule as catalog + downloads (localhost HTTP aside).""" + from unittest.mock import patch + + catalog = PresetCatalog(project_dir) + with patch.object(catalog, "_open_url") as open_url: + with pytest.raises(PresetError, match="must use HTTPS"): + catalog.download_pack_url( + "http://example.com/test-pack.zip", + "test-pack", + target_dir=project_dir, + ) + open_url.assert_not_called() + + def test_download_pack_url_rejects_malformed_authority(self, project_dir): + """An explicit URL with a malformed authority surfaces a clean + PresetError rather than leaking a raw ValueError.""" + from unittest.mock import patch + + catalog = PresetCatalog(project_dir) + with patch.object(catalog, "_open_url") as open_url: + with pytest.raises(PresetError, match="malformed"): + catalog.download_pack_url( + "https://[::1/test-pack.zip", "test-pack", target_dir=project_dir + ) + open_url.assert_not_called() + class TestPresetCatalogEntry: """Test PresetCatalogEntry dataclass.""" diff --git a/tests/test_extensions.py b/tests/test_extensions.py index b512742389..c2d27efa7c 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -6662,6 +6662,89 @@ def test_download_extension_preserves_tar_archive_format( assert archive_path.name == "test-ext-1.0.0.tar.gz" assert archive_path.read_bytes() == archive_bytes + def test_download_extension_url_downloads_without_catalog_lookup(self, temp_dir): + """download_extension_url retrieves an explicit URL without consulting + the catalog, naming the archive from the caller-supplied id/version + (the #4712 pinned-release retrieval path).""" + from unittest.mock import patch + + catalog = self._make_catalog(temp_dir) + zip_bytes = self._make_zip_bytes() + with patch.object( + catalog, + "get_extension_info", + side_effect=AssertionError( + "no catalog lookup expected for an explicit-URL download" + ), + ), patch.object(catalog, "_open_url", return_value=self._mock_response(zip_bytes)): + zip_path = catalog.download_extension_url( + "https://example.com/releases/download/v0.4.12/test-ext-0.4.12.zip", + "test-ext", + "0.4.12", + target_dir=temp_dir, + ) + + assert zip_path.name == "test-ext-0.4.12.zip" + assert zip_path.read_bytes() == zip_bytes + + def test_download_extension_url_verifies_sha256_when_provided(self, temp_dir): + """A supplied digest is enforced; omission (None) skips verification, + matching the catalogue-optional behaviour of the advertised path.""" + import hashlib + from unittest.mock import patch + + catalog = self._make_catalog(temp_dir) + zip_bytes = self._make_zip_bytes() + digest = hashlib.sha256(zip_bytes).hexdigest() + + with patch.object(catalog, "_open_url", return_value=self._mock_response(zip_bytes)): + zip_path = catalog.download_extension_url( + "https://example.com/test-ext.zip", + "test-ext", + "1.0.0", + sha256=digest, + target_dir=temp_dir, + ) + assert zip_path.read_bytes() == zip_bytes + + with patch.object(catalog, "_open_url", return_value=self._mock_response(zip_bytes)): + with pytest.raises(ExtensionError, match="Integrity check failed"): + catalog.download_extension_url( + "https://example.com/test-ext.zip", + "test-ext", + "1.0.0", + sha256="0" * 64, + target_dir=temp_dir, + ) + + def test_download_extension_url_rejects_non_https(self, temp_dir): + """Explicit-URL downloads enforce the same HTTPS rule as catalog + downloads (localhost HTTP aside).""" + from unittest.mock import patch + + catalog = self._make_catalog(temp_dir) + with patch.object(catalog, "_open_url") as open_url: + with pytest.raises(ExtensionError, match="must use HTTPS"): + catalog.download_extension_url( + "http://example.com/test-ext.zip", + "test-ext", + target_dir=temp_dir, + ) + open_url.assert_not_called() + + def test_download_extension_url_rejects_malformed_authority(self, temp_dir): + """An explicit URL with a malformed authority surfaces a clean + ExtensionError rather than leaking a raw ValueError.""" + from unittest.mock import patch + + catalog = self._make_catalog(temp_dir) + with patch.object(catalog, "_open_url") as open_url: + with pytest.raises(ExtensionError, match="malformed"): + catalog.download_extension_url( + "https://[::1/test-ext.zip", "test-ext", target_dir=temp_dir + ) + open_url.assert_not_called() + # ===== CatalogEntry Tests =====