From 00794afa9a75e3fbea2465ef39cbc788e3f08745 Mon Sep 17 00:00:00 2001 From: jawwad-ali Date: Tue, 14 Jul 2026 22:33:07 +0500 Subject: [PATCH] fix(extensions): re-validate catalog URL after redirects (HTTPS parity) ExtensionCatalog._fetch_single_catalog opened the catalog URL and trusted the payload without re-validating response.geturl() after redirects. _open_url follows redirects (stripping auth only on an HTTPS->HTTP downgrade), so an https:// catalog entry that 30x-redirects to http://attacker/... was still fetched and trusted. The payload supplies each extension's download_url + sha256, so a redirected payload can drive install of an arbitrary archive that passes sha256 verification. Add the post-redirect geturl() re-validation via _validate_catalog_url, mirroring integrations/catalog.py, presets, workflows/catalog.py, and bundler adapters. Sibling of the same fix in the presets catalog fetcher. Test: an HTTPS URL whose response.geturl() reports http:// is rejected (ExtensionError). Completed existing fetch-test mocks that predated this behavior to report geturl() like a real urllib response. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/specify_cli/extensions/__init__.py | 10 +++++++ tests/test_extensions.py | 38 ++++++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index e00b9c247a..03518f0349 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -2279,6 +2279,16 @@ def _fetch_single_catalog( # Fetch from network 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 payload + # supplies each extension's download_url + sha256, so a redirected + # payload defeats sha256 verification. Mirrors the + # integrations/presets/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) diff --git a/tests/test_extensions.py b/tests/test_extensions.py index d2f4af8264..36effe4346 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -3650,6 +3650,35 @@ def fake_open(req, timeout=None): assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken" + def test_fetch_single_catalog_revalidates_redirected_url(self, temp_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 extension's download_url + + sha256, defeating sha256 verification. Parity with the + integrations/presets/workflows catalog fetchers.""" + from unittest.mock import patch, MagicMock + + catalog = self._make_catalog(temp_dir) + + mock_response = MagicMock() + mock_response.read.return_value = json.dumps( + {"schema_version": "1.0", "extensions": {}} + ).encode() + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + mock_response.geturl.return_value = "http://evil.test/catalog.json" + + entry = CatalogEntry( + url="https://good.example/catalog.json", + name="c", + priority=1, + install_allowed=True, + ) + with patch.object(catalog, "_open_url", return_value=mock_response): + with pytest.raises(ExtensionError, match="HTTPS"): + catalog._fetch_single_catalog(entry, force_refresh=True) + @pytest.mark.parametrize( "payload", [ @@ -3683,6 +3712,7 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, temp_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" entry = CatalogEntry( url="https://example.com/catalog.json", @@ -3751,6 +3781,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 = "https://example.com/catalog.json" entry = CatalogEntry( url=ExtensionCatalog.DEFAULT_CATALOG_URL, @@ -3798,6 +3829,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, temp_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(ExtensionError, match="Invalid catalog format"): @@ -3838,6 +3870,7 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, temp_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) @@ -3876,6 +3909,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, temp_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) @@ -3950,6 +3984,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, temp_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. @@ -4001,6 +4036,7 @@ def test_fetch_catalog_survives_unwritable_cache(self, temp_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 = "https://example.com/catalog.json" # Simulate an unwritable cache dir: every write_text under the # cache directory raises PermissionError (an OSError subclass). @@ -4052,6 +4088,7 @@ def test_get_merged_extensions_skips_non_mapping_entries(self, temp_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 = CatalogEntry( url="https://example.com/catalog.json", @@ -6030,6 +6067,7 @@ def test_download_extension_allows_bundled_with_url(self, temp_dir): mock_response.read.return_value = b"fake zip data" 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, "get_extension_info", return_value=bundled_with_url), \ patch.object(urllib.request, "urlopen", return_value=mock_response):