harden: secure extension and preset archive downloads

src/specify_cli/bundler/commands_impl/catalog_config.py

@@ -153,6 +153,8 @@ def add_source(
         # keeps that ValueError inside the guard instead of leaking a raw
         # traceback past the CLI's `except BundlerError`. Reuse the value below.
         hostname = parsed.hostname
+        # Accessing ``port`` performs urllib's syntax/range validation.
+        _ = parsed.port
     except ValueError as exc:
         raise BundlerError(f"Invalid catalog url: '{url}'.") from exc
     if not (parsed.scheme or parsed.path):

src/specify_cli/bundler/models/catalog.py

@@ -139,6 +139,7 @@ class CatalogEntry:
     license: str
     download_url: str
     requires_speckit_version: str
+    sha256: str | None = None
     provides: dict[str, int] = field(default_factory=dict)
     repository: str | None = None
     tags: tuple[str, ...] = ()
@@ -181,6 +182,11 @@ def from_dict(cls, data: Any) -> "CatalogEntry":
             license=str(data.get("license", "")).strip(),
             download_url=str(data.get("download_url", "")).strip(),
             requires_speckit_version=str(requires.get("speckit_version", "")).strip(),
+            sha256=(
+                None
+                if data.get("sha256") is None
+                else str(data["sha256"]).strip()
+            ),
             provides=dict(provides_raw),
             repository=(str(data["repository"]) if data.get("repository") else None),
             tags=_parse_tags(data.get("tags"), entry_id),
@@ -193,6 +199,7 @@ def with_provenance(self, source: CatalogSource) -> "CatalogEntry":
             description=self.description, author=self.author, license=self.license,
             download_url=self.download_url,
             requires_speckit_version=self.requires_speckit_version,
+            sha256=self.sha256,
             provides=self.provides, repository=self.repository, tags=self.tags,
             verified=self.verified, source_id=source.id,
             source_policy=source.install_policy,

src/specify_cli/bundler/services/adapters.py

@@ -16,6 +16,7 @@
 from urllib.request import url2pathname
 
 from ..._assets import _locate_core_pack, _repo_root
+from ..._download_security import MAX_JSON_CATALOG_BYTES, read_response_limited
 from .. import BundlerError
 from ..lib.yamlio import loads_json
 from ..models.catalog import CatalogSource
@@ -76,6 +77,8 @@ def _validate_remote_url(source_id: str, url: str) -> None:
     try:
         parsed = urlparse(url)
         hostname = parsed.hostname
+        # Accessing ``port`` performs urllib's syntax/range validation.
+        _ = parsed.port
     except ValueError:
         raise BundlerError(
             f"Catalog '{source_id}' URL is malformed: {url}"
@@ -117,7 +120,15 @@ def make_catalog_fetcher(*, allow_network: bool = True):
 
     def fetch(source: CatalogSource) -> dict:
         url = source.url
-        parsed = urlparse(url)
+        try:
+            parsed = urlparse(url)
+            # Keep malformed authorities and ports inside the BundlerError
+            # contract even when a config file was edited by hand.
+            _ = parsed.port
+        except ValueError:
+            raise BundlerError(
+                f"Catalog {source.id!r} URL is malformed: {url!r}"
+            ) from None
         scheme = parsed.scheme.lower()
 
         if scheme == "builtin":
@@ -180,7 +191,12 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
         ) as response:
             final_url = response.geturl()
             _validate_remote_url(source_id, final_url)
-            raw = response.read().decode("utf-8")
+            raw = read_response_limited(
+                response,
+                max_bytes=MAX_JSON_CATALOG_BYTES,
+                error_type=BundlerError,
+                label=f"bundle catalog '{source_id}'",
+            ).decode("utf-8")
     except BundlerError:
         raise
     except Exception as exc:  # noqa: BLE001

src/specify_cli/commands/bundle/__init__.py

@@ -14,6 +14,7 @@
 import typer
 
 from ..._console import console, err_console
+from ..._download_security import MAX_DOWNLOAD_BYTES, read_response_limited
 from ...bundler import BundlerError
 from ...bundler.lib.project import (
     active_integration,
@@ -337,6 +338,10 @@ def bundle_install(
         local_manifest = _local_manifest_source(bundle_id)
         if local_manifest is not None:
             manifest = local_manifest
+            _validate_manifest_structure(
+                manifest,
+                source=f"Local bundle source {bundle_id!r}",
+            )
         else:
             stack = _build_stack(project_root or Path.cwd(), offline=offline)
             resolved = stack.resolve(bundle_id)
@@ -350,6 +355,16 @@ def bundle_install(
 
         if project_root is None:
             init_integration = _resolve_init_integration(integration, manifest)
+            # Resolve all hard compatibility gates before ``specify init``.
+            # Otherwise an incompatible but structurally valid bundle would
+            # initialize a project and only then fail its version/integration
+            # checks, leaving state behind after a failed install.
+            resolve_install_plan(
+                manifest,
+                speckit_version=_speckit_version(),
+                active_integration=init_integration,
+                integration_explicit=True,
+            )
             console.print(
                 f"[cyan]No Spec Kit project here; initializing with integration "
                 f"'{init_integration}'…[/cyan]"
@@ -711,17 +726,24 @@ def _local_manifest_source(arg: str):
 
     if candidate.suffix == ".zip":
         import io
-        import zipfile
 
         import yaml as _yaml
 
-        with zipfile.ZipFile(candidate) as archive:
+        from ..._download_security import open_zip_bounded, read_zip_member_limited
+
+        with open_zip_bounded(candidate, error_type=BundlerError) as archive:
             try:
-                raw = archive.read("bundle.yml")
+                archive.getinfo("bundle.yml")
             except KeyError as exc:
                 raise BundlerError(
                     f"Artifact '{candidate}' does not contain a bundle.yml."
                 ) from exc
+            raw = read_zip_member_limited(
+                archive,
+                "bundle.yml",
+                error_type=BundlerError,
+                label="bundle manifest",
+            )
         data = _yaml.safe_load(io.BytesIO(raw))
         return BundleManifest.from_dict(data)
 
@@ -805,7 +827,13 @@ def _download_manifest(resolved, *, offline: bool):
             f"Network access disabled; cannot download bundle '{resolved.entry.id}' "
             f"from {url}."
         )
-    return _download_remote_manifest(resolved.entry.id, url)
+    manifest = _download_remote_manifest(
+        resolved.entry.id,
+        url,
+        expected_sha256=getattr(resolved.entry, "sha256", None),
+    )
+    _validate_catalog_manifest(resolved.entry, manifest)
+    return manifest
 
 
 def _require_https(label: str, url: str) -> None:
@@ -817,6 +845,8 @@ def _require_https(label: str, url: str) -> None:
     try:
         parsed = urlparse(url)
         hostname = parsed.hostname
+        # Accessing ``port`` performs urllib's syntax/range validation.
+        _ = parsed.port
     except ValueError:
         raise BundlerError(
             f"Refusing to download {label}: URL is malformed: {url}"
@@ -830,7 +860,12 @@ def _require_https(label: str, url: str) -> None:
         raise BundlerError(f"Refusing to download {label} from URL with no host: {url}")
 
 
-def _download_remote_manifest(entry_id: str, url: str):
+def _download_remote_manifest(
+    entry_id: str,
+    url: str,
+    *,
+    expected_sha256: str | None = None,
+):
     """Fetch a remote bundle artifact over HTTPS and extract its manifest."""
     import io
     import tempfile
@@ -842,6 +877,7 @@ def _download_remote_manifest(entry_id: str, url: str):
     from ...authentication.http import github_provider_hosts, open_url
     from ..._github_http import resolve_github_release_asset_api_url
     from ...bundler.models.manifest import BundleManifest
+    from ...shared_infra import verify_archive_sha256
 
     def _validate_redirect(old_url: str, new_url: str) -> None:
         _require_https(f"bundle '{entry_id}'", new_url)
@@ -879,7 +915,18 @@ def _validate_redirect(old_url: str, new_url: str) -> None:
             extra_headers=extra_headers,
         ) as resp:
             _require_https(f"bundle '{entry_id}'", resp.geturl())
-            raw = resp.read()
+            raw = read_response_limited(
+                resp,
+                max_bytes=MAX_DOWNLOAD_BYTES,
+                error_type=BundlerError,
+                label=f"bundle '{entry_id}' download",
+            )
+        verify_archive_sha256(
+            raw,
+            expected_sha256,
+            entry_id,
+            BundlerError,
+        )
     except BundlerError:
         raise
     except Exception as exc:  # noqa: BLE001
@@ -940,6 +987,38 @@ def _validate_redirect(old_url: str, new_url: str) -> None:
         ) from exc
 
 
+def _validate_manifest_structure(manifest, *, source: str) -> None:
+    """Reject a malformed manifest before any project mutation can occur."""
+    from ...bundler.services.validator import validate_manifest
+
+    report = validate_manifest(manifest)
+    if report.ok:
+        return
+    raise BundlerError(
+        f"{source} contains an invalid bundle manifest:\n  - "
+        + "\n  - ".join(report.errors)
+    )
+
+
+def _validate_catalog_manifest(entry, manifest) -> None:
+    """Bind a downloaded manifest to the catalog identity that selected it."""
+    if manifest.bundle.id != entry.id:
+        raise BundlerError(
+            f"Downloaded bundle id mismatch: catalog entry {entry.id!r} points to "
+            f"a manifest for {manifest.bundle.id!r}."
+        )
+    if manifest.bundle.version != entry.version:
+        raise BundlerError(
+            f"Downloaded bundle version mismatch for {entry.id!r}: catalog declares "
+            f"{entry.version!r}, but the manifest declares "
+            f"{manifest.bundle.version!r}."
+        )
+    _validate_manifest_structure(
+        manifest,
+        source=f"Downloaded bundle {entry.id!r}",
+    )
+
+
 def register(app: typer.Typer) -> None:
     """Attach the bundle command group to the root Typer app."""
     app.add_typer(bundle_app, name="bundle")

src/specify_cli/extensions/__init__.py

@@ -17,7 +17,6 @@
 import shutil
 import stat
 import tempfile
-import zipfile
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -29,6 +28,13 @@
 from packaging.specifiers import InvalidSpecifier, SpecifierSet
 
 from .._assets import _locate_core_pack, _repo_root
+from .._download_security import (
+    MAX_JSON_CATALOG_BYTES,
+    build_safe_download_path,
+    is_https_or_localhost_http,
+    read_response_limited,
+    safe_extract_zip,
+)
 from .._init_options import is_ai_skills_enabled
 from .._invocation_style import is_dollar_skills_agent, is_slash_skills_agent
 from .._utils import dump_frontmatter, relative_extension_path_violation, version_satisfies
@@ -2053,21 +2059,7 @@ def install_from_zip(
         with tempfile.TemporaryDirectory() as tmpdir:
             temp_path = Path(tmpdir)
 
-            # Extract ZIP safely (prevent Zip Slip attack)
-            with zipfile.ZipFile(zip_path, "r") as zf:
-                # Validate all paths first before extracting anything
-                temp_path_resolved = temp_path.resolve()
-                for member in zf.namelist():
-                    member_path = (temp_path / member).resolve()
-                    # Use is_relative_to for safe path containment check
-                    try:
-                        member_path.relative_to(temp_path_resolved)
-                    except ValueError:
-                        raise ValidationError(
-                            f"Unsafe path in ZIP archive: {member} (potential path traversal)"
-                        )
-                # Only extract after all paths are validated
-                zf.extractall(temp_path)
+            safe_extract_zip(zip_path, temp_path, error_type=ValidationError)
 
             # Find extension directory (may be nested)
             extension_dir = temp_path
@@ -2862,7 +2854,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 final_url = response.geturl()
                 if final_url != entry.url:
                     self._validate_catalog_url(final_url)
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        max_bytes=MAX_JSON_CATALOG_BYTES,
+                        error_type=ExtensionError,
+                        label=f"extension catalog {entry.url}",
+                    )
+                )
 
             self._validate_catalog_payload(catalog_data, entry.url)
 
@@ -3050,7 +3049,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 final_url = response.geturl()
                 if final_url != catalog_url:
                     self._validate_catalog_url(final_url)
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        max_bytes=MAX_JSON_CATALOG_BYTES,
+                        error_type=ExtensionError,
+                        label=f"extension catalog {catalog_url}",
+                    )
+                )
 
             # Validate catalog structure. Reuses the same helper as
             # ``_fetch_single_catalog`` so all three branches (root type,
