fix(catalogs): validate cached extension and preset payload shape

Addresses Copilot review feedback on this PR (round 2).

The earlier commits in this branch added payload-shape validation on the
network fetch path. The cache-hit path still returned
``json.loads(cache_file.read_text())`` directly without re-checking the
shape, so a cache poisoned by an older spec-kit version (or a manual
edit, or an upstream that briefly served a bad payload before the
network guards landed) would re-crash every invocation of
``_get_merged_extensions`` / ``_get_merged_packs`` with
``AttributeError: 'list' object has no attribute 'items'`` despite the
cache being "valid" by age.

Extracts the shape validation into ``_validate_catalog_payload`` on both
``ExtensionCatalog`` and ``PresetCatalog``, and calls it from both the
cache-load and network-fetch branches of ``_fetch_single_catalog``. If
the cached payload fails validation, the cache read is treated like a
``json.JSONDecodeError`` — the cached value is discarded and the
function falls through to the network fetch, which refreshes the cache
with a clean payload on success. Never propagates ``AttributeError`` to
the caller.

Regression tests parametrize the four root-bad-type variants plus three
``extensions``/``presets``-bad-type variants per file, asserting that a
poisoned cache silently recovers via network refetch and returns the
freshly-fetched payload.

src/specify_cli/extensions.py

@@ -1702,6 +1702,44 @@ class ExtensionCatalog(CatalogStackBase):
         from specify_cli.authentication.http import open_url
         return open_url(url, timeout)
 
+    def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
+        """Validate a parsed catalog payload's shape.
+
+        Applied to both network-fetched and cache-loaded payloads so a
+        once-poisoned cache (older spec-kit version, manual edit, upstream
+        served a bad payload before the network-side guards were added)
+        cannot re-crash ``_get_merged_extensions`` on subsequent calls.
+
+        Checking only key presence would let a payload like
+        ``{"extensions": []}`` or ``{"extensions": null}`` slip through
+        here and then crash with ``AttributeError: 'list' object has no
+        attribute 'items'`` deep inside ``_get_merged_extensions``. The
+        sibling integration catalog reader already guards both the root
+        object and the nested mapping (see ``integrations/catalog.py``);
+        the extension catalog must stay consistent so a malformed payload
+        surfaces as the user-facing ``Invalid catalog format`` error
+        instead of a raw Python traceback.
+
+        Args:
+            catalog_data: Parsed JSON payload from the catalog source.
+            url: Source URL — used in the error message so the user can
+                tell which catalog in a multi-catalog stack is malformed.
+
+        Raises:
+            ExtensionError: If the payload's shape is invalid.
+        """
+        if not isinstance(catalog_data, dict):
+            raise ExtensionError(
+                f"Invalid catalog format from {url}: expected a JSON object"
+            )
+        if "schema_version" not in catalog_data or "extensions" not in catalog_data:
+            raise ExtensionError(f"Invalid catalog format from {url}")
+        if not isinstance(catalog_data.get("extensions"), dict):
+            raise ExtensionError(
+                f"Invalid catalog format from {url}: "
+                "'extensions' must be a JSON object"
+            )
+
     def get_active_catalogs(self) -> List[CatalogEntry]:
         """Get the ordered list of active catalogs.
 
@@ -1827,11 +1865,19 @@ class ExtensionCatalog(CatalogStackBase):
                     # If metadata is invalid or missing expected fields, treat cache as invalid
                     pass
 
-        # Use cache if valid
+        # Use cache if valid. A previously-cached payload must clear the
+        # same shape checks as a freshly-fetched one — otherwise a once-
+        # poisoned cache (older spec-kit version, manual edit, upstream
+        # served a bad payload before the network-side guards were added)
+        # would re-crash on every invocation despite the cache being
+        # "valid" by age. If validation fails on the cached read, fall
+        # through to the network fetch path so the cache gets refreshed.
         if is_valid:
             try:
-                return json.loads(cache_file.read_text())
-            except json.JSONDecodeError:
+                cached_data = json.loads(cache_file.read_text())
+                self._validate_catalog_payload(cached_data, entry.url)
+                return cached_data
+            except (json.JSONDecodeError, ExtensionError):
                 pass
 
         # Fetch from network
@@ -1839,27 +1885,7 @@ class ExtensionCatalog(CatalogStackBase):
             with self._open_url(entry.url, timeout=10) as response:
                 catalog_data = json.loads(response.read())
 
-            # Validate payload shape before iteration. Checking only key
-            # presence would let a payload like ``{"extensions": []}`` or
-            # ``{"extensions": null}`` slip through here and then crash with
-            # ``AttributeError: 'list' object has no attribute 'items'`` deep
-            # inside ``_get_merged_extensions``. The sibling integration
-            # catalog reader already guards both the root object and the
-            # nested mapping (see ``integrations/catalog.py``); the extension
-            # catalog must stay consistent so a malformed upstream surfaces as
-            # the user-facing ``Invalid catalog format`` error instead of a
-            # raw Python traceback.
-            if not isinstance(catalog_data, dict):
-                raise ExtensionError(
-                    f"Invalid catalog format from {entry.url}: expected a JSON object"
-                )
-            if "schema_version" not in catalog_data or "extensions" not in catalog_data:
-                raise ExtensionError(f"Invalid catalog format from {entry.url}")
-            if not isinstance(catalog_data.get("extensions"), dict):
-                raise ExtensionError(
-                    f"Invalid catalog format from {entry.url}: "
-                    "'extensions' must be a JSON object"
-                )
+            self._validate_catalog_payload(catalog_data, entry.url)
 
             # Save to cache
             self.cache_dir.mkdir(parents=True, exist_ok=True)

src/specify_cli/presets.py

@@ -1860,6 +1860,48 @@ class PresetCatalog:
         from specify_cli.authentication.http import open_url
         return open_url(url, timeout)
 
+    def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
+        """Validate a parsed preset-catalog payload's shape.
+
+        Applied to both network-fetched and cache-loaded payloads so a
+        once-poisoned cache (older spec-kit version, manual edit, upstream
+        served a bad payload before the network-side guards were added)
+        cannot re-crash ``_get_merged_packs`` on subsequent calls.
+
+        Checking only key presence would let a payload like
+        ``{"presets": []}`` or ``{"presets": null}`` slip through here and
+        then crash with ``AttributeError: 'list' object has no attribute
+        'items'`` deep inside ``_get_merged_packs``. The sibling
+        integration catalog reader already guards both the root object and
+        the nested mapping (see ``integrations/catalog.py``); the preset
+        catalog must stay consistent so a malformed payload surfaces as
+        the user-facing ``Invalid preset catalog format`` error instead of
+        a raw Python traceback.
+
+        Args:
+            catalog_data: Parsed JSON payload from the catalog source.
+            url: Source URL — used in the error message so the user can
+                tell which catalog in a multi-catalog stack is malformed.
+
+        Raises:
+            PresetError: If the payload's shape is invalid.
+        """
+        if not isinstance(catalog_data, dict):
+            raise PresetError(
+                f"Invalid preset catalog format from {url}: "
+                "expected a JSON object"
+            )
+        if (
+            "schema_version" not in catalog_data
+            or "presets" not in catalog_data
+        ):
+            raise PresetError("Invalid preset catalog format")
+        if not isinstance(catalog_data.get("presets"), dict):
+            raise PresetError(
+                f"Invalid preset catalog format from {url}: "
+                "'presets' must be a JSON object"
+            )
+
     def _load_catalog_config(self, config_path: Path) -> Optional[List[PresetCatalogEntry]]:
         """Load catalog stack configuration from a YAML file.
 
