Fetch a single catalog, using cache when possible.
(
self, entry: StepCatalogEntry, force_refresh: bool = False
)
| 929 | return False |
| 930 | |
| 931 | def _fetch_single_catalog( |
| 932 | self, entry: StepCatalogEntry, force_refresh: bool = False |
| 933 | ) -> dict[str, Any]: |
| 934 | """Fetch a single catalog, using cache when possible.""" |
| 935 | cache_safe = self._is_cache_path_safe() |
| 936 | cache_file, meta_file = self._get_cache_paths(entry.url) |
| 937 | |
| 938 | if cache_safe and not force_refresh and self._is_url_cache_valid(entry.url): |
| 939 | try: |
| 940 | with open(cache_file, encoding="utf-8") as f: |
| 941 | cached = json.load(f) |
| 942 | if isinstance(cached, dict): |
| 943 | return cached |
| 944 | except (json.JSONDecodeError, OSError): |
| 945 | # Ignore invalid/unreadable cache and fall back to fetching from source. |
| 946 | pass |
| 947 | |
| 948 | from urllib.parse import urlparse |
| 949 | from specify_cli.authentication.http import open_url as _open_url |
| 950 | |
| 951 | def _validate_url(url: str) -> None: |
| 952 | parsed = urlparse(url) |
| 953 | is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") |
| 954 | if parsed.scheme != "https" and not ( |
| 955 | parsed.scheme == "http" and is_localhost |
| 956 | ): |
| 957 | raise StepCatalogError( |
| 958 | f"Refusing to fetch catalog from non-HTTPS URL: {url}" |
| 959 | ) |
| 960 | if not parsed.hostname: |
| 961 | raise StepCatalogError( |
| 962 | f"Refusing to fetch catalog from URL with no hostname: {url}" |
| 963 | ) |
| 964 | |
| 965 | _validate_url(entry.url) |
| 966 | |
| 967 | try: |
| 968 | with _open_url(entry.url, timeout=30) as resp: |
| 969 | _validate_url(resp.geturl()) |
| 970 | data = json.loads(resp.read().decode("utf-8")) |
| 971 | except Exception as exc: |
| 972 | if cache_safe and cache_file.exists(): |
| 973 | try: |
| 974 | with open(cache_file, encoding="utf-8") as f: |
| 975 | cached = json.load(f) |
| 976 | if isinstance(cached, dict): |
| 977 | return cached |
| 978 | except (json.JSONDecodeError, ValueError, OSError): |
| 979 | # Stale-cache read failed; let the original fetch error propagate. |
| 980 | pass |
| 981 | raise StepCatalogError( |
| 982 | f"Failed to fetch catalog from {entry.url}: {exc}" |
| 983 | ) from exc |
| 984 | |
| 985 | if not isinstance(data, dict): |
| 986 | raise StepCatalogError( |
| 987 | f"Catalog from {entry.url} is not a valid JSON object." |
| 988 | ) |