diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 05aa35f7fb..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,7 +2313,24 @@ def _fetch_single_catalog( # Fetch from network try: - with self._open_url(entry.url, timeout=10) as response: + # 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) catalog_data = json.loads(response.read()) self._validate_catalog_payload(catalog_data, entry.url) @@ -2481,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 d2f4af8264..d3f1821245 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -3650,6 +3650,77 @@ 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) + + 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", [ @@ -3683,6 +3754,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 +3823,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 +3871,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 +3912,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 +3951,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 +4026,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 +4078,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 +4130,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 +6109,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):