Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions src/specify_cli/authentication/http.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)"
Expand Down Expand Up @@ -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)
Expand Down
21 changes: 19 additions & 2 deletions src/specify_cli/bundler/services/adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
68 changes: 68 additions & 0 deletions tests/unit/test_bundler_adapters.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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",
[
Expand Down
Loading