From fd8b7844c8d706de989b39c0b677e56ccc85f6f2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 16 Sep 2026 15:44:42 +0000 Subject: [PATCH] fix bundler catalog fallback handling Co-authored-by: markuswondrak <245696895+markuswondrak@users.noreply.github.com> --- src/specify_cli/authentication/http.py | 8 ++- src/specify_cli/bundler/services/adapters.py | 21 +++++- tests/unit/test_bundler_adapters.py | 68 ++++++++++++++++++++ 3 files changed, 93 insertions(+), 4 deletions(-) diff --git a/src/specify_cli/authentication/http.py b/src/specify_cli/authentication/http.py index d200bf9258..32a6ed67c7 100644 --- a/src/specify_cli/authentication/http.py +++ b/src/specify_cli/authentication/http.py @@ -65,9 +65,13 @@ def _hostname_in_hosts(hostname: str, hosts: tuple[str, ...]) -> bool: RedirectValidator = Callable[[str, str], None] +class RedirectPolicyError(urllib.error.URLError): + """A redirect rejected because it violates the client's security policy.""" + + def _validate_strict_redirect(old_url: str, new_url: str) -> None: if not is_safe_download_redirect(old_url, new_url): - raise urllib.error.URLError( + raise RedirectPolicyError( f"unsafe redirect to {new_url}: target must use HTTPS with a hostname, " "must not enter a local target from a remote host, and may use HTTP only " "within loopback (for example localhost, 127.0.0.1, ::1)" @@ -100,7 +104,7 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): except ValueError as exc: # Malformed redirect target (e.g. unterminated IPv6 bracket). # Surface as URLError so callers' download error handling applies. - raise urllib.error.URLError(f"malformed redirect URL: {exc}") from exc + raise RedirectPolicyError(f"malformed redirect URL: {exc}") from exc if self._redirect_validator is not None: self._redirect_validator(req.full_url, newurl) diff --git a/src/specify_cli/bundler/services/adapters.py b/src/specify_cli/bundler/services/adapters.py index ca39a2489b..c2f9a7e682 100644 --- a/src/specify_cli/bundler/services/adapters.py +++ b/src/specify_cli/bundler/services/adapters.py @@ -11,6 +11,7 @@ from __future__ import annotations import re +import urllib.error from pathlib import Path from urllib.parse import ParseResult, urlparse from urllib.request import url2pathname @@ -39,6 +40,11 @@ } HTTP_TIMEOUT_SECONDS = 10 +_TRANSIENT_HTTP_STATUS_CODES = (408, 429) + + +class _CatalogUnavailable(Exception): + """A catalog could not be fetched because its remote service is unavailable.""" # Windows absolute paths like ``C:\catalog.json`` parse with a single-letter # ``scheme`` under urlparse; treat them as local files rather than URLs. @@ -134,7 +140,10 @@ def fetch(source: CatalogSource) -> dict: if scheme == "builtin": if url == "builtin://community": if allow_network: - return _http_get_json(source.id, COMMUNITY_CATALOG_URL) + try: + return _http_get_json(source.id, COMMUNITY_CATALOG_URL) + except _CatalogUnavailable: + return _load_packaged_community_catalog() return _load_packaged_community_catalog() payload = _BUILTIN_CATALOGS.get(url) if payload is None: @@ -178,7 +187,7 @@ def _http_get_json(source_id: str, url: str) -> dict: HTTPS/host guarantee from ``_validate_remote_url`` is preserved end to end rather than only on the initial URL. """ - from ...authentication.http import open_url + from ...authentication.http import RedirectPolicyError, open_url def _validate_redirect(_old_url: str, new_url: str) -> None: _validate_remote_url(source_id, new_url) @@ -199,6 +208,14 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: ).decode("utf-8") except BundlerError: raise + except RedirectPolicyError as exc: + raise BundlerError(f"Failed to fetch catalog from {url}: {exc}") from exc + except urllib.error.HTTPError as exc: + if exc.code in _TRANSIENT_HTTP_STATUS_CODES or exc.code >= 500: + raise _CatalogUnavailable() from exc + raise BundlerError(f"Failed to fetch catalog from {url}: {exc}") from exc + except (urllib.error.URLError, OSError) as exc: + raise _CatalogUnavailable() from exc except Exception as exc: # noqa: BLE001 raise BundlerError(f"Failed to fetch catalog from {url}: {exc}") from exc return loads_json(raw, origin=final_url) diff --git a/tests/unit/test_bundler_adapters.py b/tests/unit/test_bundler_adapters.py index 854e60df3f..0c2aa06e3a 100644 --- a/tests/unit/test_bundler_adapters.py +++ b/tests/unit/test_bundler_adapters.py @@ -1,8 +1,11 @@ """Unit tests for catalog-fetch adapters (auth + redirect safety).""" from __future__ import annotations +import urllib.error + import pytest +from specify_cli.authentication.http import RedirectPolicyError from specify_cli.bundler import BundlerError from specify_cli.bundler.models.catalog import CatalogSource, InstallPolicy from specify_cli.bundler.services import adapters @@ -158,6 +161,71 @@ def test_builtin_community_catalog_uses_core_pack_snapshot_offline( assert "packaged" in result["bundles"] +@pytest.mark.parametrize("status_code", [408, 429, 500]) +def test_builtin_community_catalog_falls_back_for_transient_http_failures( + monkeypatch, tmp_path, status_code +): + catalog_path = tmp_path / "bundles" / "catalog.community.json" + catalog_path.parent.mkdir() + catalog_path.write_text( + '{"schema_version":"1.0","bundles":{}}}', encoding="utf-8" + ) + monkeypatch.setattr(adapters, "_locate_core_pack", lambda: tmp_path) + + def fail(url, timeout=10, extra_headers=None, redirect_validator=None): + raise urllib.error.HTTPError(url, status_code, "transient", {}, None) + + monkeypatch.setattr("specify_cli.authentication.http.open_url", fail) + fetcher = adapters.make_catalog_fetcher(allow_network=True) + + assert fetcher(_source("builtin://community")) == { + "schema_version": "1.0", + "bundles": {}, + } + + +def test_builtin_community_catalog_falls_back_for_transport_errors(monkeypatch, tmp_path): + catalog_path = tmp_path / "bundles" / "catalog.community.json" + catalog_path.parent.mkdir() + catalog_path.write_text( + '{"schema_version":"1.0","bundles":{}}}', encoding="utf-8" + ) + monkeypatch.setattr(adapters, "_locate_core_pack", lambda: tmp_path) + + def fail(url, timeout=10, extra_headers=None, redirect_validator=None): + raise urllib.error.URLError("network unreachable") + + monkeypatch.setattr("specify_cli.authentication.http.open_url", fail) + fetcher = adapters.make_catalog_fetcher(allow_network=True) + + assert fetcher(_source("builtin://community")) == { + "schema_version": "1.0", + "bundles": {}, + } + + +@pytest.mark.parametrize( + "error", + [ + RedirectPolicyError("unsafe redirect"), + RedirectPolicyError("malformed redirect URL"), + ], +) +def test_builtin_community_catalog_does_not_fall_back_for_redirect_policy_errors( + monkeypatch, tmp_path, error +): + monkeypatch.setattr(adapters, "_locate_core_pack", lambda: tmp_path) + + def fail(url, timeout=10, extra_headers=None, redirect_validator=None): + raise error + + monkeypatch.setattr("specify_cli.authentication.http.open_url", fail) + fetcher = adapters.make_catalog_fetcher(allow_network=True) + + with pytest.raises(BundlerError, match="Failed to fetch catalog"): + fetcher(_source("builtin://community")) + + @pytest.mark.parametrize( "url", [