fix(extensions): wire up list --available catalog query + harden add --from path traversal

src/specify_cli/extensions/__init__.py

@@ -60,6 +60,28 @@
     }
 )
 EXTENSION_COMMAND_NAME_PATTERN = re.compile(r"^speckit\.([a-z0-9-]+)\.([a-z0-9-]+)$")
+EXTENSION_ID_PATTERN = re.compile(r"^[a-z0-9-]+$")
+
+# Characters allowed verbatim in a catalog-provided version when it is used to
+# build a download filename. Anything else (path separators, "..", control
+# chars, whitespace) is collapsed to "-" so an untrusted version cannot inject
+# a nested path or an unwritable filename — we sanitize rather than rely on the
+# post-hoc Path.resolve() containment check as the primary guard.
+_UNSAFE_VERSION_TOKEN_PATTERN = re.compile(r"[^A-Za-z0-9._-]+")
+
+
+def _safe_version_token(version: Any, fallback: str = "unknown") -> str:
+    """Reduce a catalog ``version`` to a filename-safe token.
+
+    Non-strings (JSON null, numbers) and blank values degrade to ``fallback``.
+    Path separators and traversal tokens are stripped so the result can never
+    span directories or escape the download target.
+    """
+    if not isinstance(version, str):
+        return fallback
+    token = _UNSAFE_VERSION_TOKEN_PATTERN.sub("-", version).strip("-.")
+    return token or fallback
+
 
 VALID_EFFECTS = frozenset({"read-only", "read-write"})
 
@@ -306,7 +328,7 @@ def _validate(self):
                 )
 
         # Validate extension ID format
-        if not re.match(r"^[a-z0-9-]+$", ext["id"]):
+        if not EXTENSION_ID_PATTERN.fullmatch(ext["id"]):
             raise ValidationError(
                 f"Invalid extension ID '{ext['id']}': "
                 "must be lowercase alphanumeric with hyphens only"
@@ -3463,6 +3485,54 @@ def __init__(self, project_root: Path):
         self.cache_file = self.cache_dir / "catalog.json"
         self.cache_metadata_file = self.cache_dir / "catalog-metadata.json"
 
+    def _ensure_default_download_cache_dir(self) -> Path:
+        """Create the default download cache without following symlink parents."""
+        root = self.project_root.resolve()
+        target_dir = self.cache_dir / "downloads"
+        try:
+            rel = target_dir.relative_to(self.project_root)
+        except ValueError:
+            raise ExtensionError("Default extension download cache escapes project root") from None
+
+        current = self.project_root
+        for part in rel.parts:
+            current = current / part
+            try:
+                label = current.relative_to(self.project_root).as_posix()
+            except ValueError:
+                label = str(current)
+
+            if current.is_symlink():
+                raise ExtensionError(
+                    f"Refusing to use symlinked extension download cache path: {label}"
+                )
+            if current.exists():
+                if not current.is_dir():
+                    raise ExtensionError(
+                        f"Extension download cache path is not a directory: {label}"
+                    )
+                try:
+                    current.resolve().relative_to(root)
+                except (OSError, ValueError):
+                    raise ExtensionError(
+                        f"Extension download cache path escapes project root: {label}"
+                    ) from None
+                continue
+
+            current.mkdir()
+            if current.is_symlink():
+                raise ExtensionError(
+                    f"Refusing to use symlinked extension download cache path: {label}"
+                )
+            try:
+                current.resolve().relative_to(root)
+            except (OSError, ValueError):
+                raise ExtensionError(
+                    f"Extension download cache path escapes project root: {label}"
+                ) from None
+
+        return target_dir
+
     def _make_request(self, url: str):
         """Build a urllib Request, adding auth headers when a provider matches.
 
@@ -3999,42 +4069,38 @@ def search(
 
         results = []
 
+        def _search_text(value: Any) -> str:
+            return value if isinstance(value, str) else ""
+
+        def _search_tags(value: Any) -> List[str]:
+            if not isinstance(value, list):
+                return []
+            return [tag for tag in value if isinstance(tag, str)]
+
         for ext_data in all_extensions:
             ext_id = ext_data["id"]
+            tags = _search_tags(ext_data.get("tags", []))
 
             # Apply filters
             if verified_only and not ext_data.get("verified", False):
                 continue
 
-            if author:
-                author_val = ext_data.get("author", "")
-                if not isinstance(author_val, str):
-                    author_val = str(author_val) if author_val is not None else ""
-                if author_val.lower() != author.lower():
-                    continue
+            if author and _search_text(ext_data.get("author")).lower() != author.lower():
+                continue
 
-            if tag:
-                raw_tags = ext_data.get("tags", [])
-                tags_list = raw_tags if isinstance(raw_tags, list) else []
-                if tag.lower() not in [
-                    t.lower() for t in tags_list if isinstance(t, str)
-                ]:
-                    continue
+            if tag and tag.lower() not in [t.lower() for t in tags]:
+                continue
 
             if query:
                 # Search in name, description, and tags
                 query_lower = query.lower()
-                raw_tags = ext_data.get("tags", [])
-                tags_list = raw_tags if isinstance(raw_tags, list) else []
-                name_val = ext_data.get("name", "")
-                desc_val = ext_data.get("description", "")
                 searchable_text = " ".join(
                     [
-                        str(name_val) if name_val else "",
-                        str(desc_val) if desc_val else "",
-                        ext_id,
+                        _search_text(ext_data.get("name")),
+                        _search_text(ext_data.get("description")),
+                        _search_text(ext_id),
                     ]
-                    + [t for t in tags_list if isinstance(t, str)]
+                    + tags
                 ).lower()
 
                 if query_lower not in searchable_text:
@@ -4079,6 +4145,14 @@ def download_extension(
         """
         import urllib.error
 
+        if not isinstance(extension_id, str) or not EXTENSION_ID_PATTERN.fullmatch(
+            extension_id
+        ):
+            raise ExtensionError(
+                f"Invalid extension ID '{extension_id}': "
+                "must be lowercase alphanumeric with hyphens only"
+            )
+
         # Get extension info from catalog
         ext_info = self.get_extension_info(extension_id)
         if not ext_info:
@@ -4128,19 +4202,11 @@ def download_extension(
 
         # Determine target path
         if target_dir is None:
-            target_dir = self.cache_dir / "downloads"
-        target_dir = Path(target_dir)
-        version = ext_info.get("version", "unknown")
-        declared_format = archive_format_from_name(download_url)
-        build_safe_download_path(
-            target_dir,
-            extension_id,
-            version,
-            error_type=ExtensionError,
-            label="extension",
-            suffix=archive_suffix(declared_format or "tar.gz"),
-        )
-        target_dir.mkdir(parents=True, exist_ok=True)
+            target_dir = self._ensure_default_download_cache_dir()
+        else:
+            target_dir = Path(target_dir)
+            target_dir.mkdir(parents=True, exist_ok=True)
+        version = _safe_version_token(ext_info.get("version"))
 
         original_download_url = download_url
         extra_headers = None