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

src/specify_cli/presets/__init__.py

@@ -2187,6 +2187,16 @@ def _fetch_single_catalog(self, entry: PresetCatalogEntry, force_refresh: bool =
 
         try:
             with self._open_url(entry.url, timeout=10) as response:
+                # Re-validate the URL after any redirects: _open_url follows
+                # redirects (stripping auth only on an HTTPS->HTTP downgrade), so
+                # without this an https:// catalog entry that 30x-redirects to
+                # http://attacker/... would be fetched and trusted. The catalog
+                # payload supplies each preset's download_url + sha256, so a
+                # redirected payload defeats verify_archive_sha256. Mirrors the
+                # integrations/workflows catalog fetchers.
+                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)

tests/test_presets.py

@@ -1754,6 +1754,39 @@ 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: _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)
+
     @pytest.mark.parametrize(
         "payload",
         [
@@ -1787,6 +1820,9 @@ 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)
+        # 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 +1892,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,
@@ -2101,6 +2138,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 +2191,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",