Validate deeply nested workflow and step catalogs

Classify JSON recursion during parsing or cache writes as malformed catalog data, recover from poisoned caches, and keep non-fetch step catalog errors blocking exact-ID lookup.

Assisted-by: GitHub Copilot (model: GPT-6 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: mnriem

src/specify_cli/workflows/catalog/_domain.py

@@ -544,6 +544,7 @@ def validate_payload(data: Any) -> dict[str, Any]:
             except (
                 UnicodeDecodeError,
                 json.JSONDecodeError,
+                RecursionError,
                 OSError,
                 WorkflowCatalogValidationError,
             ):
@@ -613,6 +614,7 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
             RedirectPolicyError,
             UnicodeError,
             json.JSONDecodeError,
+            RecursionError,
         ) as exc:
             raise WorkflowCatalogValidationError(
                 f"Invalid workflow catalog from {entry.url}: {exc}"
@@ -627,6 +629,7 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 except (
                     json.JSONDecodeError,
                     ValueError,
+                    RecursionError,
                     OSError,
                     WorkflowCatalogValidationError,
                 ):
@@ -647,6 +650,10 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 json.dump({"url": entry.url, "fetched_at": time.time()}, f)
         except OSError:
             pass  # Proceed without caching if disk write fails
+        except RecursionError as exc:
+            raise WorkflowCatalogValidationError(
+                f"Invalid workflow catalog from {entry.url}: excessive nesting ({exc})"
+            ) from exc
 
         return data
 

src/specify_cli/workflows/step/catalog/_domain.py

@@ -509,6 +509,7 @@ def validate_payload(data: Any) -> dict[str, Any]:
             except (
                 UnicodeDecodeError,
                 json.JSONDecodeError,
+                RecursionError,
                 OSError,
                 StepCatalogValidationError,
             ):
@@ -580,6 +581,7 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
             RedirectPolicyError,
             UnicodeError,
             json.JSONDecodeError,
+            RecursionError,
         ) as exc:
             raise StepCatalogValidationError(
                 f"Invalid step catalog from {entry.url}: {exc}"
@@ -595,6 +597,7 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 except (
                     json.JSONDecodeError,
                     ValueError,
+                    RecursionError,
                     OSError,
                     StepCatalogValidationError,
                 ):
@@ -615,6 +618,10 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                     json.dump({"url": entry.url, "fetched_at": time.time()}, f)
             except OSError:
                 pass  # Proceed without caching if disk write fails
+            except RecursionError as exc:
+                raise StepCatalogValidationError(
+                    f"Invalid step catalog from {entry.url}: excessive nesting ({exc})"
+                ) from exc
 
         return data
 
@@ -635,8 +642,8 @@ def _get_merged_steps(
             except _DuplicateCatalogField:
                 raise
             except StepCatalogError as exc:
-                if target_id is not None and isinstance(
-                    exc, StepCatalogValidationError
+                if target_id is not None and not isinstance(
+                    exc, StepCatalogFetchError
                 ):
                     raise
                 if (

tests/specify_cli/bundles/test_references.py

@@ -646,6 +646,108 @@ def open_url(url, **kwargs):
     assert warnings == []
 
 
+@pytest.mark.parametrize("kind", ["workflows", "steps"])
+@pytest.mark.parametrize("source", ["network", "fresh-cache", "stale-cache"])
+def test_deeply_nested_catalog_is_handled_as_malformed_data(
+    tmp_path, monkeypatch, kind, source,
+):
+    import io
+    import json
+    from urllib.error import URLError
+
+    from specify_cli.authentication import http
+    from specify_cli.workflows.catalog import (
+        StepCatalog,
+        StepCatalogEntry,
+        StepCatalogFetchError,
+        StepCatalogValidationError,
+        WorkflowCatalog,
+        WorkflowCatalogEntry,
+        WorkflowCatalogFetchError,
+        WorkflowCatalogValidationError,
+    )
+
+    catalog_type, entry_type, validation_error, fetch_error = {
+        "workflows": (
+            WorkflowCatalog, WorkflowCatalogEntry,
+            WorkflowCatalogValidationError, WorkflowCatalogFetchError,
+        ),
+        "steps": (
+            StepCatalog, StepCatalogEntry,
+            StepCatalogValidationError, StepCatalogFetchError,
+        ),
+    }[kind]
+    catalog = catalog_type(tmp_path)
+    entry = entry_type("https://example.com/catalog.json", "trusted", 1, True)
+    nested = b'{"nested":' + b"[" * 10000 + b"0" + b"]" * 10000 + b"}"
+    valid = {"schema_version": "1.0", kind: {"requested": {"version": "1.0.0"}}}
+
+    class Response(io.BytesIO):
+        def geturl(self):
+            return entry.url
+
+    if source != "network":
+        cache_file, _ = catalog._get_cache_paths(entry.url)
+        cache_file.parent.mkdir(parents=True, exist_ok=True)
+        cache_file.write_bytes(nested)
+        monkeypatch.setattr(
+            catalog, "_is_url_cache_valid", lambda _url: source == "fresh-cache"
+        )
+
+    def open_url(*args, **kwargs):
+        if source == "stale-cache":
+            raise URLError("connection failed")
+        payload = nested if source == "network" else json.dumps(valid).encode()
+        return Response(payload)
+
+    monkeypatch.setattr(http, "open_url", open_url)
+    if source == "network":
+        with pytest.raises(validation_error, match="Invalid.*catalog"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+    elif source == "stale-cache":
+        with pytest.raises(fetch_error, match="Failed to fetch catalog"):
+            catalog._fetch_single_catalog(entry)
+    else:
+        assert catalog._fetch_single_catalog(entry) == valid
+
+
+@pytest.mark.parametrize("kind", ["workflows", "steps"])
+def test_deep_catalog_does_not_escape_during_cache_write(
+    tmp_path, monkeypatch, kind,
+):
+    import io
+
+    from specify_cli.authentication import http
+    from specify_cli.workflows.catalog import (
+        StepCatalog,
+        StepCatalogEntry,
+        StepCatalogValidationError,
+        WorkflowCatalog,
+        WorkflowCatalogEntry,
+        WorkflowCatalogValidationError,
+    )
+
+    catalog_type, entry_type, error_type = {
+        "workflows": (
+            WorkflowCatalog, WorkflowCatalogEntry, WorkflowCatalogValidationError,
+        ),
+        "steps": (StepCatalog, StepCatalogEntry, StepCatalogValidationError),
+    }[kind]
+    catalog = catalog_type(tmp_path)
+    entry = entry_type("https://example.com/catalog.json", "trusted", 1, True)
+    payload = b'{"nested":' + b"[" * 1200 + b"0" + b"]" * 1200 + b"}"
+
+    class Response(io.BytesIO):
+        def geturl(self):
+            return entry.url
+
+    monkeypatch.setattr(
+        http, "open_url", lambda *args, **kwargs: Response(payload)
+    )
+    with pytest.raises(error_type, match="Invalid.*catalog"):
+        catalog._fetch_single_catalog(entry, force_refresh=True)
+
+
 @pytest.mark.parametrize("kind", ["extensions", "workflows", "steps"])
 def test_targeted_lookup_stops_before_lower_priority_catalog(
     tmp_path, monkeypatch, kind,

tests/specify_cli/workflows/step/test_catalog_versions.py

@@ -114,6 +114,34 @@ def test_list_catalog_rejects_duplicate_step_ids(project_dir, monkeypatch):
         catalog.get_step_info("deploy")
 
 
+@pytest.mark.parametrize("raised_during_fetch", [False, True])
+def test_targeted_lookup_rejects_duplicate_ids_before_lower_catalog(
+    project_dir, monkeypatch, raised_during_fetch,
+):
+    catalog = StepCatalog(project_dir)
+    sources = [
+        StepCatalogEntry("https://example.com/high.json", "high", 1, True),
+        StepCatalogEntry("https://example.com/low.json", "low", 2, True),
+    ]
+    monkeypatch.setattr(catalog, "get_active_catalogs", lambda: sources)
+
+    def fetch(source, force_refresh=False):
+        if source.name == "high":
+            if raised_during_fetch:
+                raise StepCatalogError("Duplicate step ID 'deploy' in catalog 'high'.")
+            return {"steps": [
+                {"id": "deploy", "version": "1.0"},
+                {"id": "deploy", "version": "2.0"},
+            ]}
+        return {"steps": {"deploy": {"version": "3.0"}}}
+
+    monkeypatch.setattr(catalog, "_fetch_single_catalog", fetch)
+    with pytest.raises(StepCatalogError, match="Duplicate step ID 'deploy'"):
+        catalog.get_step_info("deploy")
+    if raised_during_fetch:
+        assert catalog.search(query="deploy")[0]["version"] == "3.0"
+
+
 @pytest.mark.parametrize("cached", [True, False])
 def test_duplicate_json_release_key_is_not_silently_overwritten(
     project_dir, monkeypatch, cached