fix(integrations): validate cached catalog shape before returning it

src/specify_cli/integrations/catalog.py

@@ -40,6 +40,25 @@ class IntegrationDescriptorError(Exception):
     """Raised when an integration.yml descriptor is invalid."""
 
 
+def _catalog_shape_error(payload: Any) -> Optional[str]:
+    """Return a human-readable reason if *payload* is not a valid integration
+    catalog document, else ``None``.
+
+    Shared by the fresh-fetch and cache-read paths so both enforce the same
+    format contract: a JSON object carrying ``schema_version`` and a mapping
+    ``integrations``. Keeping a single validator prevents the two paths from
+    drifting (e.g. a cache that skips the ``schema_version`` check and lets an
+    older/poisoned payload bypass validation).
+    """
+    if not isinstance(payload, dict):
+        return "expected a JSON object"
+    if "schema_version" not in payload or "integrations" not in payload:
+        return "missing required 'schema_version' or 'integrations' key"
+    if not isinstance(payload.get("integrations"), dict):
+        return "'integrations' must be a JSON object"
+    return None
+
+
 # ---------------------------------------------------------------------------
 # IntegrationCatalogEntry
 # ---------------------------------------------------------------------------
@@ -153,7 +172,18 @@ def _fetch_single_catalog(
                     cached_at = cached_at.replace(tzinfo=timezone.utc)
                 age = (datetime.now(timezone.utc) - cached_at).total_seconds()
                 if age < self.CACHE_DURATION:
-                    return json.loads(cache_file.read_text(encoding="utf-8"))
+                    cached = json.loads(cache_file.read_text(encoding="utf-8"))
+                    # A poisoned/older-format cache must clear the SAME shape
+                    # contract as a fresh fetch (via the shared validator) —
+                    # otherwise a payload like [], {"integrations": []}, or one
+                    # missing "schema_version" is returned and later crashes on
+                    # .items()/.get() or silently bypasses the format contract.
+                    # The ValueError is caught just below, which drops the
+                    # corrupt cache and refetches from source.
+                    shape_error = _catalog_shape_error(cached)
+                    if shape_error is not None:
+                        raise ValueError(f"cached catalog has invalid shape: {shape_error}")
+                    return cached
             except (json.JSONDecodeError, ValueError, KeyError, TypeError, AttributeError, OSError, UnicodeError):
                 # Cache is invalid or stale metadata; delete and refetch from source.
                 try:
@@ -172,20 +202,10 @@ def _fetch_single_catalog(
                     self._validate_catalog_url(final_url)
                 catalog_data = json.loads(resp.read())
 
-            if not isinstance(catalog_data, dict):
-                raise IntegrationCatalogError(
-                    f"Invalid catalog format from {entry.url}: expected a JSON object"
-                )
-            if (
-                "schema_version" not in catalog_data
-                or "integrations" not in catalog_data
-            ):
-                raise IntegrationCatalogError(
-                    f"Invalid catalog format from {entry.url}"
-                )
-            if not isinstance(catalog_data.get("integrations"), dict):
+            shape_error = _catalog_shape_error(catalog_data)
+            if shape_error is not None:
                 raise IntegrationCatalogError(
-                    f"Invalid catalog format from {entry.url}: 'integrations' must be a JSON object"
+                    f"Invalid catalog format from {entry.url}: {shape_error}"
                 )
 
             try:

tests/integrations/test_integration_catalog.py

@@ -13,9 +13,34 @@
     IntegrationDescriptor,
     IntegrationDescriptorError,
     IntegrationValidationError,
+    _catalog_shape_error,
 )
 
 
+class TestCatalogShapeValidator:
+    """The shared shape validator used by BOTH the fresh-fetch and cache-read
+    paths, so a poisoned/older cache can't bypass the format contract the fresh
+    fetch enforces (dict + 'schema_version' + dict 'integrations')."""
+
+    def test_valid_payload_returns_none(self):
+        assert _catalog_shape_error({"schema_version": "1.0", "integrations": {}}) is None
+
+    def test_missing_schema_version_is_rejected(self):
+        # The exact bypass the two paths used to disagree on: a dict with a dict
+        # 'integrations' but no 'schema_version'.
+        assert _catalog_shape_error({"integrations": {}}) is not None
+
+    def test_missing_integrations_is_rejected(self):
+        assert _catalog_shape_error({"schema_version": "1.0"}) is not None
+
+    def test_non_dict_integrations_is_rejected(self):
+        assert _catalog_shape_error({"schema_version": "1.0", "integrations": []}) is not None
+
+    @pytest.mark.parametrize("payload", [[], "x", 5, None])
+    def test_non_dict_payload_is_rejected(self, payload):
+        assert _catalog_shape_error(payload) is not None
+
+
 # ---------------------------------------------------------------------------
 # IntegrationCatalogEntry
 # ---------------------------------------------------------------------------
@@ -251,6 +276,48 @@ def test_fetch_and_search_all(self, tmp_path, monkeypatch):
         ids = [r["id"] for r in results]
         assert "acme-coder" in ids
 
+    def test_poisoned_cache_shape_is_dropped_and_refetched(self, tmp_path, monkeypatch):
+        """A fresh-but-mis-shaped cache (e.g. integrations as a list) must be
+        dropped and refetched, not returned — otherwise it later crashes on
+        .items(). The cache path must clear the same shape checks as a fresh
+        fetch."""
+        monkeypatch.setenv("HOME", str(tmp_path))
+        monkeypatch.setenv("USERPROFILE", str(tmp_path))
+        monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
+        (tmp_path / ".specify").mkdir()
+        cat = IntegrationCatalog(tmp_path)
+
+        catalog = {
+            "schema_version": "1.0",
+            "updated_at": "2026-01-01T00:00:00Z",
+            "integrations": {
+                "acme-coder": {
+                    "id": "acme-coder", "name": "Acme Coder", "version": "2.0.0",
+                    "description": "Community integration", "author": "acme-org",
+                    "tags": ["cli"],
+                },
+            },
+        }
+        self._patch_urlopen(monkeypatch, catalog)
+        cat.search()  # populate the cache legitimately
+
+        # Poison the cached payload (integrations as a list), keeping the fresh
+        # metadata so the age check passes and the cache branch is taken.
+        cache_dir = tmp_path / ".specify" / "integrations" / ".cache"
+        data_files = [
+            f for f in cache_dir.glob("catalog-*.json")
+            if not f.name.endswith("-metadata.json")
+        ]
+        assert data_files, "cache was not populated"
+        data_files[0].write_text(
+            json.dumps({"schema_version": "1.0", "integrations": []}),
+            encoding="utf-8",
+        )
+
+        # The poisoned cache is dropped and the (valid) source is refetched.
+        results = cat.search()
+        assert "acme-coder" in [r["id"] for r in results]
+
     def test_search_by_tag(self, tmp_path, monkeypatch):
         monkeypatch.setenv("HOME", str(tmp_path))
         monkeypatch.setenv("USERPROFILE", str(tmp_path))