Fetch catalog JSON over HTTP(S) via the shared authenticated client. Routing through :func:`specify_cli.authentication.http.open_url` gives ``auth.json`` token support and strips the ``Authorization`` header when a redirect leaves the entry's trusted hosts or downgrades the scheme. We a
(source_id: str, url: str)
| 126 | |
| 127 | |
| 128 | def _http_get_json(source_id: str, url: str) -> dict: |
| 129 | """Fetch catalog JSON over HTTP(S) via the shared authenticated client. |
| 130 | |
| 131 | Routing through :func:`specify_cli.authentication.http.open_url` gives |
| 132 | ``auth.json`` token support and strips the ``Authorization`` header when a |
| 133 | redirect leaves the entry's trusted hosts or downgrades the scheme. We also |
| 134 | reject any redirect that leaves HTTPS (the ``redirect_validator`` runs |
| 135 | *before* each hop) and re-validate the final URL after redirects, so the |
| 136 | HTTPS/host guarantee from ``_validate_remote_url`` is preserved end to end |
| 137 | rather than only on the initial URL. |
| 138 | """ |
| 139 | from ...authentication.http import open_url |
| 140 | |
| 141 | def _validate_redirect(_old_url: str, new_url: str) -> None: |
| 142 | _validate_remote_url(source_id, new_url) |
| 143 | |
| 144 | try: |
| 145 | with open_url( |
| 146 | url, |
| 147 | timeout=HTTP_TIMEOUT_SECONDS, |
| 148 | redirect_validator=_validate_redirect, |
| 149 | ) as response: |
| 150 | final_url = response.geturl() |
| 151 | _validate_remote_url(source_id, final_url) |
| 152 | raw = response.read().decode("utf-8") |
| 153 | except BundlerError: |
| 154 | raise |
| 155 | except Exception as exc: # noqa: BLE001 |
| 156 | raise BundlerError(f"Failed to fetch catalog from {url}: {exc}") from exc |
| 157 | return loads_json(raw, origin=final_url) |
| 158 | |
| 159 | |
| 160 | class DefaultPrimitiveInstaller: |
no test coverage detected