harden: secure extension and preset archive downloads

src/specify_cli/_utils.py

@@ -12,6 +12,7 @@
 from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import Any
 from ._console import console
+from ._download_security import normalize_zip_member_name
 
 CLAUDE_LOCAL_PATH = Path.home() / ".claude" / "local" / "claude"
 CLAUDE_NPM_LOCAL_PATH = Path.home() / ".claude" / "local" / "node_modules" / ".bin" / "claude"
@@ -27,19 +28,22 @@ def relative_extension_path_violation(value: Any) -> str | None:
     ``None`` when it is an acceptable relative path within the extension
     directory.
 
-    Policy: the value must be a non-empty string with no leading/trailing
-    whitespace, no absolute/anchored form, and no ``..`` traversal. The value is
+    Policy: the value must be a non-empty, portable file path with no
+    leading/trailing whitespace, absolute/anchored form, ``..`` traversal,
+    platform-reserved component, or directory-only suffix. The value is
     evaluated under both POSIX and Windows path semantics because a native
     ``Path`` is OS-dependent (a ``PurePosixPath`` on POSIX does not interpret
-    Windows drive/UNC forms, and ``C:foo`` is anchored but not ``is_absolute()``
-    yet resolves against the CWD on its drive). Rejecting any non-empty anchor
-    covers POSIX-absolute (``/abs``), Windows drive-relative (``C:foo``), Windows
-    absolute (``C:\\foo``), and UNC/rooted forms.
+    Windows drive/UNC forms, and ``C:foo`` is anchored but not
+    ``is_absolute()`` yet resolves against the CWD on its drive). Rejecting any
+    non-empty anchor covers POSIX-absolute (``/abs``), Windows drive-relative
+    (``C:foo``), Windows absolute (``C:\\foo``), and UNC/rooted forms.
     """
     if not isinstance(value, str) or not value:
         return "must be a non-empty string"
     if value.strip() != value:
         return "must not have leading or trailing whitespace"
+    if "\\" in value:
+        return "must use forward slashes as path separators"
     posix_path = PurePosixPath(value)
     win_path = PureWindowsPath(value)
     if (
@@ -52,6 +56,15 @@ def relative_extension_path_violation(value: Any) -> str | None:
             "must be a relative path within the extension directory "
             "(no absolute paths, drive letters, or '..' segments)"
         )
+    if value.endswith(("/", "\\")):
+        return "must name a file or command, not a directory"
+    try:
+        normalize_zip_member_name(value)
+    except ValueError:
+        return (
+            "must use portable path components "
+            "(no reserved names or platform-invalid characters)"
+        )
     return None
 
 

src/specify_cli/agents.py

@@ -675,6 +675,23 @@ def register_commands(
             cmd_name = cmd_info["name"]
             aliases = cmd_info.get("aliases", [])
             cmd_file = cmd_info["file"]
+            name_reason = relative_extension_path_violation(cmd_name)
+            if name_reason:
+                raise ValueError(
+                    f"Invalid command name {cmd_name!r}: {name_reason}"
+                )
+            if aliases is None:
+                aliases = []
+            if not isinstance(aliases, list):
+                raise ValueError(
+                    f"Aliases for command {cmd_name!r} must be a list"
+                )
+            for alias in aliases:
+                alias_reason = relative_extension_path_violation(alias)
+                if alias_reason:
+                    raise ValueError(
+                        f"Invalid command alias {alias!r}: {alias_reason}"
+                    )
 
             # Guard against path traversal using the single shared policy in
             # relative_extension_path_violation(), so the runtime guard stays
@@ -957,10 +974,16 @@ def write_copilot_prompt(project_root: Path, cmd_name: str) -> None:
             project_root: Path to project root
             cmd_name: Command name (e.g. 'speckit.my-ext.example')
         """
+        name_reason = relative_extension_path_violation(cmd_name)
+        if name_reason:
+            raise ValueError(
+                f"Invalid Copilot prompt name {cmd_name!r}: {name_reason}"
+            )
         prompts_dir = project_root / ".github" / "prompts"
         prompts_dir.mkdir(parents=True, exist_ok=True)
         prompt_file = prompts_dir / f"{cmd_name}.prompt.md"
         CommandRegistrar._ensure_inside(prompt_file, prompts_dir)
+        prompt_file.parent.mkdir(parents=True, exist_ok=True)
         prompt_file.write_text(f"---\nagent: {cmd_name}\n---\n", encoding="utf-8")
 
     @staticmethod

src/specify_cli/bundler/commands_impl/catalog_config.py

@@ -152,6 +152,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

@@ -144,6 +144,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, ...] = ()
@@ -186,6 +187,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),
@@ -198,6 +204,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")