fix(presets): re-validate catalog URL after redirects (HTTPS parity/security)

src/specify_cli/presets/__init__.py

@@ -2104,13 +2104,22 @@ def _open_url(
         url: str,
         timeout: int = 10,
         extra_headers: Optional[Dict[str, str]] = None,
+        redirect_validator=None,
     ):
         """Open a URL with provider-based auth, trying each configured provider.
 
         Delegates to :func:`specify_cli.authentication.http.open_url`.
+        *redirect_validator*, when provided, is invoked as ``(old_url, new_url)``
+        before EACH redirect hop, so an HTTPS host guarantee can be enforced on
+        every intermediate URL, not just the terminal one.
         """
         from specify_cli.authentication.http import open_url
-        return open_url(url, timeout, extra_headers=extra_headers)
+        return open_url(
+            url,
+            timeout,
+            extra_headers=extra_headers,
+            redirect_validator=redirect_validator,
+        )
 
     def _resolve_github_release_asset_api_url(
         self,
@@ -2397,7 +2406,21 @@ def _fetch_single_catalog(self, entry: PresetCatalogEntry, force_refresh: bool =
                 pass
 
         try:
-            with self._open_url(entry.url, timeout=10) as response:
+            # Validate EVERY redirect hop (not just the terminal URL): an
+            # https -> http -> attacker-controlled-https chain would pass a
+            # final-URL-only check while the insecure intermediate hop lets a
+            # network attacker rewrite the next redirect. redirect_validator runs
+            # before each hop; the final geturl() check is retained as a
+            # belt-and-braces guard. Mirrors bundler/services/adapters.py.
+            def _validate_redirect(_old_url: str, new_url: str) -> None:
+                self._validate_catalog_url(new_url)
+
+            with self._open_url(
+                entry.url, timeout=10, redirect_validator=_validate_redirect
+            ) as response:
+                final_url = response.geturl()
+                if final_url != entry.url:
+                    self._validate_catalog_url(final_url)
                 catalog_data = json.loads(response.read())
 
             self._validate_catalog_payload(catalog_data, entry.url)
@@ -2548,7 +2571,18 @@ def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]:
                 pass
 
         try:
-            with self._open_url(catalog_url, timeout=10) as response:
+            # Same redirect hardening as _fetch_single_catalog: validate every
+            # redirect hop AND the final URL so this legacy single-catalog path
+            # is not vulnerable to an HTTPS->HTTP redirected payload either.
+            def _validate_redirect(_old_url: str, new_url: str) -> None:
+                self._validate_catalog_url(new_url)
+
+            with self._open_url(
+                catalog_url, timeout=10, redirect_validator=_validate_redirect
+            ) as response:
+                final_url = response.geturl()
+                if final_url != catalog_url:
+                    self._validate_catalog_url(final_url)
                 catalog_data = json.loads(response.read())
 
             # Validate catalog structure. Reuses the same helper as

tests/test_presets.py

@@ -1754,6 +1754,86 @@ def fake_open(req, timeout=None):
 
         assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken"
 
+    def test_fetch_single_catalog_revalidates_redirected_url(self, project_dir):
+        """An HTTPS catalog URL that redirects to http:// must be rejected AFTER
+        the redirect. _open_url follows redirects (auth stripped on downgrade),
+        so without re-validating response.geturl() the http payload would still
+        be fetched and trusted — and it supplies each preset's download_url +
+        sha256, defeating verify_archive_sha256. Parity with the
+        integrations/workflows catalog fetchers."""
+        catalog = PresetCatalog(project_dir)
+
+        class _Resp:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+            def read(self):
+                return json.dumps({"schema_version": "1.0", "presets": {}}).encode()
+
+            def geturl(self):
+                return "http://evil.test/catalog.json"  # downgraded via redirect
+
+        catalog._open_url = lambda url, timeout=None, redirect_validator=None: _Resp()
+
+        entry = PresetCatalogEntry(
+            url="https://good.example/catalog.json",
+            name="c",
+            priority=1,
+            install_allowed=True,
+        )
+        with pytest.raises(PresetValidationError, match="HTTPS"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+
+    def test_fetch_single_catalog_validates_every_redirect_hop(self, project_dir):
+        """A redirect_validator is passed to _open_url and rejects a non-HTTPS
+        INTERMEDIATE hop — closing the https -> http -> attacker-https chain that
+        a terminal-URL-only check would miss."""
+        catalog = PresetCatalog(project_dir)
+        captured = {}
+
+        def fake_open(url, timeout=None, 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")
+
+        catalog._open_url = fake_open
+        entry = PresetCatalogEntry(
+            url="https://good.example/catalog.json",
+            name="c",
+            priority=1,
+            install_allowed=True,
+        )
+        with pytest.raises(PresetValidationError, match="HTTPS"):
+            catalog._fetch_single_catalog(entry, force_refresh=True)
+        assert captured["rv"] is not None
+
+    def test_fetch_catalog_legacy_revalidates_redirected_url(self, project_dir):
+        """The legacy single-catalog fetch_catalog() path also rejects an
+        HTTPS -> http redirected payload (final geturl() check), matching
+        _fetch_single_catalog — it previously parsed the body with no check."""
+        catalog = PresetCatalog(project_dir)
+
+        class _Resp:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+            def read(self):
+                return json.dumps({"schema_version": "1.0", "presets": {}}).encode()
+
+            def geturl(self):
+                return "http://evil.test/catalog.json"
+
+        catalog._open_url = lambda url, timeout=None, redirect_validator=None: _Resp()
+        with pytest.raises(PresetError, match="HTTPS"):
+            catalog.fetch_catalog(force_refresh=True)
+
     @pytest.mark.parametrize(
         "payload",
         [
@@ -1787,6 +1867,10 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, project_dir, paylo
         mock_response.read.return_value = json.dumps(payload).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
+        # A real urllib response reports the final URL (== request URL with no
+        # redirect); the fetcher re-validates it after redirects.
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         entry = PresetCatalogEntry(
             url="https://example.com/catalog.json",
@@ -1856,6 +1940,7 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload(
         mock_response.read.return_value = json.dumps(valid).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = catalog.DEFAULT_CATALOG_URL
 
         entry = PresetCatalogEntry(
             url=catalog.DEFAULT_CATALOG_URL,
@@ -1903,6 +1988,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, project_dir, payload):
         mock_response.read.return_value = json.dumps(payload).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         with patch.object(catalog, "_open_url", return_value=mock_response):
             with pytest.raises(PresetError, match="Invalid preset catalog format"):
@@ -1944,6 +2030,7 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, project_dir):
         mock_response.read.return_value = json.dumps(valid).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         with patch.object(catalog, "_open_url", return_value=mock_response):
             result = catalog.fetch_catalog(force_refresh=False)
@@ -1982,6 +2069,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, project_dir):
         mock_response.read.return_value = json.dumps(valid).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         with patch.object(catalog, "_open_url", return_value=mock_response):
             result = catalog.fetch_catalog(force_refresh=False)
@@ -2052,6 +2140,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, project_dir, monkeypatch):
         mock_response.read.return_value = json.dumps(payload).encode("utf-8")
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         # Record every ``write_text`` call's encoding kwarg so the
         # assertion observes the production writer's argument directly.
@@ -2101,6 +2190,7 @@ def test_fetch_catalog_survives_unwritable_cache(self, project_dir, monkeypatch)
         mock_response.read.return_value = json.dumps(valid).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = catalog.DEFAULT_CATALOG_URL
 
         # Simulate an unwritable cache dir: every write_text under the
         # cache directory raises PermissionError (an OSError subclass).
@@ -2153,6 +2243,7 @@ def test_get_merged_packs_skips_non_mapping_entries(self, project_dir):
         mock_response.read.return_value = json.dumps(payload).encode()
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
+        mock_response.geturl.return_value = "https://example.com/catalog.json"
 
         entry = PresetCatalogEntry(
             url="https://example.com/catalog.json",