fix(workflows): validate download redirects before following them

All three workflow download sites (add --from, catalog install, step
install) passed no redirect_validator to open_url, so an HTTPS URL
redirecting to cleartext HTTP issued the insecure request before the
post-hoc geturl() check reported it. Shared validator now rejects
non-HTTPS redirects (loopback HTTP allowed) pre-follow, matching the
preset download path.

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

src/specify_cli/workflows/_commands.py

@@ -108,6 +108,26 @@ _WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
 _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"})
 
 
+def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None:
+    """Reject a redirect before it is followed unless HTTPS (or loopback HTTP)."""
+    import urllib.error
+    from ipaddress import ip_address
+    from urllib.parse import urlparse
+
+    parsed = urlparse(new_url)
+    host = parsed.hostname or ""
+    loopback = host == "localhost"
+    if not loopback:
+        try:
+            loopback = ip_address(host).is_loopback
+        except ValueError:
+            pass
+    if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback):
+        raise urllib.error.URLError(
+            "redirect target must use HTTPS, or HTTP for localhost/loopback"
+        )
+
+
 def _validate_workflow_id_or_exit(workflow_id: str) -> None:
     """Validate that ``workflow_id`` is a safe installed-workflow directory name."""
     if (
@@ -756,7 +776,12 @@ def workflow_add(
 
         import tempfile
         try:
-            with _open_url(download_url, timeout=30, extra_headers=_wf_url_extra_headers) as resp:
+            with _open_url(
+                download_url,
+                timeout=30,
+                extra_headers=_wf_url_extra_headers,
+                redirect_validator=_reject_insecure_download_redirect,
+            ) as resp:
                 final_url = resp.geturl()
                 final_parsed = urlparse(final_url)
                 final_host = final_parsed.hostname or ""
@@ -903,7 +928,12 @@ def _install_workflow_from_catalog(
             _wf_cat_extra_headers = {"Accept": "application/octet-stream"}
 
         workflow_dir.mkdir(parents=True, exist_ok=True)
-        with _open_url(workflow_url, timeout=30, extra_headers=_wf_cat_extra_headers) as response:
+        with _open_url(
+            workflow_url,
+            timeout=30,
+            extra_headers=_wf_cat_extra_headers,
+            redirect_validator=_reject_insecure_download_redirect,
+        ) as response:
             # Validate final URL after redirects
             final_url = response.geturl()
             final_parsed = urlparse(final_url)
@@ -1621,7 +1651,9 @@ def workflow_step_add(
             raise ValueError(f"Refusing to fetch from non-HTTPS URL: {url}")
         if not parsed.hostname:
             raise ValueError(f"Refusing to fetch from URL with no hostname: {url}")
-        with _open_url(url, timeout=30) as resp:
+        with _open_url(
+            url, timeout=30, redirect_validator=_reject_insecure_download_redirect
+        ) as resp:
             final_url = resp.geturl()
             final_parsed = urlparse(final_url)
             final_is_localhost = final_parsed.hostname in ("localhost", "127.0.0.1", "::1")

tests/test_workflows.py

@@ -6579,7 +6579,7 @@ steps:
             def __exit__(self, *a):
                 return False
 
-        def fake_open_url(url, timeout=None, extra_headers=None):
+        def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
             captured_urls.append((url, extra_headers, timeout))
             if "releases/tags/" in url:
                 return FakeResponse(json.dumps({
@@ -6631,7 +6631,7 @@ steps:
             def __exit__(self, *a):
                 return False
 
-        def fake_open_url(url, timeout=None, extra_headers=None):
+        def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
             captured_urls.append((url, extra_headers))
             return FakeResponse(self.VALID_WORKFLOW_YAML.encode())
 
@@ -6674,7 +6674,7 @@ steps:
             def __exit__(self, *a):
                 return False
 
-        def fake_open_url(url, timeout=None, extra_headers=None):
+        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({
@@ -6750,7 +6750,7 @@ steps:
             def __exit__(self, *a):
                 return False
 
-        def fake_open_url(url, timeout=None, extra_headers=None):
+        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({
@@ -6818,7 +6818,7 @@ steps:
     run: "echo hello"
 """
 
-        def fake_open_url(url, timeout=None, extra_headers=None):
+        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({
@@ -7364,7 +7364,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
         ):
             result = runner.invoke(
                 app,
@@ -7384,7 +7384,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
         ):
             result = runner.invoke(
                 app,
@@ -7418,7 +7418,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", redirected_url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(b"", redirected_url),
         ):
             result = runner.invoke(
                 app,
@@ -7436,7 +7436,7 @@ steps:
         monkeypatch.chdir(project_dir)
         calls: list[str] = []
 
-        def _fake_open(url, timeout=None, extra_headers=None):
+        def _fake_open(url, timeout=None, extra_headers=None, redirect_validator=None):
             calls.append(url)
             raise AssertionError(f"network fetch attempted: {url}")
 
@@ -7601,6 +7601,51 @@ steps:
         assert result.exception is None or isinstance(result.exception, SystemExit)
         assert "No workflow.yml found" in result.output
 
+    def test_download_redirect_validator_rejects_http_before_follow(self):
+        import urllib.error
+
+        from specify_cli.workflows._commands import _reject_insecure_download_redirect
+
+        with pytest.raises(urllib.error.URLError):
+            _reject_insecure_download_redirect(
+                "https://example.com/wf.yml", "http://evil.example.com/wf.yml"
+            )
+        # Allowed: HTTPS anywhere, HTTP on loopback.
+        _reject_insecure_download_redirect(
+            "https://example.com/wf.yml", "https://cdn.example.com/wf.yml"
+        )
+        _reject_insecure_download_redirect(
+            "https://example.com/wf.yml", "http://localhost:8000/wf.yml"
+        )
+        _reject_insecure_download_redirect(
+            "https://example.com/wf.yml", "http://127.0.0.1/wf.yml"
+        )
+
+    def test_add_from_url_passes_redirect_validator(self, project_dir, monkeypatch):
+        from unittest.mock import patch
+
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(project_dir)
+        data = self.WORKFLOW_YAML.format(version="1.0.0").encode()
+        seen: dict[str, object] = {}
+
+        def fake_open(url, timeout=None, extra_headers=None, redirect_validator=None):
+            seen["validator"] = redirect_validator
+            return self._FakeResponse(data, url)
+
+        runner = CliRunner()
+        with patch("specify_cli.authentication.http.open_url", side_effect=fake_open):
+            result = runner.invoke(
+                app,
+                ["workflow", "add", "align-wf", "--from", "https://example.com/workflow.yml"],
+            )
+        assert result.exit_code == 0, result.output
+        from specify_cli.workflows._commands import _reject_insecure_download_redirect
+
+        assert seen["validator"] is _reject_insecure_download_redirect
+
     def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch):
         """A failed dump must not truncate the persisted registry."""
         from specify_cli.workflows.catalog import WorkflowRegistry
@@ -7712,7 +7757,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
         ):
             result = runner.invoke(app, ["workflow", "update"], input="y\n")
         assert result.exit_code == 0, result.output
@@ -7760,7 +7805,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(b"", url),
         ), patch.object(
             WorkflowDefinition,
             "from_yaml",
@@ -7901,7 +7946,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
         ):
             result = runner.invoke(app, ["workflow", "update"], input="y\n")
         assert "does not match the catalog version" in result.output
@@ -7939,7 +7984,7 @@ steps:
         runner = CliRunner()
         with patch(
             "specify_cli.authentication.http.open_url",
-            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+            side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
         ):
             result = runner.invoke(app, ["workflow", "update"], input="y\n")
         assert result.exit_code == 0, result.output
@@ -8187,7 +8232,7 @@ steps:
             },
         )
 
-        def boom(url, timeout=None, extra_headers=None):
+        def boom(url, timeout=None, extra_headers=None, redirect_validator=None):
             raise OSError("network down")
 
         runner = CliRunner()