fix(workflows): validate every redirect hop when fetching workflow/step catalogs

src/specify_cli/workflows/catalog.py

@@ -523,8 +523,20 @@ def _validate_catalog_url(url: str) -> None:
 
         _validate_catalog_url(entry.url)
 
+        # Validate EVERY redirect hop, not just the final URL: _open_url follows
+        # redirects, so an https:// entry that 30x-redirects through http:// (or
+        # to a non-HTTPS host mid-chain) could otherwise let a network attacker
+        # rewrite the next hop and slip a payload past a final-URL-only check.
+        # redirect_validator runs before each hop; the geturl() check below is
+        # retained as a defense-in-depth backstop. Mirrors the presets/extensions
+        # catalog fix (#3523 / #3524).
+        def _validate_redirect(_old_url: str, new_url: str) -> None:
+            _validate_catalog_url(new_url)
+
         try:
-            with _open_url(entry.url, timeout=30) as resp:
+            with _open_url(
+                entry.url, timeout=30, redirect_validator=_validate_redirect
+            ) as resp:
                 _validate_catalog_url(resp.geturl())
                 data = json.loads(resp.read().decode("utf-8"))
         except Exception as exc:
@@ -1180,8 +1192,20 @@ def _validate_url(url: str) -> None:
 
         _validate_url(entry.url)
 
+        # Validate EVERY redirect hop, not just the final URL: _open_url follows
+        # redirects, so an https:// entry that 30x-redirects through http:// (or
+        # to a non-HTTPS host mid-chain) could otherwise let a network attacker
+        # rewrite the next hop and slip a payload past a final-URL-only check.
+        # redirect_validator runs before each hop; the geturl() check below is
+        # retained as a defense-in-depth backstop. Mirrors the presets/extensions
+        # catalog fix (#3523 / #3524).
+        def _validate_redirect(_old_url: str, new_url: str) -> None:
+            _validate_url(new_url)
+
         try:
-            with _open_url(entry.url, timeout=30) as resp:
+            with _open_url(
+                entry.url, timeout=30, redirect_validator=_validate_redirect
+            ) as resp:
                 _validate_url(resp.geturl())
                 data = json.loads(resp.read().decode("utf-8"))
         except Exception as exc:

tests/test_workflows.py

@@ -6161,7 +6161,9 @@ def geturl(self):
                 return "https://[::1"
 
         monkeypatch.setattr(
-            auth_http, "open_url", lambda url, timeout=30: _FakeResponse()
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(),
         )
 
         catalog = WorkflowCatalog(project_dir)
@@ -6176,6 +6178,41 @@ def geturl(self):
         with pytest.raises(WorkflowCatalogError, match="malformed"):
             catalog._fetch_single_catalog(entry, force_refresh=True)
 
+    def test_fetch_validates_every_redirect_hop(self, project_dir, monkeypatch):
+        """A redirect_validator is passed to open_url and rejects a non-HTTPS
+        INTERMEDIATE hop — closing the https -> http -> attacker-https chain a
+        terminal-URL-only check would miss. Mirrors presets/extensions
+        (#3523 / #3524)."""
+        from specify_cli.workflows.catalog import (
+            WorkflowCatalog,
+            WorkflowCatalogEntry,
+            WorkflowCatalogError,
+        )
+        from specify_cli.authentication import http as auth_http
+
+        captured = {}
+
+        def fake_open(url, timeout=30, redirect_validator=None):
+            captured["rv"] = redirect_validator
+            # Simulate the hop urllib validates before following the redirect.
+            redirect_validator(
+                "https://good.example/catalog.json", "http://evil.test/hop"
+            )
+            raise AssertionError("redirect_validator should have raised")
+
+        monkeypatch.setattr(auth_http, "open_url", fake_open)
+
+        catalog = WorkflowCatalog(project_dir)
+        entry = WorkflowCatalogEntry(
+            url="https://good.example/catalog.json",
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+        with pytest.raises(WorkflowCatalogError, match="HTTPS"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+        assert captured["rv"] is not None
+
     def test_add_catalog(self, project_dir):
         from specify_cli.workflows.catalog import WorkflowCatalog
 
@@ -6720,7 +6757,9 @@ def geturl(self):
                 return "https://[not-an-ip]/x"
 
         monkeypatch.setattr(
-            auth_http, "open_url", lambda url, timeout=30: _FakeResponse()
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(),
         )
 
         catalog = StepCatalog(project_dir)
@@ -6735,6 +6774,41 @@ def geturl(self):
         with pytest.raises(StepCatalogError, match="malformed"):
             catalog._fetch_single_catalog(entry, force_refresh=True)
 
+    def test_fetch_validates_every_redirect_hop(self, project_dir, monkeypatch):
+        """A redirect_validator is passed to open_url and rejects a non-HTTPS
+        INTERMEDIATE hop — closing the https -> http -> attacker-https chain a
+        terminal-URL-only check would miss. Mirrors presets/extensions
+        (#3523 / #3524)."""
+        from specify_cli.workflows.catalog import (
+            StepCatalog,
+            StepCatalogEntry,
+            StepCatalogError,
+        )
+        from specify_cli.authentication import http as auth_http
+
+        captured = {}
+
+        def fake_open(url, timeout=30, redirect_validator=None):
+            captured["rv"] = redirect_validator
+            # Simulate the hop urllib validates before following the redirect.
+            redirect_validator(
+                "https://good.example/steps.json", "http://evil.test/hop"
+            )
+            raise AssertionError("redirect_validator should have raised")
+
+        monkeypatch.setattr(auth_http, "open_url", fake_open)
+
+        catalog = StepCatalog(project_dir)
+        entry = StepCatalogEntry(
+            url="https://good.example/steps.json",
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+        with pytest.raises(StepCatalogError, match="HTTPS"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+        assert captured["rv"] is not None
+
     def test_add_catalog(self, project_dir):
         from specify_cli.workflows.catalog import StepCatalog