diff --git a/docs/reference/presets.md b/docs/reference/presets.md index d51d3ace8a..139d884f2c 100644 --- a/docs/reference/presets.md +++ b/docs/reference/presets.md @@ -19,15 +19,29 @@ Searches all active catalogs for presets matching the query. Without a query, li ```bash specify preset add [] +specify preset add --version ``` -| Option | Description | -| ---------------- | -------------------------------------------------------- | -| `--dev ` | Install from a local directory (for development) | -| `--from ` | Install from a custom URL instead of the catalog | -| `--priority ` | Resolution priority (default: 10; lower = higher precedence) | - -Installs a preset from the catalog, a URL, or a local directory. Preset commands are automatically registered with supported active AI coding agent integrations. The generic integration currently delivers extension invocations but does not register preset command or skill overrides. +| Option | Description | +| --------------------- | -------------------------------------------------------------------- | +| `--dev ` | Install from a local directory (for development) | +| `--from ` | Install from a custom URL instead of the catalog | +| `--version ` | Select an exact release from the winning catalog (ID installs only) | +| `--priority ` | Resolution priority (default: 10; lower = higher precedence) | + +Installs a preset from the catalog, a URL, or a local directory. Preset commands +are automatically registered with supported active AI coding agent integrations. +The generic integration currently delivers extension invocations but does not +register preset command or skill overrides. +`--version` cannot be combined with `--from` or `--dev`. Direct URL installs +remain independent of catalog lookup. Without `--version`, installation still +selects the advertised current release (or the locally bundled preset). A +requested release absent from the winning catalog is an error; lower-priority +catalogs cannot supply it. Discovery-only catalogs cannot install any release. +Version-specific catalog installs verify the selected archive's `preset.yml` ID +and version before modifying installed presets. Historical releases require a +SHA-256 digest, which is also verified on download; a legacy current release +may omit the digest. > **Note:** All preset commands require a project already initialized with `specify init`. @@ -105,9 +119,14 @@ Presets are printed in **resolution/precedence order**: the highest-precedence p ```bash specify preset info +specify preset info --versions ``` Shows detailed information about an installed or available preset, including its templates, metadata, and tags. +`--versions` lists the advertised current version followed by historical +catalog versions, even for discovery-only entries; listing does not make +them installable. This view consults the catalog rather than the installed +preset. ## Resolve a File @@ -182,6 +201,60 @@ Catalogs are resolved in this order (first match wins): 3. **User config** — `~/.specify/preset-catalogs.yml` 4. **Built-in defaults** — official catalog + community catalog +### Versioned catalog entries + +Existing single-version entries remain valid: the top-level `version`, +`download_url`, optional `sha256`, and `requires` describe the advertised +current release. To retain older installable releases, add a `releases` +mapping keyed by version. Each historical record needs its own archive +`download_url` (HTTPS, or loopback HTTP for local development) and 64-digit +SHA-256 digest (optionally `sha256:`-prefixed, with surrounding whitespace); +other algorithm prefixes are rejected. Optional `requires` and `provides` +apply to that release instead of inheriting the current release's fields. +Other shared metadata, such as the name and description, is inherited. +Version keys must be distinct, including PEP 440-equivalent spellings, and +cannot repeat the current version. +Duplicate JSON keys are rejected before parsing can discard a release record. +Historical `requires.extensions` entries follow the preset manifest format: +extension IDs or mappings with an `id`, optional version constraint, and +optional boolean `required` flag. +For an ID lookup, catalogs are checked in priority order and stop at the +winning entry. Invalid or oversized payloads encountered before that entry +fail resolution rather than allowing a lower-priority entry to bypass its +installation policy. A malformed lower-priority source cannot block a valid +higher-priority match; searches across all sources still fail on malformed +catalogs. A non-object entry fails exact lookup for its own ID but is skipped +during all-catalog searches, which retain other valid entries. Unreachable +catalogs can still be skipped when another source is readable. +If every source fails, ID lookup (including `info --versions`) reports the +fetch error instead of claiming the preset has no catalog versions. Search +validates historical releases on each winning entry; invalid release history +from a shadowed source does not block its valid higher-priority replacement. + +```json +{ + "presets": { + "my-preset": { + "name": "My Preset", + "version": "2.0.0", + "download_url": "https://example.com/my-preset-2.0.0.zip", + "sha256": "<64 hex digits for the current archive>", + "releases": { + "1.5.0": { + "download_url": "https://example.com/my-preset-1.5.0.zip", + "sha256": "<64 hex digits for the older archive>", + "requires": {"speckit_version": ">=0.8.0"} + } + } + } + } +} +``` + +The bundled community catalog stays discovery-only and need not publish +release histories. Bundle pin resolution is a separate capability; adding +these preset records alone does not make bundle pins installable. + Example `.specify/preset-catalogs.yml`: ```yaml diff --git a/src/specify_cli/presets/_catalog.py b/src/specify_cli/presets/_catalog.py index a4768cc22d..7f99443551 100644 --- a/src/specify_cli/presets/_catalog.py +++ b/src/specify_cli/presets/_catalog.py @@ -17,10 +17,62 @@ build_safe_download_path, detect_archive_format, is_https_or_localhost_http, + is_safe_download_redirect, ) +from ._catalog_versions import available_versions, select_release from ._manifest import PresetError, PresetValidationError +class PresetCatalogValidationError(PresetError): + """A catalog supplied invalid content rather than being unreachable.""" + + +def _decode_catalog_json(raw: str | bytes, url: str) -> Any: + """Reject duplicate keys before JSON parsing discards conflicting records.""" + + def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise PresetCatalogValidationError( + f"Invalid preset catalog format from {url}: duplicate JSON key '{key}'." + ) + result[key] = value + return result + + try: + return json.loads(raw, object_pairs_hook=unique_object) + except json.JSONDecodeError as exc: + raise PresetCatalogValidationError( + f"Invalid preset catalog format from {url}: invalid JSON ({exc})" + ) from exc + except UnicodeError as exc: + raise PresetCatalogValidationError( + f"Invalid preset catalog format from {url}: invalid encoding ({exc})" + ) from exc + except RecursionError as exc: + raise PresetCatalogValidationError( + f"Invalid preset catalog format from {url}: excessive nesting ({exc})" + ) from exc + except ValueError as exc: + raise PresetCatalogValidationError( + f"Invalid preset catalog format from {url}: invalid JSON value ({exc})" + ) from exc + + +def _select_catalog_release( + pack_id: str, pack: dict[str, Any], version: str | None +) -> dict[str, Any] | None: + if "releases" in pack and pack.get("id", pack_id) != pack_id: + raise PresetCatalogValidationError( + f"Preset '{pack_id}' has an inconsistent catalog ID." + ) + try: + return select_release({**pack, "id": pack_id}, version) + except PresetError as exc: + raise PresetCatalogValidationError(str(exc)) from exc + + @dataclass class PresetCatalogEntry: """Represents a single entry in the preset catalog stack.""" @@ -172,7 +224,7 @@ def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None: PresetError: If the payload's shape is invalid. """ if not isinstance(catalog_data, dict): - raise PresetError( + raise PresetCatalogValidationError( f"Invalid preset catalog format from {url}: " "expected a JSON object" ) @@ -180,9 +232,9 @@ def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None: "schema_version" not in catalog_data or "presets" not in catalog_data ): - raise PresetError(f"Invalid preset catalog format from {url}") + raise PresetCatalogValidationError(f"Invalid preset catalog format from {url}") if not isinstance(catalog_data.get("presets"), dict): - raise PresetError( + raise PresetCatalogValidationError( f"Invalid preset catalog format from {url}: " "'presets' must be a JSON object" ) @@ -424,7 +476,9 @@ def _fetch_single_catalog(self, entry: PresetCatalogEntry, force_refresh: bool = # refreshed. if not force_refresh and self._is_url_cache_valid(entry.url): try: - cached_data = json.loads(cache_file.read_text(encoding="utf-8")) + cached_data = _decode_catalog_json( + cache_file.read_text(encoding="utf-8"), entry.url + ) self._validate_catalog_payload(cached_data, entry.url) return cached_data except (json.JSONDecodeError, OSError, UnicodeError, PresetError): @@ -451,13 +505,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != entry.url: self._validate_catalog_url(final_url) - catalog_data = json.loads( + catalog_data = _decode_catalog_json( read_response_limited( response, max_bytes=MAX_JSON_CATALOG_BYTES, - error_type=PresetError, + error_type=PresetCatalogValidationError, label=f"preset catalog {entry.url}", - ) + ), + entry.url, ) self._validate_catalog_payload(catalog_data, entry.url) @@ -496,37 +551,54 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: f"Failed to fetch preset catalog from {entry.url}: {e}" ) - def _get_merged_packs(self, force_refresh: bool = False) -> Dict[str, Dict[str, Any]]: + def _get_merged_packs( + self, force_refresh: bool = False, *, pack_id: str | None = None + ) -> Dict[str, Dict[str, Any]]: """Fetch and merge presets from all active catalogs. Higher-priority catalogs (lower priority number) win on ID conflicts. + For a requested ID, stop at the first matching catalog so malformed + lower-priority sources cannot block its winning entry. Returns: Merged dictionary of pack_id -> pack_data """ active_catalogs = self.get_active_catalogs() merged: Dict[str, Dict[str, Any]] = {} + first_fetch_error: PresetError | None = None + readable_source = False - for entry in reversed(active_catalogs): + sources = active_catalogs if pack_id is not None else reversed(active_catalogs) + for entry in sources: try: data = self._fetch_single_catalog(entry, force_refresh) - for pack_id, pack_data in data.get("presets", {}).items(): - # Per-entry guard: ``_fetch_single_catalog`` already - # validates that ``data["presets"]`` is a mapping, but it - # does not (and should not) validate every entry shape - # there — one malformed entry shouldn't poison an - # otherwise valid catalog. Skip non-mapping entries here - # so a payload like ``{"presets": {"foo": [], "bar": - # {...}}}`` still merges the valid entries without - # crashing on ``**pack_data``. Mirrors - # ``integrations/catalog.py:245``. + readable_source = True + for found_id, pack_data in data.get("presets", {}).items(): + if pack_id is not None and found_id != pack_id: + continue + # Untargeted searches skip malformed entries; an exact + # matching ID must fail instead of falling through to a + # lower-priority installable source. if not isinstance(pack_data, dict): + if pack_id is not None: + raise PresetCatalogValidationError( + f"Invalid preset catalog entry for '{pack_id}' " + f"from {entry.url}: expected a JSON object" + ) continue pack_data_with_catalog = {**pack_data, "_catalog_name": entry.name, "_install_allowed": entry.install_allowed} - merged[pack_id] = pack_data_with_catalog - except PresetError: + merged[found_id] = pack_data_with_catalog + if pack_id is not None: + return merged + except PresetCatalogValidationError: + raise + except PresetError as exc: + if first_fetch_error is None: + first_fetch_error = exc continue + if not readable_source and first_fetch_error is not None: + raise first_fetch_error return merged def is_cache_valid(self) -> bool: @@ -599,8 +671,8 @@ def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]: self.cache_metadata_file.read_text(encoding="utf-8") ) if metadata.get("catalog_url") == catalog_url: - cached_data = json.loads( - self.cache_file.read_text(encoding="utf-8") + cached_data = _decode_catalog_json( + self.cache_file.read_text(encoding="utf-8"), catalog_url ) self._validate_catalog_payload(cached_data, catalog_url) return cached_data @@ -622,13 +694,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: final_url = response.geturl() if final_url != catalog_url: self._validate_catalog_url(final_url) - catalog_data = json.loads( + catalog_data = _decode_catalog_json( read_response_limited( response, max_bytes=MAX_JSON_CATALOG_BYTES, - error_type=PresetError, + error_type=PresetCatalogValidationError, label=f"preset catalog {catalog_url}", - ) + ), + catalog_url, ) # Validate catalog structure. Reuses the same helper as @@ -691,12 +764,15 @@ def search( """ try: packs = self._get_merged_packs() + except PresetCatalogValidationError: + raise except PresetError: return [] results = [] for pack_id, pack_data in packs.items(): + _select_catalog_release(pack_id, pack_data, None) if author: author_val = pack_data.get("author", "") if not isinstance(author_val, str): @@ -735,11 +811,11 @@ def search( return results def get_pack_info( - self, pack_id: str - ) -> Optional[Dict[str, Any]]: + self, pack_id: str, version: str | None = None + ) -> dict[str, Any] | None: """Get detailed information about a specific preset. - Searches across all active catalogs (merged by priority). + Searches active catalogs in priority order, stopping at the winning ID. Args: pack_id: ID of the preset @@ -747,22 +823,33 @@ def get_pack_info( Returns: Pack metadata or None if not found """ - try: - packs = self._get_merged_packs() - except PresetError: - return None + packs = self._get_merged_packs(pack_id=pack_id) if pack_id in packs: - return {**packs[pack_id], "id": pack_id} + return _select_catalog_release(pack_id, packs[pack_id], version) return None + def get_pack_versions(self, pack_id: str) -> list[str]: + """List the versions advertised by the winning catalog entry.""" + pack = self.get_pack_info(pack_id) + return available_versions(pack) if pack is not None else [] + def download_pack( self, pack_id: str, target_dir: Optional[Path] = None ) -> Path: - """Download a preset archive from a catalog. + """Download the advertised current preset archive from a catalog.""" + pack_info = self.get_pack_info(pack_id) + if pack_info is None: + raise PresetError(f"Preset '{pack_id}' not found in catalog") + return self.download_pack_info(pack_info, target_dir) + + def download_pack_info( + self, pack_info: dict[str, Any], target_dir: Path | None = None + ) -> Path: + """Download an already-selected release without resolving its ID again. Args: - pack_id: ID of the preset to download + pack_info: Metadata returned by get_pack_info target_dir: Directory to save the archive Returns: @@ -775,11 +862,7 @@ def download_pack( from . import read_response_limited, verify_archive_sha256 - pack_info = self.get_pack_info(pack_id) - if not pack_info: - raise PresetError( - f"Preset '{pack_id}' not found in catalog" - ) + pack_id = pack_info["id"] # Bundled presets without a download URL must be installed locally if pack_info.get("bundled") and not pack_info.get("download_url"): @@ -857,17 +940,36 @@ def download_pack( staging_path: Path | None = None try: - with self._open_url(download_url, timeout=60, extra_headers=extra_headers) as response: + def _validate_redirect(old_url: str, new_url: str) -> None: + if not is_safe_download_redirect(old_url, new_url): + raise PresetError( + f"Preset download redirected to a disallowed URL: {new_url}" + ) + + with self._open_url( + download_url, + timeout=60, + extra_headers=extra_headers, + redirect_validator=_validate_redirect, + ) as response: archive_data = read_response_limited( response, error_type=PresetError, label=f"preset '{pack_id}' download", ) - final_url = ( + response_url = ( response.geturl() if hasattr(response, "geturl") else download_url ) + final_url = response_url if isinstance(response_url, str) else download_url + if not is_https_or_localhost_http(final_url) or ( + final_url != download_url + and not is_safe_download_redirect(download_url, final_url) + ): + raise PresetError( + f"Preset download redirected to a disallowed URL: {final_url}" + ) content_type = ( response.getheader("Content-Type") if hasattr(response, "getheader") diff --git a/src/specify_cli/presets/_catalog_versions.py b/src/specify_cli/presets/_catalog_versions.py new file mode 100644 index 0000000000..5328ee3bd4 --- /dev/null +++ b/src/specify_cli/presets/_catalog_versions.py @@ -0,0 +1,152 @@ +"""Validate and select releases from a preset catalog entry.""" + +from __future__ import annotations + +import re +from typing import Any + +from packaging.specifiers import InvalidSpecifier, SpecifierSet +from packaging.version import InvalidVersion, Version + +from .._download_security import is_https_or_localhost_http +from ._manifest import PresetError, PresetManifest, PresetValidationError + +_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$") +_CURRENT_FIELDS = frozenset( + {"version", "download_url", "sha256", "requires", "provides", "bundled", "releases"} +) + + +def _validated_releases(entry: dict[str, Any]) -> dict[str, dict[str, Any]]: + if "releases" not in entry: + return {} + pack_id = entry.get("id", "") + releases = entry["releases"] + if not isinstance(releases, dict): + raise PresetError(f"Preset '{pack_id}' has an invalid releases mapping.") + current = entry.get("version") + if not isinstance(current, str) or not current.strip(): + raise PresetError(f"Preset '{pack_id}' has releases but no current version.") + try: + current_version = Version(current) + except InvalidVersion: + raise PresetError( + f"Preset '{pack_id}' has invalid current version '{current}'." + ) from None + + seen = {current_version} + for release_version, record in releases.items(): + if not isinstance(release_version, str) or not release_version.strip(): + raise PresetError(f"Preset '{pack_id}' has an invalid release version key.") + try: + parsed = Version(release_version) + except InvalidVersion: + raise PresetError( + f"Preset '{pack_id}' has invalid release version '{release_version}'." + ) from None + if parsed in seen: + raise PresetError( + f"Preset '{pack_id}' repeats release version '{release_version}'." + ) + seen.add(parsed) + if not isinstance(record, dict): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' must be an object." + ) + if any( + field in record + for field in ( + "id", + "version", + "releases", + "_catalog_name", + "_install_allowed", + ) + ): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' contains reserved fields." + ) + if ( + not isinstance(record.get("download_url"), str) + or not record["download_url"].strip() + ): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' needs a download_url." + ) + if not is_https_or_localhost_http(record["download_url"]): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has an invalid download_url." + ) + digest = record.get("sha256") + if isinstance(digest, str): + digest = digest.strip() + if digest[:7].lower() == "sha256:": + digest = digest[7:].strip() + if not isinstance(digest, str) or not _SHA256.fullmatch(digest): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' needs a SHA-256 digest." + ) + for field in ("requires", "provides"): + if field in record and not isinstance(record[field], dict): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has invalid {field}." + ) + requires = record.get("requires", {}) + if "speckit_version" in requires: + specifier = requires["speckit_version"] + if not isinstance(specifier, str) or not specifier.strip(): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has invalid requires.speckit_version." + ) + try: + SpecifierSet(specifier) + except InvalidSpecifier: + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has invalid requires.speckit_version." + ) from None + if "extensions" in requires: + try: + PresetManifest._validate_requires_extensions(requires["extensions"]) + except PresetValidationError as exc: + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has {exc}" + ) from exc + if "bundled" in record and not isinstance(record["bundled"], bool): + raise PresetError( + f"Preset '{pack_id}' release '{release_version}' has invalid bundled." + ) + return releases + + +def select_release(entry: dict[str, Any], version: str | None) -> dict[str, Any] | None: + """Return current or exact historical metadata from the winning entry.""" + releases = _validated_releases(entry) + current = entry.get("version") + if version is None or version == current: + return entry + try: + requested = Version(version) + except (InvalidVersion, TypeError): + return None + if isinstance(current, str): + try: + if requested == Version(current): + return entry + except InvalidVersion: + pass # Legacy entries may use a non-PEP-440 current version. + for advertised, record in releases.items(): + if requested == Version(advertised): + common = { + key: value for key, value in entry.items() if key not in _CURRENT_FIELDS + } + return {**common, **record, "version": advertised} + return None + + +def available_versions(entry: dict[str, Any]) -> list[str]: + """Return advertised current first, then historical versions descending.""" + releases = _validated_releases(entry) + current = entry.get("version") + if not isinstance(current, str) or not current: + return [] + return [current, *sorted(releases, key=Version, reverse=True)] diff --git a/src/specify_cli/presets/_manager.py b/src/specify_cli/presets/_manager.py index 33274dfdfc..5793d8512a 100644 --- a/src/specify_cli/presets/_manager.py +++ b/src/specify_cli/presets/_manager.py @@ -559,6 +559,8 @@ def install_from_archive( force: bool = False, *, catalog_name: str | None = None, + expected_id: str | None = None, + expected_version: str | None = None, ) -> PresetManifest: """Install a preset from a supported archive. @@ -602,6 +604,27 @@ def install_from_archive( "No preset.yml found in archive" ) + if expected_id is not None or expected_version is not None: + manifest = PresetManifest(manifest_path) + if expected_id is not None and manifest.id != expected_id: + raise PresetValidationError( + f"Preset archive ID '{manifest.id}' does not match catalog ID '{expected_id}'." + ) + if expected_version is not None: + try: + matches_version = ( + pkg_version.Version(manifest.version) + == pkg_version.Version(expected_version) + ) + except (pkg_version.InvalidVersion, TypeError): + raise PresetValidationError( + f"Invalid expected catalog version: {expected_version!r}" + ) from None + if not matches_version: + raise PresetValidationError( + f"Preset archive version '{manifest.version}' does not match catalog version '{expected_version}'." + ) + return self.install_from_directory( pack_dir, speckit_version, @@ -618,6 +641,8 @@ def install_from_zip( force: bool = False, *, catalog_name: str | None = None, + expected_id: str | None = None, + expected_version: str | None = None, ) -> PresetManifest: """Backward-compatible wrapper for archive installation.""" return self.install_from_archive( @@ -626,6 +651,8 @@ def install_from_zip( priority, force=force, catalog_name=catalog_name, + expected_id=expected_id, + expected_version=expected_version, ) def remove(self, pack_id: str) -> bool: diff --git a/src/specify_cli/presets/command_add.py b/src/specify_cli/presets/command_add.py index 17bb778e9b..f9994f6e75 100644 --- a/src/specify_cli/presets/command_add.py +++ b/src/specify_cli/presets/command_add.py @@ -6,6 +6,7 @@ from pathlib import Path import typer +from packaging.version import InvalidVersion, Version from rich.markup import escape as _escape_markup from .._console import console @@ -17,6 +18,7 @@ is_safe_download_redirect, ) from . import _commands +from ._catalog_versions import select_release from ._commands import preset_app @@ -146,6 +148,9 @@ def preset_add( "--priority", help="Resolution priority (lower = higher precedence, default 10)", ), + version: str | None = typer.Option( + None, "--version", help="Install an exact version from a catalog" + ), ): """Install a preset.""" from .. import _locate_bundled_preset, _require_specify_project, get_speckit_version @@ -159,6 +164,20 @@ def preset_add( project_root = _require_specify_project() _commands._validate_priority(priority) + # Direct callers of the command function receive Typer's OptionInfo default. + if not isinstance(version, str): + version = None + if version is not None and ( + not version.strip() + or dev is not None + or from_url is not None + or not preset_id + ): + console.print( + "[red]Error:[/red] --version requires a catalog preset ID " + "(without --dev or --from)." + ) + raise typer.Exit(1) manager = PresetManager(project_root) speckit_version = get_speckit_version() @@ -295,7 +314,7 @@ def _validate_download_redirect(old_url, new_url): elif preset_id: # Try bundled preset first, then catalog - bundled_path = _locate_bundled_preset(preset_id) + bundled_path = _locate_bundled_preset(preset_id) if version is None else None if bundled_path: console.print(f"Installing bundled preset [cyan]{preset_id}[/cyan]...") manifest = manager.install_from_directory( @@ -314,14 +333,50 @@ def _validate_download_redirect(old_url, new_url): ) raise typer.Exit(1) + if version is not None: + if not pack_info.get("_install_allowed", True): + console.print( + f"[red]Error:[/red] Preset '{_escape_markup(preset_id)}' " + "is from a discovery-only catalog (install not allowed)." + ) + raise typer.Exit(1) + selected = select_release(pack_info, version) + if selected is None: + console.print( + f"[red]Error:[/red] Preset '{_escape_markup(preset_id)}' " + f"has no catalog release for version {_escape_markup(version)}." + ) + raise typer.Exit(1) + pack_info = selected + # Bundled presets should have been caught above; if we reach # here the bundled files are missing from the installation. if pack_info.get("bundled") and not pack_info.get("download_url"): + packaged = _locate_bundled_preset(preset_id) + if version is not None and packaged is not None: + from . import PresetManifest + + packaged_manifest = PresetManifest(packaged / "preset.yml") + try: + matches = Version(packaged_manifest.version) == Version( + pack_info["version"] + ) + except InvalidVersion: + matches = False + if packaged_manifest.id == preset_id and matches: + manifest = manager.install_from_directory( + packaged, speckit_version, priority + ) + console.print( + f"[green]✓[/green] Preset '{manifest.name}' v{manifest.version} installed (priority {priority})" + ) + _commands._warn_unmet_extension_dependencies(manager, manifest) + return from ..extensions import REINSTALL_COMMAND console.print( - f"[red]Error:[/red] Preset '{preset_id}' is bundled with spec-kit " - f"but could not be found in the installed package." + f"[red]Error:[/red] Preset '{_escape_markup(preset_id)}' is bundled with spec-kit " + "but the requested version could not be found in the installed package." ) console.print( "\nThis usually means the spec-kit installation is incomplete or corrupted." @@ -345,12 +400,21 @@ def _validate_download_redirect(old_url, new_url): ) try: - archive_path = catalog.download_pack(preset_id) + archive_path = ( + catalog.download_pack_info(pack_info) + if version is not None + else catalog.download_pack(preset_id) + ) manifest = manager.install_from_zip( archive_path, speckit_version, priority, catalog_name=pack_info.get("_catalog_name"), + **( + {"expected_id": preset_id, "expected_version": pack_info["version"]} + if version is not None + else {} + ), ) console.print( f"[green]✓[/green] Preset '{manifest.name}' v{manifest.version} installed (priority {priority})" diff --git a/src/specify_cli/presets/command_info.py b/src/specify_cli/presets/command_info.py index a3920a6605..0a4b908237 100644 --- a/src/specify_cli/presets/command_info.py +++ b/src/specify_cli/presets/command_info.py @@ -6,12 +6,15 @@ from rich.markup import escape as _escape_markup from .._console import console +from ._catalog import PresetCatalogValidationError +from ._catalog_versions import available_versions from ._commands import preset_app @preset_app.command("info") def preset_info( preset_id: str = typer.Argument(..., help="Preset ID to get info about"), + versions: bool = typer.Option(False, "--versions", help="List catalog versions"), ): """Show detailed information about a preset.""" from .. import _require_specify_project @@ -20,6 +23,27 @@ def preset_info( project_root = _require_specify_project() safe_preset_id = _escape_markup(str(preset_id)) + if versions is True: + catalog = PresetCatalog(project_root) + try: + pack_info = catalog.get_pack_info(preset_id) + available = available_versions(pack_info) if pack_info else [] + except PresetError as exc: + console.print(f"[red]Error:[/red] {_escape_markup(str(exc))}") + raise typer.Exit(1) from exc + if not available: + console.print( + f"[red]Error:[/red] No catalog versions found for {safe_preset_id}." + ) + raise typer.Exit(1) + console.print(f"Catalog versions for {safe_preset_id}:") + for index, item in enumerate(available): + console.print( + f" {_escape_markup(item)}{' (current)' if index == 0 else ''}" + ) + if not pack_info.get("_install_allowed", True): + console.print("[yellow]Discovery only; catalog installation is disabled.[/yellow]") + return # Check if installed locally first manager = PresetManager(project_root) local_pack = manager.get_pack(preset_id) @@ -63,6 +87,9 @@ def preset_info( catalog = PresetCatalog(project_root) try: pack_info = catalog.get_pack_info(preset_id) + except PresetCatalogValidationError as exc: + console.print(f"[red]Error:[/red] {_escape_markup(str(exc))}") + raise typer.Exit(1) from exc except PresetError: pack_info = None diff --git a/tests/specify_cli/presets/test_catalog.py b/tests/specify_cli/presets/test_catalog.py index 2ffbddfed9..6d50da9ca1 100644 --- a/tests/specify_cli/presets/test_catalog.py +++ b/tests/specify_cli/presets/test_catalog.py @@ -969,11 +969,13 @@ def _pack_zip_and_response(self): resp.__exit__.return_value = False return zip_bytes, resp - def test_fetch_single_catalog_rejects_oversized_body_without_cache( - self, project_dir, monkeypatch + @pytest.mark.parametrize("legacy", [False, True], ids=["stacked", "legacy"]) + def test_catalog_rejects_oversized_body_without_cache( + self, project_dir, monkeypatch, legacy ): """Catalog bounds are enforced at the preset call site.""" import specify_cli.presets as preset_module + from specify_cli.presets._catalog import PresetCatalogValidationError from unittest.mock import patch catalog = PresetCatalog(project_dir) @@ -996,8 +998,14 @@ def test_fetch_single_catalog_rejects_oversized_body_without_cache( ) with patch.object(catalog, "_open_url", return_value=response): - with pytest.raises(PresetError, match="exceeds maximum size"): - catalog._fetch_single_catalog(entry, force_refresh=True) + with pytest.raises( + PresetCatalogValidationError, match="exceeds maximum size" + ): + if legacy: + with patch.object(catalog, "get_catalog_url", return_value=entry.url): + catalog.fetch_catalog(force_refresh=True) + else: + catalog._fetch_single_catalog(entry, force_refresh=True) assert not catalog.cache_dir.exists() or not any(catalog.cache_dir.iterdir()) diff --git a/tests/specify_cli/presets/test_catalog_versions.py b/tests/specify_cli/presets/test_catalog_versions.py new file mode 100644 index 0000000000..4080de89a9 --- /dev/null +++ b/tests/specify_cli/presets/test_catalog_versions.py @@ -0,0 +1,981 @@ +"""Exact-release preset catalog lookup, downloads, and CLI regressions.""" + +from __future__ import annotations + +import hashlib +import io +import json +import sys +import zipfile +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import yaml +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli.presets import ( + PresetCatalog, + PresetCatalogEntry, + PresetError, + PresetManager, + PresetValidationError, +) +from specify_cli.presets._catalog import PresetCatalogValidationError, _decode_catalog_json + +CURRENT_URL = "https://example.com/preset-current.zip" +OLD_URL = "https://example.com/preset-old.zip" + + +def _archive(pack_id: str = "sample", version: str = "1.0.0") -> bytes: + manifest = { + "schema_version": "1.0", + "preset": { + "id": pack_id, + "name": "Sample", + "version": version, + "description": "Sample preset", + }, + "requires": {"speckit_version": ">=0.1.0"}, + "provides": { + "templates": [ + { + "type": "template", + "name": "spec-template", + "file": "templates/spec-template.md", + } + ] + }, + } + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("preset.yml", yaml.safe_dump(manifest)) + archive.writestr("templates/spec-template.md", "# Sample\n") + return buffer.getvalue() + + +def _entry(old_bytes: bytes | None = None) -> dict: + old_bytes = old_bytes if old_bytes is not None else _archive() + return { + "id": "sample", + "name": "Sample", + "version": "2.0.0", + "download_url": CURRENT_URL, + "sha256": "a" * 64, + "requires": {"speckit_version": ">=2"}, + "provides": {"templates": 2}, + "releases": { + "1.0.0": { + "download_url": OLD_URL, + "sha256": hashlib.sha256(old_bytes).hexdigest(), + "requires": {"speckit_version": ">=0.1.0"}, + "provides": {"templates": 0}, + } + }, + } + + +def _response(data: bytes, url: str) -> MagicMock: + response = MagicMock() + response.read.side_effect = io.BytesIO(data).read + response.geturl.return_value = url + response.getheader.return_value = "application/zip" + response.__enter__.return_value = response + return response + + +def _duplicate_release_json() -> bytes: + entry = _entry() + payload = json.dumps({"schema_version": "1.0", "presets": {"sample": entry}}) + record = f'"1.0.0": {json.dumps(entry["releases"]["1.0.0"])}' + conflicting = {**entry["releases"]["1.0.0"], "download_url": CURRENT_URL} + assert record in payload + return payload.replace( + record, f'{record}, "1.0.0": {json.dumps(conflicting)}', 1 + ).encode() + + +def _oversized_integer_json() -> bytes: + digit_limit = sys.get_int_max_str_digits() + if digit_limit == 0: + pytest.skip("Python's JSON integer digit limit is disabled") + return ( + b'{"schema_version":"1.0","presets":{"sample":' + + b"9" * (digit_limit + 1) + b"}}" + ) + + +@pytest.mark.parametrize("legacy", [False, True], ids=["stack", "single-catalog"]) +def test_duplicate_release_key_rejected_from_network(project_dir, legacy): + catalog = PresetCatalog(project_dir) + url = catalog.DEFAULT_CATALOG_URL + entry = PresetCatalogEntry(url, "default", 1, True) + with ( + patch.object(catalog, "get_catalog_url", return_value=url), + patch.object( + catalog, "_open_url", return_value=_response(_duplicate_release_json(), url) + ), + pytest.raises(PresetError, match="duplicate.*1.0.0"), + ): + if legacy: + catalog.fetch_catalog(force_refresh=True) + else: + catalog._fetch_single_catalog(entry, force_refresh=True) + assert not catalog.cache_file.exists() + + +@pytest.mark.parametrize("legacy", [False, True], ids=["stack", "single-catalog"]) +def test_duplicate_release_key_in_cache_refetches(project_dir, legacy): + catalog = PresetCatalog(project_dir) + url = catalog.DEFAULT_CATALOG_URL + entry = PresetCatalogEntry(url, "default", 1, True) + catalog.cache_dir.mkdir(parents=True) + catalog.cache_file.write_bytes(_duplicate_release_json()) + catalog.cache_metadata_file.write_text( + json.dumps({ + "cached_at": datetime.now(timezone.utc).isoformat(), + "catalog_url": url, + }) + ) + valid = {"schema_version": "1.0", "presets": {"sample": _entry()}} + with ( + patch.object(catalog, "get_catalog_url", return_value=url), + patch.object( + catalog, "_open_url", return_value=_response(json.dumps(valid).encode(), url) + ) as opened, + ): + result = catalog.fetch_catalog() if legacy else catalog._fetch_single_catalog(entry) + assert result == valid + opened.assert_called_once() + assert json.loads(catalog.cache_file.read_text()) == valid + + +def test_current_and_exact_selection_keep_current_fields(project_dir): + catalog = PresetCatalog(project_dir) + entry = _entry() + with patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}): + current = catalog.get_pack_info("sample") + old = catalog.get_pack_info("sample", "1.0") + assert catalog.get_pack_info("sample", "0.4.12") is None + assert catalog.get_pack_versions("sample") == ["2.0.0", "1.0.0"] + assert current["version"] == "2.0.0" + assert current["download_url"] == CURRENT_URL + assert old["version"] == "1.0.0" + assert old["download_url"] == OLD_URL + assert old["requires"] == {"speckit_version": ">=0.1.0"} + assert old["provides"] == {"templates": 0} + assert old["sha256"] != current["sha256"] + assert "releases" not in old + + +def test_single_release_entry_remains_compatible(project_dir): + catalog = PresetCatalog(project_dir) + entry = {"name": "Legacy", "version": "1.0.0", "download_url": OLD_URL} + with patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}): + assert catalog.get_pack_info("sample")["version"] == "1.0.0" + assert catalog.get_pack_info("sample", "1.0")["version"] == "1.0.0" + assert catalog.get_pack_info("sample", "2.0") is None + assert catalog.get_pack_versions("sample") == ["1.0.0"] + + +@pytest.mark.parametrize( + "change, error", + [ + ({"releases": []}, "releases mapping"), + ({"version": None}, "current version"), + ({"version": "garbage"}, "current version"), + ( + {"releases": {"2.0": {"download_url": OLD_URL, "sha256": "f" * 64}}}, + "repeats", + ), + ( + { + "releases": { + "1.0": {"download_url": OLD_URL, "sha256": "f" * 64}, + "1.0.0": {"download_url": OLD_URL, "sha256": "f" * 64}, + } + }, + "repeats", + ), + ({"releases": {"oops": {}}}, "release version"), + ({"releases": {"1.0": []}}, "must be an object"), + ({"releases": {"1.0": {"sha256": "f" * 64}}}, "download_url"), + ( + { + "releases": { + "1.0": { + "download_url": "http://evil.test/a.zip", + "sha256": "f" * 64, + } + } + }, + "download_url", + ), + ( + {"releases": {"1.0": {"download_url": OLD_URL, "sha256": "broken"}}}, + "SHA-256", + ), + ( + {"releases": {"1.0": {"download_url": OLD_URL, "sha256": "md5:" + "f" * 64}}}, + "SHA-256", + ), + ( + { + "releases": { + "1.0": { + "download_url": OLD_URL, + "sha256": "f" * 64, + "version": "1.0", + } + } + }, + "reserved", + ), + ( + { + "releases": { + "1.0": {"download_url": OLD_URL, "sha256": "f" * 64, "requires": []} + } + }, + "requires", + ), + ( + { + "releases": { + "1.0": { + "download_url": OLD_URL, + "sha256": "f" * 64, + "requires": {"speckit_version": 2}, + } + } + }, + "requires.speckit_version", + ), + ( + { + "releases": { + "1.0": { + "download_url": OLD_URL, + "sha256": "f" * 64, + "requires": {"speckit_version": "not a specifier"}, + } + } + }, + "requires.speckit_version", + ), + ({"id": "other"}, "inconsistent"), + ], +) +def test_malformed_history_rejected_even_for_current(project_dir, change, error): + entry = {**_entry(), **change} + catalog = PresetCatalog(project_dir) + with ( + patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}), + pytest.raises(PresetError, match=error), + ): + catalog.get_pack_info("sample") + + +@pytest.mark.parametrize( + "dependencies", + [ + [123], + [{}], + [{"id": "dep", "version": 2}], + [{"id": "dep", "required": 0}], + ["bad id"], + ], +) +def test_historical_release_rejects_malformed_extension_dependencies( + project_dir, dependencies +): + entry = _entry() + entry["releases"]["1.0.0"]["requires"]["extensions"] = dependencies + catalog = PresetCatalog(project_dir) + with ( + patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}), + pytest.raises(PresetError, match="requires.extensions"), + ): + catalog.get_pack_info("sample") + + +def test_historical_release_accepts_manifest_extension_dependencies(project_dir): + dependencies = [ + "plain-ext", + {"id": "other-ext", "version": ">=1.2", "required": False}, + ] + entry = _entry() + entry["releases"]["1.0.0"]["requires"]["extensions"] = dependencies + catalog = PresetCatalog(project_dir) + with patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}): + selected = catalog.get_pack_info("sample", "1.0.0") + assert selected["requires"]["extensions"] == dependencies + + +def test_malformed_history_info_reports_validation_error(project_dir): + entry = {**_entry(), "releases": []} + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch.object(PresetCatalog, "_get_merged_packs", return_value={"sample": entry}), + ): + with pytest.raises(PresetCatalogValidationError, match="releases mapping"): + PresetCatalog(project_dir).get_pack_info("sample") + result = CliRunner().invoke(app, ["preset", "info", "sample"]) + assert result.exit_code == 1 + assert "invalid releases mapping" in result.output + assert "not found" not in result.output + + +@pytest.mark.parametrize("digest_format", ["plain", "prefix", "uppercase-prefix"]) +def test_historical_digest_accepts_download_supported_forms(project_dir, digest_format): + old_bytes = _archive() + digest = hashlib.sha256(old_bytes).hexdigest() + declared = { + "plain": f" {digest} ", + "prefix": f" sha256:{digest} ", + "uppercase-prefix": f" SHA256: {digest} ", + }[digest_format] + entry = _entry(old_bytes) + entry["releases"]["1.0.0"]["sha256"] = declared + catalog = PresetCatalog(project_dir) + with patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}): + selected = catalog.get_pack_info("sample", "1.0.0") + with patch.object(catalog, "_open_url", return_value=_response(old_bytes, OLD_URL)): + downloaded = catalog.download_pack_info(selected, target_dir=project_dir) + assert downloaded.read_bytes() == old_bytes + + +def test_winning_source_does_not_fall_back_to_lower_release(project_dir): + catalog = PresetCatalog(project_dir) + sources = [ + PresetCatalogEntry("https://example.com/high.json", "high", 1, True), + PresetCatalogEntry("https://example.com/low.json", "low", 2, True), + ] + older = {**_entry(), "version": "3.0.0"} + higher = {"version": "2.0.0", "download_url": CURRENT_URL} + + def fetch(source, _refresh): + return {"presets": {"sample": higher if source.name == "high" else older}} + + with ( + patch.object(catalog, "get_active_catalogs", return_value=sources), + patch.object(catalog, "_fetch_single_catalog", side_effect=fetch), + ): + assert catalog.get_pack_info("sample")["_catalog_name"] == "high" + assert catalog.get_pack_info("sample", "1.0.0") is None + + +@pytest.mark.parametrize( + "bad_payload, error", + [ + (_duplicate_release_json, "duplicate JSON key"), + (lambda: b'{"schema_version": "1.0", "presets": []}', "Invalid preset catalog format"), + (lambda: b'{"schema_version":', "invalid JSON"), + (lambda: b'{"schema_version":"1.0","presets":' + b"\xff" + b"}", "invalid encoding"), + (_oversized_integer_json, "invalid JSON value"), + ( + lambda: b'{"schema_version":"1.0","presets":{"sample":' + + b"[" * 20000 + b"0" + b"]" * 20000 + b"}}", + "excessive nesting", + ), + ], +) +def test_invalid_discovery_catalog_cannot_delegate_install( + project_dir, bad_payload, error +): + high_url = "https://example.com/discovery.json" + low_url = "https://example.com/trusted.json" + sources = [ + PresetCatalogEntry(high_url, "discovery", 1, False), + PresetCatalogEntry(low_url, "trusted", 2, True), + ] + old_bytes = _archive() + lower = json.dumps({ + "schema_version": "1.0", + "presets": {"sample": _entry(old_bytes)}, + }).encode() + invalid = bad_payload() + opened: list[str] = [] + + def open_url(_self, url, **_kwargs): + opened.append(url) + data = { + high_url: invalid, + low_url: lower, + OLD_URL: old_bytes, + } + return _response(data[url], url) + + original_loads = json.loads + + def parse_json(raw, **kwargs): + # Decoder nesting limits vary across supported Python versions. + if error == "excessive nesting" and raw == invalid: + raise RecursionError("too deep") + return original_loads(raw, **kwargs) + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_open_url", open_url), + patch("specify_cli.presets._catalog.json.loads", side_effect=parse_json), + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="1.0.0"), + ): + with pytest.raises(PresetError, match=error): + PresetCatalog(project_dir).get_pack_info("sample", "1.0.0") + result = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0.0"] + ) + info = CliRunner().invoke(app, ["preset", "info", "sample"]) + search = CliRunner().invoke(app, ["preset", "search", "sample"]) + assert result.exit_code == 1, result.output + assert error in result.output + assert info.exit_code == 1 and error in info.output + assert search.exit_code == 1 and error in search.output + assert OLD_URL not in opened + assert PresetManager(project_dir).get_pack("sample") is None + + +def test_recursion_error_is_invalid_catalog_content(): + with ( + patch( + "specify_cli.presets._catalog.json.loads", + side_effect=RecursionError("too deep"), + ), + pytest.raises(PresetCatalogValidationError, match="excessive nesting"), + ): + _decode_catalog_json(b"{}", "https://example.com/catalog.json") + + +def test_malformed_matching_discovery_entry_prevents_lower_install(project_dir): + high_url = "https://example.com/discovery.json" + low_url = "https://example.com/trusted.json" + sources = [ + PresetCatalogEntry(high_url, "discovery", 1, False), + PresetCatalogEntry(low_url, "trusted", 2, True), + ] + old_bytes = _archive() + upper = b'{"schema_version":"1.0","presets":{"sample":[]}}' + lower = json.dumps({ + "schema_version": "1.0", + "presets": {"sample": _entry(old_bytes)}, + }).encode() + opened: list[str] = [] + + def open_url(_self, url, **_kwargs): + opened.append(url) + return _response({ + high_url: upper, + low_url: lower, + OLD_URL: old_bytes, + }[url], url) + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_open_url", open_url), + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="1.0.0"), + ): + catalog = PresetCatalog(project_dir) + with pytest.raises(PresetCatalogValidationError, match="expected a JSON object"): + catalog.get_pack_info("sample", "1.0.0") + refused = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0.0"] + ) + info = CliRunner().invoke(app, ["preset", "info", "sample"]) + results = catalog.search("sample") + assert refused.exit_code == 1 and "expected a JSON object" in refused.output + assert info.exit_code == 1 and "expected a JSON object" in info.output + assert results[0]["_catalog_name"] == "trusted" + assert OLD_URL not in opened + assert PresetManager(project_dir).get_pack("sample") is None + + +@pytest.mark.parametrize("malformed_winner", [True, False], ids=["winner", "shadowed"]) +def test_search_validates_only_winning_release_history(project_dir, malformed_winner): + catalog = PresetCatalog(project_dir) + sources = [ + PresetCatalogEntry("https://example.com/official.json", "official", 1, True), + PresetCatalogEntry("https://example.com/community.json", "community", 2, False), + ] + invalid = {**_entry(), "releases": []} + valid = _entry() + + def fetch(source, _refresh): + entry = ( + invalid if (source.name == "official") == malformed_winner else valid + ) + return {"presets": {"sample": entry}} + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_fetch_single_catalog", side_effect=fetch), + patch.object(Path, "cwd", return_value=project_dir), + ): + if malformed_winner: + with pytest.raises(PresetCatalogValidationError, match="releases mapping"): + catalog.search("sample") + cli_result = CliRunner().invoke(app, ["preset", "search", "sample"]) + assert cli_result.exit_code == 1 + assert "releases mapping" in cli_result.output + else: + matches = catalog.search("sample") + assert len(matches) == 1 + assert matches[0]["_catalog_name"] == "official" + + +def test_valid_higher_priority_id_ignores_invalid_lower_catalog(project_dir): + high_url = "https://example.com/official.json" + low_url = "https://example.com/community.json" + sources = [ + PresetCatalogEntry(high_url, "official", 1, True), + PresetCatalogEntry(low_url, "community", 2, False), + ] + archive = _archive() + higher = json.dumps({ + "schema_version": "1.0", + "presets": {"sample": _entry(archive)}, + }).encode() + opened: list[str] = [] + + def open_url(_self, url, **_kwargs): + opened.append(url) + return _response({ + high_url: higher, + low_url: b'{"schema_version":', + OLD_URL: archive, + }[url], url) + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_open_url", open_url), + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="1.0.0"), + ): + catalog = PresetCatalog(project_dir) + selected = catalog.get_pack_info("sample", "1.0.0") + listed = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + installed = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0.0"] + ) + assert low_url not in opened + with pytest.raises(PresetCatalogValidationError, match="invalid JSON"): + catalog.search("sample") + assert selected["version"] == "1.0.0" + assert selected["_catalog_name"] == "official" + assert listed.exit_code == 0, listed.output + assert "1.0.0" in listed.output + assert installed.exit_code == 0, installed.output + assert low_url in opened + assert opened.count(OLD_URL) == 1 + assert PresetManager(project_dir).get_pack("sample").version == "1.0.0" + + +def test_oversized_discovery_catalog_cannot_delegate_install(project_dir): + high_url = "https://example.com/discovery.json" + low_url = "https://example.com/trusted.json" + sources = [ + PresetCatalogEntry(high_url, "discovery", 1, False), + PresetCatalogEntry(low_url, "trusted", 2, True), + ] + lower = json.dumps({ + "schema_version": "1.0", "presets": {"sample": _entry()} + }).encode() + oversized = json.dumps({ + "schema_version": "1.0", + "presets": {"sample": _entry()}, + "padding": "x" * len(lower), + }).encode() + assert len(oversized) > len(lower) + opened: list[str] = [] + + def open_url(_self, url, **_kwargs): + opened.append(url) + return _response({high_url: oversized, low_url: lower}[url], url) + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_open_url", open_url), + patch("specify_cli.presets.MAX_JSON_CATALOG_BYTES", len(lower)), + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="1.0.0"), + ): + with pytest.raises(PresetError, match="exceeds maximum size"): + PresetCatalog(project_dir).get_pack_info("sample", "1.0.0") + result = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0.0"] + ) + assert result.exit_code == 1, result.output + assert "exceeds maximum size" in result.output + assert OLD_URL not in opened + assert PresetManager(project_dir).get_pack("sample") is None + + +def test_unreachable_high_priority_catalog_still_uses_lower_source(project_dir): + catalog = PresetCatalog(project_dir) + sources = [ + PresetCatalogEntry("https://example.com/unavailable.json", "high", 1, False), + PresetCatalogEntry("https://example.com/trusted.json", "low", 2, True), + ] + + def fetch(source, _refresh): + if source.name == "high": + raise PresetError("Failed to fetch preset catalog: offline") + return {"presets": {"sample": _entry()}} + + with ( + patch.object(catalog, "get_active_catalogs", return_value=sources), + patch.object(catalog, "_fetch_single_catalog", side_effect=fetch), + ): + selected = catalog.get_pack_info("sample", "1.0.0") + assert selected["_catalog_name"] == "low" + assert selected["_install_allowed"] is True + + +def test_versions_report_all_source_outage_instead_of_missing_preset(project_dir): + sources = [ + PresetCatalogEntry("https://example.com/high.json", "high", 1, True), + PresetCatalogEntry("https://example.com/low.json", "low", 2, True), + ] + + def fetch(source, _refresh): + raise PresetError(f"Failed to fetch preset catalog from {source.url}: offline") + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_fetch_single_catalog", side_effect=fetch), + patch.object(Path, "cwd", return_value=project_dir), + ): + with pytest.raises(PresetError, match="high.json: offline"): + PresetCatalog(project_dir).get_pack_info("sample") + result = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + + assert result.exit_code == 1, result.output + assert "high.json:" in result.output and "offline" in result.output + assert "No catalog versions found" not in result.output + + +def test_versions_report_missing_preset_when_catalog_is_readable(project_dir): + sources = [ + PresetCatalogEntry("https://example.com/high.json", "high", 1, True), + PresetCatalogEntry("https://example.com/low.json", "low", 2, True), + ] + + def fetch(source, _refresh): + if source.name == "high": + raise PresetError(f"Failed to fetch preset catalog from {source.url}: offline") + return {"presets": {"another-preset": _entry()}} + + with ( + patch.object(PresetCatalog, "get_active_catalogs", return_value=sources), + patch.object(PresetCatalog, "_fetch_single_catalog", side_effect=fetch), + patch.object(Path, "cwd", return_value=project_dir), + ): + assert PresetCatalog(project_dir).get_pack_info("sample") is None + result = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + + assert result.exit_code == 1, result.output + assert "No catalog versions found for sample" in result.output + assert "offline" not in result.output + + +def test_discovery_only_winner_does_not_delegate_exact_release(project_dir): + catalog = PresetCatalog(project_dir) + sources = [ + PresetCatalogEntry("https://example.com/high.json", "discovery", 1, False), + PresetCatalogEntry("https://example.com/low.json", "trusted", 2, True), + ] + + def fetch(_source, _refresh): + return {"presets": {"sample": _entry()}} + + with ( + patch.object(catalog, "get_active_catalogs", return_value=sources), + patch.object(catalog, "_fetch_single_catalog", side_effect=fetch), + patch.object(catalog, "_open_url") as open_url, + ): + selected = catalog.get_pack_info("sample", "1.0") + assert selected["_catalog_name"] == "discovery" + with pytest.raises(PresetError, match="does not allow installation"): + catalog.download_pack_info(selected, project_dir) + open_url.assert_not_called() + + +def test_historical_release_does_not_inherit_current_requirements(project_dir): + catalog = PresetCatalog(project_dir) + entry = _entry() + del entry["releases"]["1.0.0"]["requires"] + del entry["releases"]["1.0.0"]["provides"] + with patch.object(catalog, "_get_merged_packs", return_value={"sample": entry}): + selected = catalog.get_pack_info("sample", "1.0.0") + assert "requires" not in selected + assert "provides" not in selected + + +def test_selected_download_uses_old_url_and_digest_without_lookup(project_dir): + old_bytes = _archive() + catalog = PresetCatalog(project_dir) + info = {**_entry(old_bytes), "_install_allowed": True, "_catalog_name": "trusted"} + with patch.object(catalog, "_get_merged_packs", return_value={"sample": info}): + selected = catalog.get_pack_info("sample", "1.0") + with ( + patch.object(catalog, "get_pack_info", side_effect=AssertionError("re-lookup")), + patch.object( + catalog, "_open_url", return_value=_response(old_bytes, OLD_URL) + ) as opened, + ): + saved = catalog.download_pack_info(selected, target_dir=project_dir) + assert saved.read_bytes() == old_bytes + assert opened.call_args.args[0] == OLD_URL + + +def test_selected_download_rejects_discovery_digest_and_redirect(project_dir): + old_bytes = _archive() + selected = { + "id": "sample", + "version": "1.0.0", + "download_url": OLD_URL, + "sha256": hashlib.sha256(old_bytes).hexdigest(), + "_install_allowed": False, + "_catalog_name": "community", + } + catalog = PresetCatalog(project_dir) + with patch.object(catalog, "_open_url") as open_url: + with pytest.raises(PresetError, match="does not allow installation"): + catalog.download_pack_info(selected, project_dir) + open_url.assert_not_called() + selected["_install_allowed"] = True + with ( + patch.object(catalog, "_open_url", return_value=_response(b"wrong", OLD_URL)), + pytest.raises(PresetError, match="[Ii]ntegrity"), + ): + catalog.download_pack_info(selected, project_dir) + with ( + patch.object( + catalog, + "_open_url", + return_value=_response(old_bytes, "http://evil.test/a.zip"), + ), + pytest.raises(PresetError, match="disallowed URL"), + ): + catalog.download_pack_info(selected, project_dir) + assert not list(project_dir.glob("sample-*.zip")) + + +def test_selected_download_rejects_unsafe_intermediate_redirect(project_dir): + selected = { + "id": "sample", + "version": "1.0.0", + "download_url": OLD_URL, + "sha256": "f" * 64, + "_install_allowed": True, + } + catalog = PresetCatalog(project_dir) + + def redirect(_url, **kwargs): + kwargs["redirect_validator"](OLD_URL, "http://evil.test/transit") + return _response(_archive(), OLD_URL) + + with ( + patch.object(catalog, "_open_url", side_effect=redirect), + pytest.raises(PresetError, match="disallowed URL"), + ): + catalog.download_pack_info(selected, project_dir) + assert not list(project_dir.glob("sample-*.zip")) + + +@pytest.mark.parametrize("historical", [False, True], ids=["current", "historical"]) +def test_direct_localhost_https_download_needs_no_redirect(project_dir, historical): + archive = _archive() + url = "https://dev.localhost/preset.zip" + selected = { + "id": "sample", + "version": "1.0.0", + "download_url": url, + "sha256": hashlib.sha256(archive).hexdigest(), + "_install_allowed": True, + } + catalog = PresetCatalog(project_dir) + with ( + patch.object(catalog, "get_pack_info", return_value=selected), + patch.object(catalog, "_open_url", return_value=_response(archive, url)), + ): + saved = ( + catalog.download_pack_info(selected, project_dir) + if historical + else catalog.download_pack("sample", project_dir) + ) + assert saved.read_bytes() == archive + + +def test_download_rejects_actual_redirect_to_localhost_https(project_dir): + selected = { + "id": "sample", + "version": "1.0.0", + "download_url": OLD_URL, + "_install_allowed": True, + } + catalog = PresetCatalog(project_dir) + with ( + patch.object( + catalog, "_open_url", + return_value=_response(_archive(), "https://dev.localhost/preset.zip"), + ), + pytest.raises(PresetError, match="disallowed URL"), + ): + catalog.download_pack_info(selected, project_dir) + + +@pytest.mark.parametrize( + "bad_id,bad_version", [("other", "1.0.0"), ("sample", "2.0.0")] +) +def test_archive_identity_checked_before_install( + project_dir, tmp_path, bad_id, bad_version +): + archive_path = tmp_path / "sample.zip" + archive_path.write_bytes(_archive(bad_id, bad_version)) + manager = PresetManager(project_dir) + with pytest.raises(PresetValidationError, match="does not match catalog"): + manager.install_from_zip( + archive_path, "1.0.0", expected_id="sample", expected_version="1.0.0" + ) + assert not manager.registry.is_installed(bad_id) + + +def test_archive_mismatch_does_not_replace_installed_preset(project_dir, tmp_path): + good_path = tmp_path / "good.zip" + bad_path = tmp_path / "bad.zip" + good_path.write_bytes(_archive("sample", "1.0.0")) + bad_path.write_bytes(_archive("sample", "2.0.0")) + manager = PresetManager(project_dir) + manager.install_from_zip(good_path, "1.0.0") + with pytest.raises(PresetValidationError, match="does not match catalog"): + manager.install_from_zip( + bad_path, + "1.0.0", + force=True, + expected_id="sample", + expected_version="1.0.0", + ) + assert manager.get_pack("sample").version == "1.0.0" + + +def test_cli_installs_exact_archive_and_lists_versions(project_dir): + old_bytes = _archive() + catalog_entry = _entry(old_bytes) + urls: list[str] = [] + + def open_url(_self, url, **_kwargs): + urls.append(url) + return _response(old_bytes, url) + + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="1.0.0"), + patch.object( + PresetCatalog, + "_get_merged_packs", + return_value={ + "sample": { + **catalog_entry, + "_catalog_name": "trusted", + "_install_allowed": True, + } + }, + ), + patch.object(PresetCatalog, "_open_url", open_url), + ): + listed = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + installed = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0"] + ) + + assert listed.exit_code == 0, listed.output + assert "2.0.0 (current)" in listed.output and "1.0.0" in listed.output + assert installed.exit_code == 0, installed.output + assert urls == [OLD_URL] + assert PresetManager(project_dir).get_pack("sample").version == "1.0.0" + + +def test_cli_versions_use_winning_entry_snapshot(project_dir): + first = {**_entry(), "_install_allowed": False} + second = {"id": "sample", "version": "3.0.0"} + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch.object(PresetCatalog, "get_pack_info", side_effect=[first, second]) as lookup, + ): + result = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + assert result.exit_code == 0, result.output + assert "2.0.0 (current)" in result.output and "1.0.0" in result.output + assert "3.0.0" not in result.output + assert "Discovery only" in result.output + lookup.assert_called_once_with("sample") + + +def test_cli_rejects_missing_release_and_discovery_without_download(project_dir): + entry = _entry() + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch.object( + PresetCatalog, + "_get_merged_packs", + return_value={ + "sample": { + **entry, + "_catalog_name": "discovery", + "_install_allowed": False, + } + }, + ), + patch.object(PresetCatalog, "_open_url") as open_url, + ): + info = CliRunner().invoke(app, ["preset", "info", "sample", "--versions"]) + refused = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "1.0.0"] + ) + assert info.exit_code == 0 and "Discovery only" in info.output + assert refused.exit_code == 1 and "discovery-only" in refused.output + open_url.assert_not_called() + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch.object( + PresetCatalog, + "_get_merged_packs", + return_value={ + "sample": { + **entry, + "_catalog_name": "trusted", + "_install_allowed": True, + } + }, + ), + patch.object(PresetCatalog, "_open_url") as open_url, + ): + absent = CliRunner().invoke( + app, ["preset", "add", "sample", "--version", "9.0"] + ) + assert absent.exit_code == 1 and "no catalog release" in absent.output + open_url.assert_not_called() + + +@pytest.mark.parametrize( + "args", + [ + ["preset", "add", "sample", "--from", OLD_URL, "--version", "1.0"], + ["preset", "add", "sample", "--from", "", "--version", "1.0"], + ["preset", "add", "sample", "--dev", ".", "--version", "1.0"], + ["preset", "add", "sample", "--dev", "", "--version", "1.0"], + ["preset", "add", "sample", "--version", ""], + ], +) +def test_cli_rejects_version_with_non_catalog_source(project_dir, args): + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch.object(PresetCatalog, "get_pack_info") as lookup, + ): + result = CliRunner().invoke(app, args) + assert result.exit_code == 1 + assert "--version requires a catalog" in result.output + lookup.assert_not_called() diff --git a/tests/specify_cli/presets/test_command_add.py b/tests/specify_cli/presets/test_command_add.py index ae46f2bb9d..c892e4476c 100644 --- a/tests/specify_cli/presets/test_command_add.py +++ b/tests/specify_cli/presets/test_command_add.py @@ -5,7 +5,7 @@ import zipfile from pathlib import Path from types import SimpleNamespace -from unittest.mock import ANY, MagicMock +from unittest.mock import ANY, MagicMock, patch import pytest import yaml @@ -221,6 +221,80 @@ def test_bundled_preset_add_via_cli(self, project_dir): assert "Lean Workflow" in result.output assert "installed" in result.output.lower() + def test_bundled_exact_version_installs_packaged_preset(self, project_dir): + from typer.testing import CliRunner + + from specify_cli import app + + entry = { + "id": "lean", + "name": "Lean Workflow", + "version": "1.0.0", + "bundled": True, + "_install_allowed": True, + } + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli.get_speckit_version", return_value="0.6.0"), + patch.object(PresetCatalog, "get_pack_info", return_value=entry), + patch.object(PresetCatalog, "download_pack_info") as download, + patch.object(PresetManager, "install_from_zip") as install_zip, + ): + result = CliRunner().invoke( + app, ["preset", "add", "lean", "--version", "1.0"] + ) + + assert result.exit_code == 0, result.output + assert "Lean Workflow" in result.output + assert PresetManager(project_dir).get_pack("lean").version == "1.0.0" + download.assert_not_called() + install_zip.assert_not_called() + + @pytest.mark.parametrize("packaged", ["missing", "wrong-id", "wrong-version"]) + def test_bundled_exact_version_rejects_wrong_package( + self, project_dir, pack_dir, packaged + ): + from typer.testing import CliRunner + + from specify_cli import app + + entry = { + "id": "lean", + "name": "Lean Workflow", + "version": "1.0.0", + "bundled": True, + "_install_allowed": True, + } + if packaged == "missing": + packaged_path = None + else: + packaged_path = pack_dir + if packaged == "wrong-version": + manifest_path = pack_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) + manifest["preset"].update(id="lean", version="2.0.0") + manifest_path.write_text(yaml.safe_dump(manifest), encoding="utf-8") + + with ( + patch.object(Path, "cwd", return_value=project_dir), + patch("specify_cli._locate_bundled_preset", return_value=packaged_path), + patch("specify_cli.get_speckit_version", return_value="0.6.0"), + patch.object(PresetCatalog, "get_pack_info", return_value=entry), + patch.object(PresetCatalog, "download_pack_info") as download, + patch.object(PresetManager, "install_from_directory") as install_dir, + ): + result = CliRunner().invoke( + app, ["preset", "add", "lean", "--version", "1.0.0"] + ) + + assert result.exit_code == 1, result.output + assert "requested version could not be found" in " ".join( + result.output.split() + ) + assert PresetManager(project_dir).get_pack("lean") is None + download.assert_not_called() + install_dir.assert_not_called() + def test_preset_add_catalog_forwards_catalog_name(self, project_dir, monkeypatch): """Catalog installs pass resolved provenance into the manager boundary.""" from specify_cli.presets._commands import preset_add diff --git a/tests/specify_cli/workflows/test_catalog_versions.py b/tests/specify_cli/workflows/test_catalog_versions.py index f9828b3278..6a1900f7be 100644 --- a/tests/specify_cli/workflows/test_catalog_versions.py +++ b/tests/specify_cli/workflows/test_catalog_versions.py @@ -5,6 +5,7 @@ import hashlib import io import zipfile +from unittest.mock import patch import pytest from typer.testing import CliRunner @@ -49,10 +50,21 @@ def _archive(version: str, workflow_id: str = "history-wf", requires=None) -> by document["requires"] = requires output = io.BytesIO() with zipfile.ZipFile(output, "w") as archive: - archive.writestr("workflow.yml", yaml.safe_dump(document)) + archive.writestr( + zipfile.ZipInfo("workflow.yml", date_time=(2020, 1, 1, 0, 0, 0)), + yaml.safe_dump(document), + ) return output.getvalue() +def test_archive_fixture_is_stable_across_clock_ticks(): + with patch("zipfile.time.localtime", return_value=(2020, 1, 1, 0, 0, 0)): + first = _archive("2.0.0") + with patch("zipfile.time.localtime", return_value=(2020, 1, 1, 0, 0, 2)): + second = _archive("2.0.0") + assert first == second + + def _entry() -> dict: old = _archive("1.0.0", requires={"integrations": ["copilot"]}) current = _archive("2.0.0")