fix(workflows): validate origin and release metadata

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: marcelsafin

src/specify_cli/_github_http.py

@@ -24,6 +24,7 @@
     "api.github.com",
     "codeload.github.com",
 })
+_MAX_RELEASE_METADATA_BYTES = 5 * 1024 * 1024
 
 
 def build_github_request(url: str) -> urllib.request.Request:
@@ -68,6 +69,8 @@ def resolve_github_release_asset_api_url(
     open_url_fn: Callable,
     timeout: int = 60,
     github_hosts: tuple[str, ...] = (),
+    redirect_validator: Callable[[str, str], None] | None = None,
+    max_metadata_bytes: int = _MAX_RELEASE_METADATA_BYTES,
 ) -> Optional[str]:
     """Resolve a GitHub release browser-download URL to its REST API asset URL.
 
@@ -91,6 +94,8 @@ def resolve_github_release_asset_api_url(
             authenticated release-metadata lookup.
         timeout: Per-request timeout in seconds.
         github_hosts: Host patterns to treat as GitHub Enterprise Server.
+        redirect_validator: Optional policy applied to metadata redirects.
+        max_metadata_bytes: Maximum release-metadata response size.
     """
     import json
     import urllib.error
@@ -149,13 +154,33 @@ def _is_asset_path(segments: list[str]) -> bool:
     release_url = f"{api_base}/repos/{owner}/{repo}/releases/tags/{encoded_tag}"
 
     try:
-        with open_url_fn(release_url, timeout=timeout) as response:
-            release_data = json.loads(response.read())
-    except (urllib.error.URLError, json.JSONDecodeError):
+        open_kwargs = {"timeout": timeout}
+        if redirect_validator is not None:
+            open_kwargs["redirect_validator"] = redirect_validator
+        with open_url_fn(release_url, **open_kwargs) as response:
+            raw_release_data = response.read(max_metadata_bytes + 1)
+            if len(raw_release_data) > max_metadata_bytes:
+                raise ValueError("GitHub release metadata exceeds size limit")
+            release_data = json.loads(raw_release_data)
+    except (
+        urllib.error.URLError,
+        json.JSONDecodeError,
+        TypeError,
+        ValueError,
+    ):
         return None
 
-    for asset in release_data.get("assets", []):
-        if asset.get("name") == asset_name and asset.get("url"):
+    if not isinstance(release_data, dict):
+        return None
+    assets = release_data.get("assets", [])
+    if not isinstance(assets, list):
+        return None
+    for asset in assets:
+        if (
+            isinstance(asset, dict)
+            and asset.get("name") == asset_name
+            and asset.get("url")
+        ):
             return str(asset["url"])
 
     return None

src/specify_cli/workflows/_commands.py

@@ -1612,7 +1612,11 @@ def _validate_and_install_local(
 
         _wf_url_extra_headers = None
         _resolved_wf_url = _resolve_gh_asset(
-            download_url, _open_url, timeout=30, github_hosts=_github_provider_hosts()
+            download_url,
+            _open_url,
+            timeout=30,
+            github_hosts=_github_provider_hosts(),
+            redirect_validator=_reject_insecure_download_redirect,
         )
         if _resolved_wf_url:
             download_url = _resolved_wf_url
@@ -1825,7 +1829,11 @@ def versions_match(actual: object, expected: str) -> bool:
 
         _wf_cat_extra_headers = None
         _resolved_workflow_url = _resolve_gh_asset(
-            workflow_url, _open_url, timeout=30, github_hosts=_github_provider_hosts()
+            workflow_url,
+            _open_url,
+            timeout=30,
+            github_hosts=_github_provider_hosts(),
+            redirect_validator=_reject_insecure_download_redirect,
         )
         if _resolved_workflow_url:
             workflow_url = _resolved_workflow_url

src/specify_cli/workflows/engine.py

@@ -423,6 +423,42 @@ def _validate_run_id(cls, run_id: str) -> None:
                 "alphanumeric character)."
             )
 
