From 0b9ab28a981cb366c0ea89f219c055dbf5a5b135 Mon Sep 17 00:00:00 2001 From: muhammadumer-waheed Date: Fri, 25 Sep 2026 21:35:11 +0500 Subject: [PATCH 1/3] fix(bundles): retrieve pinned component releases when a catalog advertises a newer version Bundle manifests pin component versions for reproducibility, but extension/preset bundle installation compared the pin against the version the catalog currently advertises and refused the install when they differed, even when the pinned release remained available at its original download URL (#4712). Catalog entries advertise a single release, so the resolver never attempted to retrieve the pinned one. When the pin differs from the advertised version, the bundler now derives the pinned release's URL from the catalog entry's own download_url by substituting the advertised version token in the URL path (bounded token matching: a v-prefixed tag, a versioned asset filename, or a version glued to an archive suffix) and retrieves that release through the catalog's existing download pipeline (HTTPS validation, size limits, safe cache path). The catalog's SHA-256 covers only the advertised release, so a retrieved pinned release is verified against no digest; the same host and the HTTPS rule still apply. When the pinned release cannot be identified from the catalog URL, or the retrieval fails, the error names the pinned and advertised versions instead of a bare pin mismatch or network error. To expose explicit-URL retrieval without duplicating the download pipeline, the fetch stage of ExtensionCatalog.download_extension and PresetCatalog.download_pack is now download_extension_url / download_pack_url, which the ID-based methods delegate to without behavior change (a catalog entry with a null version now names the cached archive "unknown" instead of erroring). Workflows and bundled-asset installs keep their existing hard pin check: a workflow's URL install path carries an interactive untrusted-source confirmation that a derived-URL fetch would bypass, and a bundled asset has no alternative release to retrieve. New regression tests in tests/specify_cli/bundles/test_primitives.py fail on main (the pin mismatch raised before any retrieval attempt) and pass with this change, plus URL-derivation unit tests and explicit-URL download coverage in tests/test_extensions.py and tests/specify_cli/presets/test_catalog.py. Verified end-to-end with a local catalog advertising 0.5.1 while a 0.4.12 release stays available: `specify bundle install` now installs the pinned 0.4.12 release (on main it fails with the reported error). Fixes #4712 Assisted-by: opencode (model: Qwen3.8-27B (local), autonomous) --- docs/reference/bundles.md | 2 + src/specify_cli/bundles/primitives.py | 211 ++++++++++-- src/specify_cli/extensions/__init__.py | 64 +++- src/specify_cli/presets/_catalog.py | 70 +++- tests/specify_cli/bundles/test_primitives.py | 327 +++++++++++++++++++ tests/specify_cli/presets/test_catalog.py | 81 +++++ tests/test_extensions.py | 83 +++++ 7 files changed, 797 insertions(+), 41 deletions(-) diff --git a/docs/reference/bundles.md b/docs/reference/bundles.md index 139028a377..37f1d741f0 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. When the pinned release cannot be identified from the catalog's download URL, or cannot be retrieved, 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..dbfa29b897 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,162 @@ 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: - return - actual = str(advertised).strip() - if not actual: + if _pinned_release_matches(pinned, advertised): 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. + """ + 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 advertised == pinned: + return None + + 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; preserve that tag + # prefix form (the pin is bare semver, so the prefix is never its own + # version character). + candidates: list[tuple[str, str]] = [(advertised, advertised[:1])] + else: + candidates = [ + (f"V{advertised}", "V"), + (f"v{advertised}", "v"), + (advertised, ""), + ] + + segments = parts.path.split("/") + changed = False + for index, segment in enumerate(segments): + for token, prefix in candidates: + replaced = _replace_version_token(segment, token, prefix, 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 _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. When no derivation is possible, or the pinned + retrieval 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: + return 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 + return download_by_id(component.id) def _bundled_manifest_version(manifest_path: Path, root_key: str) -> str | None: @@ -192,7 +351,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 +362,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 +443,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 +456,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..ffcf813dda 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -125,6 +125,333 @@ 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_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) -> Path: + """Zip the minimal real extension from ``_write_extension_with_config``.""" + import zipfile + + source = tmp_path / "xt-source" + _write_extension_with_config(source) + zip_path = tmp_path / "xt.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 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() + zip_path = _extension_zip(tmp_path) + 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 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 + + archive = tmp_path / "preset.zip" + archive.write_bytes(b"placeholder") + 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_workflow_version_mismatch_refuses(tmp_path: Path, monkeypatch): from specify_cli.workflows.catalog import WorkflowCatalog 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 ===== From 883b8fde9dc1a98d488a2fca3fbb682b004fc0c7 Mon Sep 17 00:00:00 2001 From: muhammadumer-waheed Date: Mon, 28 Sep 2026 23:00:36 +0500 Subject: [PATCH 2/3] fix(bundles): verify retrieved pinned releases; normalize v-prefixed versions in URL derivation Address the Copilot review findings on this PR: - A successful request to a derived pinned-release URL does not prove 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 digest covers only the advertised release and the primitive installers trust the archive manifest, so the bundler now verifies the retrieved archive's manifest declares the pinned component (id and normalized version) before installing, and removes the unverified artifact when it does not. - URL derivation now normalizes the optional v/V prefix on both sides: a v-prefixed pin no longer produces "vv0.4.12" URLs, and a v-prefixed advertised version also rewrites the bare-form version token in asset filenames (e.g. v0.5.1 advertises -> v0.4.12/asset-0.4.12.zip). New tests fail on the pre-review code (verified by stashing the src change): 2 URL-derivation regressions and 3 negative archive-verification cases (extension version/id mismatch, preset version mismatch). E2E: a mislabeled 0.4.12 artifact (manifest declares 0.5.1) is refused with a pin-aware error; the happy path still installs the pinned release. Assisted-by: opencode (model: Qwen3.8-27B (local), autonomous) --- docs/reference/bundles.md | 2 +- src/specify_cli/bundles/primitives.py | 109 ++++++++- tests/specify_cli/bundles/test_primitives.py | 232 ++++++++++++++++++- 3 files changed, 320 insertions(+), 23 deletions(-) diff --git a/docs/reference/bundles.md b/docs/reference/bundles.md index 37f1d741f0..52e8b3e957 100644 --- a/docs/reference/bundles.md +++ b/docs/reference/bundles.md @@ -97,7 +97,7 @@ 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. When the pinned release cannot be identified from the catalog's download URL, or cannot be retrieved, the install fails with an error that names both the pinned and the advertised 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 diff --git a/src/specify_cli/bundles/primitives.py b/src/specify_cli/bundles/primitives.py index dbfa29b897..58786e53e7 100644 --- a/src/specify_cli/bundles/primitives.py +++ b/src/specify_cli/bundles/primitives.py @@ -123,6 +123,11 @@ def _pinned_release_url( 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``. """ if not isinstance(download_url, str) or not download_url: return None @@ -132,8 +137,9 @@ def _pinned_release_url( pinned = str(pinned).strip() if not advertised or not pinned: return None - if advertised == pinned: + 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 @@ -145,10 +151,13 @@ def _pinned_release_url( return None if advertised[:1] in ("v", "V"): - # The catalog spells the version v/V-prefixed; preserve that tag - # prefix form (the pin is bare semver, so the prefix is never its own - # version character). - candidates: list[tuple[str, str]] = [(advertised, advertised[:1])] + # 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"), @@ -160,7 +169,7 @@ def _pinned_release_url( changed = False for index, segment in enumerate(segments): for token, prefix in candidates: - replaced = _replace_version_token(segment, token, prefix, pinned) + replaced = _replace_version_token(segment, token, prefix, bare_pinned) if replaced is not None: segments[index] = replaced changed = True @@ -170,6 +179,69 @@ def _pinned_release_url( 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], @@ -187,9 +259,11 @@ def _download_catalog_component( 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. When no derivation is possible, or the pinned - retrieval fails, the error names the pin and the advertised version so - the failure reports the pin mismatch rather than a bare network error. + 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")): @@ -207,7 +281,7 @@ def _download_catalog_component( "before installing." ) try: - return download_by_url(derived, component.id, pinned) + 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 " @@ -216,6 +290,21 @@ def _download_catalog_component( "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) diff --git a/tests/specify_cli/bundles/test_primitives.py b/tests/specify_cli/bundles/test_primitives.py index ffcf813dda..cd9b73fe0c 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -166,6 +166,30 @@ def test_pinned_release_url_tolerates_v_prefixed_advertised_version(): ) == "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_refuses_ambiguous_or_missing_tokens(): from specify_cli.bundles.primitives import _pinned_release_url @@ -202,13 +226,15 @@ def test_pinned_release_url_refuses_ambiguous_or_missing_tokens(): assert _pinned_release_url(123, "0.5.1", "0.4.12") is None -def _extension_zip(tmp_path: Path) -> Path: +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 / "xt-source" - _write_extension_with_config(source) - zip_path = tmp_path / "xt.zip" + 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(): @@ -216,6 +242,37 @@ def _extension_zip(tmp_path: Path) -> Path: 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 ): @@ -227,7 +284,9 @@ def test_catalog_extension_pin_mismatch_fetches_pinned_release( project = tmp_path / "project" project.mkdir() - zip_path = _extension_zip(tmp_path) + # 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 = [] @@ -362,14 +421,107 @@ def _boom(*args, **kwargs): 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 - archive = tmp_path / "preset.zip" - archive.write_bytes(b"placeholder") + # 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: @@ -452,6 +604,57 @@ def test_catalog_preset_pin_mismatch_unresolvable_reports_pin( 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 @@ -622,7 +825,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 @@ -630,18 +835,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", + }, ], }, } From 23d5ccbbe73dcece3ad2e6f68b6e0ed60630c5d4 Mon Sep 17 00:00:00 2001 From: muhammadumer-waheed Date: Mon, 28 Sep 2026 23:51:37 +0500 Subject: [PATCH 3/3] fix(bundles): restrict pinned-release URL substitution to recognized version positions Address the second Copilot review round: a bounded version token inside a static path component (a repository named "tool-0.5.1") used to be rewritten as well, so deriving a pinned release from https://github.com/acme/tool-0.5.1/releases/download/v0.5.1/tool.zip moved the component's home repository to tool-0.4.12 and the valid pinned release was never fetched. Substitution is now restricted to recognized version positions: the asset filename (the final path segment) and a segment that is exactly a version token (a release tag or a versioned directory). Version-looking static components in any other position are left untouched. New coverage: a repository named tool-0.5.1 keeps its name in the derived URL (release download and archive tag shapes, plus a non-GitHub host); a segment that is exactly a version token is still treated as a version position. The test fails on the pre-fix code (verified by stashing the src change); full suite and ruff unchanged otherwise. Assisted-by: opencode (model: Qwen3.8-27B (local), autonomous) --- src/specify_cli/bundles/primitives.py | 19 +++++++++++++ tests/specify_cli/bundles/test_primitives.py | 28 ++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/src/specify_cli/bundles/primitives.py b/src/specify_cli/bundles/primitives.py index 58786e53e7..16c50b1d9f 100644 --- a/src/specify_cli/bundles/primitives.py +++ b/src/specify_cli/bundles/primitives.py @@ -128,6 +128,15 @@ def _pinned_release_url( 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 @@ -165,9 +174,19 @@ def _pinned_release_url( (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: diff --git a/tests/specify_cli/bundles/test_primitives.py b/tests/specify_cli/bundles/test_primitives.py index cd9b73fe0c..ca686aa54e 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -190,6 +190,34 @@ def test_pinned_release_url_rewrites_bare_tokens_for_v_prefixed_advertised(): ) == "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