@@ -2035,41 +2077,25 @@ class PresetCatalog:
         """
         cache_file, metadata_file = self._get_cache_paths(entry.url)
 
+        # Use cache if valid. A previously-cached payload must clear the
+        # same shape checks as a freshly-fetched one — otherwise a once-
+        # poisoned cache would re-crash on every invocation despite the
+        # cache being "valid" by age. If validation fails on the cached
+        # read, fall through to the network fetch path so the cache gets
+        # refreshed.
         if not force_refresh and self._is_url_cache_valid(entry.url):
             try:
-                return json.loads(cache_file.read_text())
-            except json.JSONDecodeError:
+                cached_data = json.loads(cache_file.read_text())
+                self._validate_catalog_payload(cached_data, entry.url)
+                return cached_data
+            except (json.JSONDecodeError, PresetError):
                 pass
 
         try:
             with self._open_url(entry.url, timeout=10) as response:
                 catalog_data = json.loads(response.read())
 
-            # Validate payload shape before iteration. Checking only key
-            # presence would let a payload like ``{"presets": []}`` or
-            # ``{"presets": null}`` slip through here and then crash with
-            # ``AttributeError: 'list' object has no attribute 'items'`` deep
-            # inside ``_get_merged_packs``. The sibling integration catalog
-            # reader already guards both the root object and the nested
-            # mapping (see ``integrations/catalog.py``); the preset catalog
-            # must stay consistent so a malformed upstream surfaces as the
-            # user-facing ``Invalid preset catalog format`` error instead of
-            # a raw Python traceback.
-            if not isinstance(catalog_data, dict):
-                raise PresetError(
-                    f"Invalid preset catalog format from {entry.url}: "
-                    "expected a JSON object"
-                )
-            if (
-                "schema_version" not in catalog_data
-                or "presets" not in catalog_data
-            ):
-                raise PresetError("Invalid preset catalog format")
-            if not isinstance(catalog_data.get("presets"), dict):
-                raise PresetError(
-                    f"Invalid preset catalog format from {entry.url}: "
-                    "'presets' must be a JSON object"
-                )
+            self._validate_catalog_payload(catalog_data, entry.url)
 
             self.cache_dir.mkdir(parents=True, exist_ok=True)
             cache_file.write_text(json.dumps(catalog_data, indent=2))

tests/test_extensions.py

@@ -2622,6 +2622,76 @@ class TestExtensionCatalog:
             with pytest.raises(ExtensionError, match="Invalid catalog format"):
                 catalog._fetch_single_catalog(entry, force_refresh=True)
 
+    @pytest.mark.parametrize(
+        "cached_payload",
+        [
+            [],
+            "oops",
+            42,
+            None,
+            {"schema_version": "1.0", "extensions": []},
+            {"schema_version": "1.0", "extensions": "oops"},
+            {"schema_version": "1.0", "extensions": None},
+        ],
+    )
+    def test_fetch_single_catalog_rejects_malformed_cached_payload(
+        self, temp_dir, cached_payload
+    ):
+        """A poisoned cache silently falls back to the network instead of
+        crashing — cached payloads pass through the same shape validation
+        as freshly-fetched ones.
+
+        Without this, a cache poisoned by an older spec-kit version (or a
+        manual edit, or an upstream that briefly served a bad payload
+        before the network guards landed) would re-crash every invocation
+        of ``_get_merged_extensions`` despite the cache being "valid" by
+        age. The recovery contract is: if the cached payload fails
+        validation, drop it and refetch — never propagate
+        ``AttributeError`` to the caller.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = self._make_catalog(temp_dir)
+
+        # Poison the default-URL cache. ``DEFAULT_CATALOG_URL`` is the
+        # branch that goes through ``is_cache_valid()`` (the non-default
+        # branch uses per-URL hashed cache files but the same code path
+        # below).
+        catalog.cache_dir.mkdir(parents=True, exist_ok=True)
+        catalog.cache_file.write_text(json.dumps(cached_payload))
+        catalog.cache_metadata_file.write_text(
+            json.dumps(
+                {
+                    "cached_at": datetime.now(timezone.utc).isoformat(),
+                    "catalog_url": ExtensionCatalog.DEFAULT_CATALOG_URL,
+                }
+            )
+        )
+
+        # Network refetch returns a valid payload so the recovery path
+        # can complete.
+        valid = {
+            "schema_version": "1.0",
+            "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
+        }
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(valid).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = CatalogEntry(
+            url=ExtensionCatalog.DEFAULT_CATALOG_URL,
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response):
+            result = catalog._fetch_single_catalog(entry, force_refresh=False)
+
+        # The poisoned cache was discarded and the network payload returned.
+        assert result == valid
+
     def test_get_merged_extensions_skips_non_mapping_entries(self, temp_dir):
         """Per-entry guard: one malformed entry shouldn't poison the merge.
 

tests/test_presets.py

@@ -1559,6 +1559,77 @@ class TestPresetCatalog:
             with pytest.raises(PresetError, match="Invalid preset catalog format"):
                 catalog._fetch_single_catalog(entry, force_refresh=True)
 
+    @pytest.mark.parametrize(
+        "cached_payload",
+        [
+            [],
+            "oops",
+            42,
+            None,
+            {"schema_version": "1.0", "presets": []},
+            {"schema_version": "1.0", "presets": "oops"},
+            {"schema_version": "1.0", "presets": None},
+        ],
+    )
+    def test_fetch_single_catalog_rejects_malformed_cached_payload(
+        self, project_dir, cached_payload
+    ):
+        """A poisoned cache silently falls back to the network instead of
+        crashing — cached payloads pass through the same shape validation
+        as freshly-fetched ones.
+
+        Without this, a cache poisoned by an older spec-kit version (or a
+        manual edit, or an upstream that briefly served a bad payload
+        before the network guards landed) would re-crash every invocation
+        of ``_get_merged_packs`` despite the cache being "valid" by age.
+        The recovery contract is: if the cached payload fails validation,
+        drop it and refetch — never propagate ``AttributeError`` to the
+        caller.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = PresetCatalog(project_dir)
+
+        # Poison the default-URL cache. ``DEFAULT_CATALOG_URL`` and
+        # non-default URLs both flow through the same cache-load branch.
+        cache_file, metadata_file = catalog._get_cache_paths(
+            catalog.DEFAULT_CATALOG_URL
+        )
+        cache_file.parent.mkdir(parents=True, exist_ok=True)
+        cache_file.write_text(json.dumps(cached_payload))
+        metadata_file.write_text(
+            json.dumps(
+                {
+                    "cached_at": datetime.now(timezone.utc).isoformat(),
+                    "catalog_url": catalog.DEFAULT_CATALOG_URL,
+                }
+            )
+        )
+
+        # Network refetch returns a valid payload so the recovery path
+        # can complete.
+        valid = {
+            "schema_version": "1.0",
+            "presets": {"foo": {"name": "Foo", "version": "1.0.0"}},
+        }
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(valid).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = PresetCatalogEntry(
+            url=catalog.DEFAULT_CATALOG_URL,
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response):
+            result = catalog._fetch_single_catalog(entry, force_refresh=False)
+
+        # The poisoned cache was discarded and the network payload returned.
+        assert result == valid
+
     def test_get_merged_packs_skips_non_mapping_entries(self, project_dir):
         """Per-entry guard: one malformed entry shouldn't poison the merge.