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

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

The catalog cache-read branch returned json.loads(cache_file) directly, skipping
the shape validation the fresh-fetch branch enforces (dict root + 'integrations'
mapping). A poisoned or older-format cache (e.g. {"integrations": []}) was
therefore returned as-is and later crashed with 'AttributeError: list object has
no attribute items' when the caller iterated integrations. Validate the cached
object the same way; the raised ValueError is already caught by the surrounding
handler, which drops the corrupt cache and refetches from source.

Test: a fresh-but-mis-shaped cache is dropped and the valid source refetched
(fails before: AttributeError).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(integrations): share one catalog-shape validator across cache and fetch

Address review: the cache-read path checked only that the payload was a
dict with a dict 'integrations', while the fresh-fetch path also required
'schema_version'. That asymmetry let an older/poisoned cache such as
{"integrations": {}} (no schema_version) bypass the format contract
instead of being dropped and refetched.

Introduce a shared `_catalog_shape_error()` helper and use it in both
paths so they enforce the same contract (dict + schema_version + dict
integrations). The fresh path still raises IntegrationCatalogError with
the "Invalid catalog format from <url>" prefix; the cache path still
raises ValueError (caught to drop+refetch). Add a test for the
missing-schema_version cache case.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(integrations): unit-test the shared catalog-shape validator directly

Replace the integration-level missing-schema_version cache test (which
was masked by multi-source merging — a sibling catalog source still
supplied the entry, so it passed regardless of the fix) with a direct
unit test of _catalog_shape_error. This deterministically proves both
paths now reject a payload missing schema_version, a non-dict
integrations, or a non-dict payload, and accept a well-formed one.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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 @@ class IntegrationCatalog(CatalogStackBase):
                     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 @@ class IntegrationCatalog(CatalogStackBase):
                     self._validate_catalog_url(final_url)
                 catalog_data = json.loads(resp.read())
 
-            if not isinstance(catalog_data, dict):
+            shape_error = _catalog_shape_error(catalog_data)
+            if shape_error is not None:
                 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):
-                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 @@ from specify_cli.integrations.catalog import (
     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 @@ class TestCatalogFetch:
         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))