diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index 98bba08fd4..5b166ada0b 100644 --- a/src/specify_cli/presets/__init__.py +++ b/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 diff --git a/tests/test_presets.py b/tests/test_presets.py index 797835a88c..8794840380 100644 --- a/tests/test_presets.py +++ b/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",