From b4d16578e870222dfc3112e2261d9a16a93dc360 Mon Sep 17 00:00:00 2001 From: jawwad-ali Date: Tue, 14 Jul 2026 22:33:07 +0500 Subject: [PATCH 1/2] 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 05aa35f7fb..3ead300cb9 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -2305,6 +2305,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): From 8b7fdacf5a33d1e89273709dedd81b14359aba91 Mon Sep 17 00:00:00 2001 From: jawwad-ali Date: Fri, 17 Jul 2026 15:53:20 +0500 Subject: [PATCH 2/2] fix(extensions): correct auth-strip comment; validate every redirect hop + guard legacy fetch_catalog - Correct the comment: _StripAuthOnRedirect strips auth not only on an HTTPS->HTTP downgrade but also whenever the redirect leaves the configured trusted hosts. The comment now describes both cases. - Parity with the presets fix: validate EVERY redirect hop (not just the terminal URL) so an https -> http -> attacker-https chain can't slip a redirected payload past the final-URL check. _open_url forwards a redirect_validator to open_url; _fetch_single_catalog passes _validate_catalog_url through it while keeping the final geturl() check. - Give the legacy public fetch_catalog() single-catalog path the same redirect_validator + final geturl() validation (it previously parsed the body with no redirect check). Tests: an intermediate http hop is rejected, and the legacy fetch_catalog() rejects an HTTPS->http redirected payload (both fail before). Full test_extensions.py (356) green. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/specify_cli/extensions/__init__.py | 47 ++++++++++++++++++++------ tests/test_extensions.py | 42 +++++++++++++++++++++++ 2 files changed, 79 insertions(+), 10 deletions(-) diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 3ead300cb9..62a004fe8b 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -2073,14 +2073,23 @@ 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, @@ -2304,14 +2313,21 @@ 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. + # Validate EVERY redirect hop, not just the terminal URL. _open_url + # follows redirects; _StripAuthOnRedirect drops auth on an HTTPS->HTTP + # downgrade AND whenever the redirect leaves the configured trusted + # hosts, but the payload itself is still fetched and trusted, and it + # supplies each extension's download_url + sha256 (so a redirected + # payload defeats sha256 verification). A terminal-only check also + # misses an https -> http -> attacker-https chain. redirect_validator + # runs before each hop; the final geturl() check is kept 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) @@ -2491,7 +2507,18 @@ def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]: try: import urllib.error - 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 diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 36effe4346..d3f1821245 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -3679,6 +3679,48 @@ def test_fetch_single_catalog_revalidates_redirected_url(self, temp_dir): with pytest.raises(ExtensionError, match="HTTPS"): catalog._fetch_single_catalog(entry, force_refresh=True) + def test_fetch_single_catalog_validates_every_redirect_hop(self, temp_dir): + """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.""" + catalog = self._make_catalog(temp_dir) + captured = {} + + def fake_open(url, timeout=None, extra_headers=None, redirect_validator=None): + captured["rv"] = redirect_validator + redirect_validator("https://good.example/catalog.json", "http://evil.test/hop") + raise AssertionError("redirect_validator should have raised") + + catalog._open_url = fake_open + entry = CatalogEntry( + url="https://good.example/catalog.json", + name="c", + priority=1, + install_allowed=True, + ) + with pytest.raises(ExtensionError, 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, temp_dir): + """The legacy single-catalog fetch_catalog() path also rejects an + HTTPS -> http redirected payload (final geturl() check) — it previously + parsed the body with no redirect check.""" + 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" + + with patch.object(catalog, "_open_url", return_value=mock_response): + with pytest.raises(ExtensionError, match="HTTPS"): + catalog.fetch_catalog(force_refresh=True) + @pytest.mark.parametrize( "payload", [