Function open_url

src/specify_cli/authentication/http.py:135–188

Open *url* with config-driven auth, redirect stripping, and fallthrough. 1. Find ``auth.json`` entries whose hosts match the URL. 2. For each entry, resolve the token and try the request. 3. On 401/403 move to the next matching entry. 4. After all entries exhausted (or none matched)

(
    url: str,
    timeout: int = 10,
    extra_headers: dict[str, str] | None = None,
    redirect_validator: RedirectValidator | None = None,
)

def open_url(
    url: str,
    timeout: int = 10,
    extra_headers: dict[str, str] | None = None,
    redirect_validator: RedirectValidator | None = None,
):
    """Open *url* with config-driven auth, redirect stripping, and fallthrough.

    1. Find ``auth.json`` entries whose hosts match the URL.
    2. For each entry, resolve the token and try the request.
    3. On 401/403 move to the next matching entry.
    4. After all entries exhausted (or none matched), try unauthenticated.
    5. Non-auth errors (404, 500, network) raise immediately.

    *extra_headers* (e.g. ``Accept``) are merged into every attempt.
    *redirect_validator*, when provided, is called with ``(old_url, new_url)``
    before following each redirect and may raise to reject the redirect.
    """
    entries = find_entries_for_url(url, _load_config())

    def _make_req(auth_headers: dict[str, str]) -> urllib.request.Request:
        merged = {}
        if extra_headers:
            # Strip Authorization from extra_headers to prevent bypass
            merged.update({k: v for k, v in extra_headers.items() if k.lower() != "authorization"})
        # Auth headers applied last — cannot be overridden by extra_headers
        merged.update(auth_headers)
        return urllib.request.Request(url, headers=merged)

    # Try each matching entry
    for entry in entries:
        provider = get_provider(entry.provider)
        if provider is None:
            continue
        token = provider.resolve_token(entry)
        if not token:
            continue

        req = _make_req(provider.auth_headers(token, entry.auth))
        opener = urllib.request.build_opener(_StripAuthOnRedirect(entry.hosts, redirect_validator))
        try:
            return opener.open(req, timeout=timeout)
        except urllib.error.HTTPError as exc:
            if exc.code in (401, 403):
                exc.close()
                continue  # try next entry
            raise

    # No entry worked (or none matched) — unauthenticated fallback
    req = _make_req({})
    if redirect_validator is not None:
        opener = urllib.request.build_opener(_StripAuthOnRedirect((), redirect_validator))
        return opener.open(req, timeout=timeout)
    return urllib.request.urlopen(req, timeout=timeout)  # noqa: S310