fix: reject non-object workflow caches

src/specify_cli/workflows/catalog.py

@@ -495,6 +495,8 @@ def _is_url_cache_valid(self, url: str) -> bool:
         try:
             with open(meta_file, encoding="utf-8") as f:
                 meta = json.load(f)
+            if not isinstance(meta, dict):
+                return False
             fetched_at = float(meta.get("fetched_at", 0))
             return (time.time() - fetched_at) < self.CACHE_DURATION
         except (json.JSONDecodeError, OSError, TypeError, ValueError):
@@ -509,7 +511,9 @@ def _fetch_single_catalog(
         if not force_refresh and self._is_url_cache_valid(entry.url):
             try:
                 with open(cache_file, encoding="utf-8") as f:
-                    return json.load(f)
+                    cached = json.load(f)
+                if isinstance(cached, dict):
+                    return cached
             except (json.JSONDecodeError, OSError):
                 # Ignore invalid/unreadable cache and fall back to fetching from source.
                 pass
@@ -574,7 +578,9 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
             if cache_file.exists():
                 try:
                     with open(cache_file, encoding="utf-8") as f:
-                        return json.load(f)
+                        cached = json.load(f)
+                    if isinstance(cached, dict):
+                        return cached
                 except (json.JSONDecodeError, ValueError, OSError):
                     # Stale-cache read failed; let the original fetch error propagate.
                     pass
@@ -1184,6 +1190,8 @@ def _is_url_cache_valid(self, url: str) -> bool:
         try:
             with open(meta_file, encoding="utf-8") as f:
                 meta = json.load(f)
+            if not isinstance(meta, dict):
+                return False
             fetched_at = float(meta.get("fetched_at", 0))
             return (time.time() - fetched_at) < self.CACHE_DURATION
         except (json.JSONDecodeError, OSError, TypeError, ValueError):

tests/test_workflows.py

@@ -6482,6 +6482,97 @@ def test_load_symlinked_workflows_dir_fails_closed_not_silently_empty(
 class TestWorkflowCatalog:
     """Test WorkflowCatalog catalog resolution."""
 
+    @pytest.mark.parametrize("catalog_type", ["workflow", "step"])
+    def test_non_mapping_cache_metadata_is_invalid(
+        self, project_dir, catalog_type
+    ):
+        from specify_cli.workflows.catalog import StepCatalog, WorkflowCatalog
+
+        catalog_cls = WorkflowCatalog if catalog_type == "workflow" else StepCatalog
+        catalog = catalog_cls(project_dir)
+        _, metadata_path = catalog._get_cache_paths(
+            f"https://example.com/{catalog_type}.json"
+        )
+        metadata_path.parent.mkdir(parents=True, exist_ok=True)
+        metadata_path.write_text("[]", encoding="utf-8")
+
+        assert catalog._is_url_cache_valid(
+            f"https://example.com/{catalog_type}.json"
+        ) is False
+
+    def test_non_mapping_cached_workflow_catalog_is_refetched(
+        self, project_dir, monkeypatch
+    ):
+        import io
+
+        from specify_cli.authentication import http as auth_http
+        from specify_cli.workflows.catalog import (
+            WorkflowCatalog,
+            WorkflowCatalogEntry,
+        )
+
+        url = "https://example.com/workflows.json"
+        catalog = WorkflowCatalog(project_dir)
+        cache_path, metadata_path = catalog._get_cache_paths(url)
+        cache_path.parent.mkdir(parents=True, exist_ok=True)
+        cache_path.write_text("[]", encoding="utf-8")
+        metadata_path.write_text(
+            json.dumps({"fetched_at": 4_102_444_800}),
+            encoding="utf-8",
+        )
+
+        payload = {"schema_version": "1.0", "workflows": {}}
+
+        class _FakeResponse(io.BytesIO):
+            def geturl(self):
+                return url
+
+        monkeypatch.setattr(
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(
+                json.dumps(payload).encode("utf-8")
+            ),
+        )
+        entry = WorkflowCatalogEntry(
+            url=url,
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+
+        assert catalog._fetch_single_catalog(entry) == payload
+
+    def test_non_mapping_stale_workflow_catalog_is_rejected(
+        self, project_dir, monkeypatch
+    ):
+        from specify_cli.authentication import http as auth_http
+        from specify_cli.workflows.catalog import (
+            WorkflowCatalog,
+            WorkflowCatalogEntry,
+            WorkflowCatalogError,
+        )
+
+        url = "https://example.com/workflows.json"
+        catalog = WorkflowCatalog(project_dir)
+        cache_path, _ = catalog._get_cache_paths(url)
+        cache_path.parent.mkdir(parents=True, exist_ok=True)
+        cache_path.write_text("[]", encoding="utf-8")
+
+        def _offline(url, timeout=30, redirect_validator=None):
+            raise OSError("offline")
+
+        monkeypatch.setattr(auth_http, "open_url", _offline)
+        entry = WorkflowCatalogEntry(
+            url=url,
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+
+        with pytest.raises(WorkflowCatalogError, match="Failed to fetch catalog"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+
     def test_search_with_non_string_fields(self, project_dir, monkeypatch):
         """Non-string workflow fields (null/int name/description) must not
         raise TypeError in search — StepCatalog.search already coerces these."""