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

src/specify_cli/extensions.py

@@ -1839,8 +1839,27 @@ def _fetch_single_catalog(self, entry: CatalogEntry, force_refresh: bool = False
             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"
+                )
 
             # Save to cache
             self.cache_dir.mkdir(parents=True, exist_ok=True)
@@ -1895,6 +1914,16 @@ def _get_merged_extensions(self, force_refresh: bool = False) -> List[Dict[str,
                 continue
 
             for ext_id, ext_data in catalog_data.get("extensions", {}).items():
+                # Per-entry guard: ``_fetch_single_catalog`` already validates
+                # that ``catalog_data["extensions"]`` is a mapping, but it
+                # does not (and should not) validate every entry shape there
+                # — one malformed entry shouldn't poison an otherwise valid
+                # catalog. Skip non-mapping entries here so a payload like
+                # ``{"extensions": {"foo": [], "bar": {...}}}`` still merges
+                # the valid entries without crashing on ``**ext_data``.
+                # Mirrors ``integrations/catalog.py:245``.
+                if not isinstance(ext_data, dict):
+                    continue
                 if ext_id not in merged:  # Higher-priority catalog wins
                     merged[ext_id] = {
                         **ext_data,

src/specify_cli/presets.py

@@ -2045,11 +2045,31 @@ def _fetch_single_catalog(self, entry: PresetCatalogEntry, force_refresh: bool =
             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.cache_dir.mkdir(parents=True, exist_ok=True)
             cache_file.write_text(json.dumps(catalog_data, indent=2))
@@ -2083,6 +2103,17 @@ def _get_merged_packs(self, force_refresh: bool = False) -> Dict[str, Dict[str,
             try:
                 data = self._fetch_single_catalog(entry, force_refresh)
                 for pack_id, pack_data in data.get("presets", {}).items():
+                    # Per-entry guard: ``_fetch_single_catalog`` already
+                    # validates that ``data["presets"]`` is a mapping, but it
+                    # does not (and should not) validate every entry shape
+                    # there — one malformed entry shouldn't poison an
+                    # otherwise valid catalog. Skip non-mapping entries here
+                    # so a payload like ``{"presets": {"foo": [], "bar":
+                    # {...}}}`` still merges the valid entries without
+                    # crashing on ``**pack_data``. Mirrors
+                    # ``integrations/catalog.py:245``.
+                    if not isinstance(pack_data, dict):
+                        continue
                     pack_data_with_catalog = {**pack_data, "_catalog_name": entry.name, "_install_allowed": entry.install_allowed}
                     merged[pack_id] = pack_data_with_catalog
             except PresetError:

tests/test_extensions.py

@@ -2577,6 +2577,93 @@ def fake_open(req, timeout=None):
 
         assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken"
 
+    @pytest.mark.parametrize(
+        "payload",
+        [
+            # Root is not a JSON object.
+            [],
+            "oops",
+            42,
+            None,
+            # Root is fine but ``extensions`` is the wrong type.
+            {"schema_version": "1.0", "extensions": []},
+            {"schema_version": "1.0", "extensions": "oops"},
+            {"schema_version": "1.0", "extensions": None},
+            {"schema_version": "1.0", "extensions": 42},
+        ],
+    )
+    def test_fetch_single_catalog_rejects_malformed_payload(self, temp_dir, payload):
+        """Malformed catalog payloads raise ExtensionError, not AttributeError.
+
+        Without this guard, a payload like ``{"extensions": []}`` would pass the
+        key-presence check and then crash with ``AttributeError: 'list' object
+        has no attribute 'items'`` deep inside ``_get_merged_extensions``. The
+        sibling integration catalog reader already validates both the root
+        object and the nested mapping (see ``integrations/catalog.py``); the
+        extension catalog must stay consistent.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = self._make_catalog(temp_dir)
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = CatalogEntry(
+            url="https://example.com/catalog.json",
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response):
+            with pytest.raises(ExtensionError, match="Invalid catalog format"):
+                catalog._fetch_single_catalog(entry, force_refresh=True)
+
+    def test_get_merged_extensions_skips_non_mapping_entries(self, temp_dir):
+        """Per-entry guard: one malformed entry shouldn't poison the merge.
+
+        ``_fetch_single_catalog`` validates that ``extensions`` is a mapping,
+        but it doesn't (and shouldn't) validate every entry inside it — a
+        single bad entry in an otherwise-valid catalog should be skipped, not
+        crash the whole resolve path. Mirrors the per-entry skip in
+        ``integrations/catalog.py``: a malformed entry returns no error,
+        valid entries continue to merge normally.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = self._make_catalog(temp_dir)
+        # Mix of valid entry, list-shaped entry, and string-shaped entry.
+        payload = {
+            "schema_version": "1.0",
+            "extensions": {
+                "good": {"name": "Good", "version": "1.0.0"},
+                "bad-list": [],
+                "bad-str": "oops",
+            },
+        }
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = CatalogEntry(
+            url="https://example.com/catalog.json",
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response), \
+             patch.object(catalog, "get_active_catalogs", return_value=[entry]):
+            merged = catalog._get_merged_extensions(force_refresh=True)
+
+        # Only the well-formed entry survives; the two malformed entries are
+        # silently dropped rather than raising or crashing.
+        assert [ext["id"] for ext in merged] == ["good"]
+
     def test_download_extension_sends_auth_header(self, temp_dir, monkeypatch):
         """download_extension passes Authorization header when a provider is configured."""
         from unittest.mock import patch, MagicMock

tests/test_presets.py

@@ -1514,6 +1514,92 @@ def fake_open(req, timeout=None):
 
         assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken"
 
+    @pytest.mark.parametrize(
+        "payload",
+        [
+            # Root is not a JSON object.
+            [],
+            "oops",
+            42,
+            None,
+            # Root is fine but ``presets`` is the wrong type.
+            {"schema_version": "1.0", "presets": []},
+            {"schema_version": "1.0", "presets": "oops"},
+            {"schema_version": "1.0", "presets": None},
+            {"schema_version": "1.0", "presets": 42},
+        ],
+    )
+    def test_fetch_single_catalog_rejects_malformed_payload(self, project_dir, payload):
+        """Malformed catalog payloads raise PresetError, not AttributeError.
+
+        Without this guard, a payload like ``{"presets": []}`` would pass the
+        key-presence check and then crash with ``AttributeError: 'list' object
+        has no attribute 'items'`` deep inside ``_get_merged_packs``. The
+        sibling integration catalog reader already validates both the root
+        object and the nested mapping (see ``integrations/catalog.py``); the
+        preset catalog must stay consistent.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = PresetCatalog(project_dir)
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = PresetCatalogEntry(
+            url="https://example.com/catalog.json",
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response):
+            with pytest.raises(PresetError, match="Invalid preset catalog format"):
+                catalog._fetch_single_catalog(entry, force_refresh=True)
+
+    def test_get_merged_packs_skips_non_mapping_entries(self, project_dir):
+        """Per-entry guard: one malformed entry shouldn't poison the merge.
+
+        ``_fetch_single_catalog`` validates that ``presets`` is a mapping,
+        but it doesn't (and shouldn't) validate every entry inside it — a
+        single bad entry in an otherwise-valid catalog should be skipped,
+        not crash the whole resolve path. Mirrors the per-entry skip in
+        ``integrations/catalog.py``: a malformed entry returns no error,
+        valid entries continue to merge normally.
+        """
+        from unittest.mock import patch, MagicMock
+
+        catalog = PresetCatalog(project_dir)
+        payload = {
+            "schema_version": "1.0",
+            "presets": {
+                "good": {"name": "Good", "version": "1.0.0"},
+                "bad-list": [],
+                "bad-str": "oops",
+            },
+        }
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        entry = PresetCatalogEntry(
+            url="https://example.com/catalog.json",
+            name="default",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with patch.object(catalog, "_open_url", return_value=mock_response), \
+             patch.object(catalog, "get_active_catalogs", return_value=[entry]):
+            merged = catalog._get_merged_packs(force_refresh=True)
+
+        # Only the well-formed entry survives; the two malformed entries are
+        # silently dropped rather than raising or crashing.
+        assert list(merged.keys()) == ["good"]
+
     def test_download_pack_sends_auth_header(self, project_dir, monkeypatch):
         """download_pack passes Authorization header when configured."""
         from unittest.mock import patch, MagicMock