@@ -3195,6 +3201,10 @@ def download_extension(
         download_url = ext_info.get("download_url")
         if not download_url:
             raise ExtensionError(f"Extension '{extension_id}' has no download URL")
+        if not isinstance(download_url, str):
+            raise ExtensionError(
+                f"Extension download URL is malformed: {download_url}"
+            )
 
         # Validate download URL requires HTTPS (prevent man-in-the-middle attacks)
         from urllib.parse import urlparse
@@ -3208,24 +3218,33 @@ def download_extension(
         try:
             parsed = urlparse(download_url)
             hostname = parsed.hostname
+            parsed.port
         except ValueError:
             raise ExtensionError(
                 f"Extension download URL is malformed: {download_url}"
             ) from None
-        is_localhost = hostname in ("localhost", "127.0.0.1", "::1")
-        if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
+        if not hostname:
+            raise ExtensionError(
+                f"Extension download URL is malformed: {download_url}"
+            )
+        if not is_https_or_localhost_http(download_url):
             raise ExtensionError(
                 f"Extension download URL must use HTTPS: {download_url}"
             )
 
         # Determine target path
         if target_dir is None:
             target_dir = self.cache_dir / "downloads"
-        target_dir.mkdir(parents=True, exist_ok=True)
-
+        target_dir = Path(target_dir)
         version = ext_info.get("version", "unknown")
-        zip_filename = f"{extension_id}-{version}.zip"
-        zip_path = target_dir / zip_filename
+        zip_path = build_safe_download_path(
+            target_dir,
+            extension_id,
+            version,
+            error_type=ExtensionError,
+            label="extension",
+        )
+        target_dir.mkdir(parents=True, exist_ok=True)
 
         extra_headers = None
         resolved_download_url = self._resolve_github_release_asset_api_url(download_url)
@@ -3238,7 +3257,11 @@ def download_extension(
             with self._open_url(
                 download_url, timeout=60, extra_headers=extra_headers
             ) as response:
-                zip_data = response.read()
+                zip_data = read_response_limited(
+                    response,
+                    error_type=ExtensionError,
+                    label=f"extension '{extension_id}' download",
+                )
 
             verify_archive_sha256(
                 zip_data, ext_info.get("sha256"), extension_id, ExtensionError