+    @staticmethod
+    def _validate_installed_origin(
+        installed_workflow_id: str | None,
+        installed_registry_root: str | None,
+    ) -> None:
+        """Validate persisted installed-workflow ownership metadata."""
+        if installed_workflow_id is not None:
+            if not isinstance(installed_workflow_id, str):
+                raise ValueError(
+                    "Invalid run state: 'installed_workflow_id' must be a "
+                    f"string or null, got {type(installed_workflow_id).__name__}"
+                )
+            if not _ID_PATTERN.fullmatch(installed_workflow_id):
+                raise ValueError(
+                    "Invalid run state: 'installed_workflow_id' must be a "
+                    "lowercase alphanumeric workflow ID with hyphens"
+                )
+        if installed_registry_root is not None:
+            if not isinstance(installed_registry_root, str):
+                raise ValueError(
+                    "Invalid run state: 'installed_registry_root' must be a "
+                    f"string or null, got {type(installed_registry_root).__name__}"
+                )
+            if not installed_registry_root or not Path(
+                installed_registry_root
+            ).is_absolute():
+                raise ValueError(
+                    "Invalid run state: 'installed_registry_root' must be "
+                    "an absolute path or null"
+                )
+            if installed_workflow_id is None:
+                raise ValueError(
+                    "Invalid run state: 'installed_registry_root' requires "
+                    "'installed_workflow_id'"
+                )
+
     def __init__(
         self,
         run_id: str | None = None,
@@ -442,6 +478,9 @@ def __init__(
         else:
             self.run_id = run_id
         self._validate_run_id(self.run_id)
+        self._validate_installed_origin(
+            installed_workflow_id, installed_registry_root
+        )
         self.workflow_id = workflow_id
         self.project_root = project_root or Path(".")
         # Identifies the installed workflow (if any) this run was started
@@ -601,36 +640,7 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
             )
 
         installed_workflow_id = state_data.get("installed_workflow_id")
-        if installed_workflow_id is not None:
-            if not isinstance(installed_workflow_id, str):
-                raise ValueError(
-                    "Invalid run state: 'installed_workflow_id' must be a "
-                    f"string or null, got {type(installed_workflow_id).__name__}"
-                )
-            if not _ID_PATTERN.fullmatch(installed_workflow_id):
-                raise ValueError(
-                    "Invalid run state: 'installed_workflow_id' must be a "
-                    "lowercase alphanumeric workflow ID with hyphens"
-                )
         installed_registry_root = state_data.get("installed_registry_root")
-        if installed_registry_root is not None:
-            if not isinstance(installed_registry_root, str):
-                raise ValueError(
-                    "Invalid run state: 'installed_registry_root' must be a "
-                    f"string or null, got {type(installed_registry_root).__name__}"
-                )
-            if not installed_registry_root or not Path(
-                installed_registry_root
-            ).is_absolute():
-                raise ValueError(
-                    "Invalid run state: 'installed_registry_root' must be "
-                    "an absolute path or null"
-                )
-            if installed_workflow_id is None:
-                raise ValueError(
-                    "Invalid run state: 'installed_registry_root' requires "
-                    "'installed_workflow_id'"
-                )
 
         state = cls(
             run_id=state_data["run_id"],

tests/contract/test_bundle_cli.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+import io
 import json
 from pathlib import Path
 from unittest.mock import patch
@@ -468,25 +469,16 @@ def test_install_integration_override_cannot_bypass_clash_guard(project: Path):
 # ===== Private GitHub release asset URL resolution =====
 
 
-class FakeBundleResponse:
+class FakeBundleResponse(io.BytesIO):
     """Minimal context-manager response stub for open_url fakes."""
 
     def __init__(self, data: bytes, url: str = "https://api.github.com/repos/org/repo/releases/assets/99"):
-        self._data = data
+        super().__init__(data)
         self._url = url
 
-    def read(self) -> bytes:
-        return self._data
-
     def geturl(self) -> str:
         return self._url
 
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *_):
-        return False
-
 
 def _make_catalog_config(catalog_path: Path, project: Path) -> None:
     """Write a bundle-catalogs.yml pointing at *catalog_path* in *project*."""

tests/test_github_http.py

@@ -152,6 +152,44 @@ def failing_open(url, timeout=None, extra_headers=None):
         )
         assert result is None
 
+    def test_metadata_lookup_is_bounded_and_redirect_validated(self):
+        """Release metadata reads stay bounded and use the caller's policy."""
+        captured = {}
+
+        class OversizedResponse:
+            def read(self, amount=None):
+                captured["read_amount"] = amount
+                return b"x" * amount
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *args):
+                return False
+
+        def redirect_validator(old_url, new_url):
+            return None
+
+        def fake_open(
+            url,
+            timeout=None,
+            extra_headers=None,
+            redirect_validator=None,
+        ):
+            captured["redirect_validator"] = redirect_validator
+            return OversizedResponse()
+
+        result = resolve_github_release_asset_api_url(
+            "https://github.com/org/repo/releases/download/v1/pack.zip",
+            fake_open,
+            redirect_validator=redirect_validator,
+            max_metadata_bytes=8,
+        )
+
+        assert result is None
+        assert captured["read_amount"] == 9
+        assert captured["redirect_validator"] is redirect_validator
+
     def test_tag_with_special_characters_is_url_encoded(self):
         """Tags with reserved characters (e.g. '/') are encoded in the API URL."""
         captured_urls = []

tests/test_presets.py

@@ -5016,26 +5016,13 @@ def test_preset_add_from_github_release_url_resolves_and_downloads(self, project
 
         captured_urls = []
 
-        class FakeResponse:
-            def __init__(self, data):
-                self._data = data
-
-            def read(self):
-                return self._data
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                return False
-
         def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
             captured_urls.append((url, extra_headers))
             if "releases/tags/" in url:
-                return FakeResponse(json.dumps({
+                return io.BytesIO(json.dumps({
                     "assets": [{"name": "preset.zip", "url": "https://api.github.com/repos/org/repo/releases/assets/42"}]
                 }).encode())
-            return FakeResponse(zip_bytes)
+            return io.BytesIO(zip_bytes)
 
         runner = CliRunner()
         with patch.object(Path, "cwd", return_value=project_dir), \
@@ -5074,22 +5061,9 @@ def test_preset_add_from_direct_api_asset_url_passes_through(self, project_dir):
 
         captured_urls = []
 
-        class FakeResponse:
-            def __init__(self, data):
-                self._data = data
-
-            def read(self):
-                return self._data
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                return False
-
         def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
             captured_urls.append((url, extra_headers))
-            return FakeResponse(zip_bytes)
+            return io.BytesIO(zip_bytes)
 
         runner = CliRunner()
         with patch.object(Path, "cwd", return_value=project_dir), \
@@ -5131,26 +5105,13 @@ def test_preset_add_from_ghes_release_url_resolves_via_api_v3(self, project_dir,
 
         captured_urls = []
 
-        class FakeResponse:
-            def __init__(self, data):
-                self._data = data
-
-            def read(self):
-                return self._data
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                return False
-
         def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
             captured_urls.append((url, extra_headers))
             if "releases/tags/" in url:
-                return FakeResponse(json.dumps({
+                return io.BytesIO(json.dumps({
                     "assets": [{"name": "preset.zip", "url": "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"}]
                 }).encode())
-            return FakeResponse(zip_bytes)
+            return io.BytesIO(zip_bytes)
 
         runner = CliRunner()
         with patch.object(Path, "cwd", return_value=project_dir), \