fix: harden redirect validation and bounded reads

src/specify_cli/_download_security.py

@@ -2,14 +2,17 @@
 
 from __future__ import annotations
 
+import io
+import socket
+from ipaddress import IPv4Address, IPv6Address, ip_address
 from typing import NoReturn, TypeVar
-from urllib.parse import urlparse
+from urllib.parse import ParseResult, urlparse
 
 
 ErrorT = TypeVar("ErrorT", bound=Exception)
 
 MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024
-READ_CHUNK_SIZE = 1024 * 1024
+READ_CHUNK_SIZE = 64 * 1024
 
 # Tighter ceiling for responses that are read fully into memory and parsed as
 # JSON. The 50 MiB MAX_DOWNLOAD_BYTES default is sized for archive/payload
@@ -20,35 +23,156 @@
 # METADATA covers fixed-shape single-object responses (an OAuth token, one
 # release's metadata): a few KiB in practice, 1 MiB is already generous.
 MAX_JSON_METADATA_BYTES = 1 * 1024 * 1024
-_LOOPBACK_HOSTS = frozenset(("localhost", "127.0.0.1", "::1"))
+
+
+def _ip_address_without_scope(
+    hostname: str,
+) -> IPv4Address | IPv6Address | None:
+    """Parse a canonical IP literal, validating an optional IPv6 zone ID."""
+    if "%" in hostname:
+        # Accept only the RFC 6874 ``%25<zone>`` spelling. Other escapes can
+        # alter the IPv6 address when urllib unquotes the authority.
+        address_text, separator, zone = hostname.partition("%25")
+        if (
+            not separator
+            or ":" not in address_text
+            or "%" in address_text
+            or "%" in zone
+        ):
+            return None
+        if not zone or any(
+            not (character.isascii() and (character.isalnum() or character in "._~-"))
+            for character in zone
+        ):
+            return None
+    else:
+        address_text = hostname
+    try:
+        address = ip_address(address_text)
+    except ValueError:
+        return None
+    if "%" in hostname and not isinstance(address, IPv6Address):
+        return None
+    return address
+
+
+def _is_ip_loopback(address: IPv4Address | IPv6Address | None) -> bool:
+    if address is None:
+        return False
+    mapped = getattr(address, "ipv4_mapped", None)
+    return address.is_loopback or bool(mapped and mapped.is_loopback)
+
+
+def _is_ip_local_redirect_target(
+    address: IPv4Address | IPv6Address | None,
+) -> bool:
+    """Treat loopback and unspecified listener aliases as local targets."""
+    if address is None:
+        return False
+    mapped = getattr(address, "ipv4_mapped", None)
+    return _is_ip_loopback(address) or address.is_unspecified or bool(
+        mapped and mapped.is_unspecified
+    )
+
+
+def _parse_url(url: str) -> ParseResult | None:
+    """Parse *url*, rejecting missing hosts and malformed ports."""
+    try:
+        parsed = urlparse(url)
+        hostname = parsed.hostname
+        # Accessing ``port`` performs urllib's range and syntax validation.
+        parsed.port
+    except (TypeError, ValueError):
+        return None
+    if not hostname:
+        return None
+
+    if "%" in hostname:
+        # urllib unquotes reg-name/IPv4 authorities before connecting. Reject
+        # them so encoded dots, characters, ports, or brackets cannot make the
+        # validated hostname differ from the effective target. The only safe
+        # percent form retained is a validated bracketed IPv6 zone ID.
+        if _ip_address_without_scope(hostname) is None:
+            return None
+    elif ":" not in hostname:
+        try:
+            hostname.encode("idna")
+        except UnicodeError:
+            return None
+    return parsed
+
+
+def _is_definite_loopback_host(hostname: str) -> bool:
+    """Recognize only unambiguous hosts that may safely authorize HTTP."""
+    if not hostname.isascii():
+        return False
+    if hostname == "localhost":
+        return True
+    return _is_ip_loopback(_ip_address_without_scope(hostname))
+
+
+def _is_potential_local_target_host(hostname: str) -> bool:
+    """Conservatively classify aliases that could reach a local listener."""
+    if ":" in hostname:
+        return _is_ip_local_redirect_target(_ip_address_without_scope(hostname))
+    try:
+        host = hostname.encode("idna").decode("ascii").lower().removesuffix(".")
+    except UnicodeError:
+        return False
+    if host == "localhost" or host.endswith(".localhost"):
+        return True
+
+    address = _ip_address_without_scope(host)
+    if address is None:
+        # Historical IPv4 spellings are resolver-dependent. They are never
+        # trusted to authorize HTTP, but treating them as potentially local
+        # prevents them from bypassing a remote-to-loopback redirect check.
+        try:
+            address = ip_address(socket.inet_aton(host))
+        except OSError:
+            return False
+    return _is_ip_local_redirect_target(address)
 
 
 def is_loopback_url(url: str) -> bool:
-    """Return whether *url* targets an explicitly allowed loopback host."""
-    return urlparse(url).hostname in _LOOPBACK_HOSTS
+    """Return whether *url* has an unambiguous loopback host."""
+    parsed = _parse_url(url)
+    return parsed is not None and _is_definite_loopback_host(parsed.hostname)
+
+
+def _is_potential_local_target_url(url: str) -> bool:
+    parsed = _parse_url(url)
+    return parsed is not None and _is_potential_local_target_host(parsed.hostname)
 
 
 def is_https_or_localhost_http(url: str) -> bool:
     """Return True if *url* is HTTPS, or HTTP limited to loopback hosts.
 
     Shared scheme-safety predicate used by the auth HTTP redirect handler and
-    by the direct URL validations in the CLI download flows, so the rule (and
-    any future tightening of it) lives in one place.
+    direct URL validations in CLI download flows.
 
     A hostname is always required: a URL without one (e.g. ``https:///x``)
     has no real target and is rejected regardless of scheme.
 
-    The loopback allowance is a deliberate *exact-string* match on
-    ``localhost`` / ``127.0.0.1`` / ``::1``, not an IP-range check: other
-    loopback addresses (e.g. ``127.0.0.2``) are intentionally not covered.
-    ``urlparse`` already lower-cases the hostname, so the comparison is
-    case-insensitive.
+    The HTTP exception is deliberately limited to unambiguous ``localhost``
+    and canonical IPv4/IPv6 loopback literals. Ambiguous numeric, Unicode, and
+    unspecified-address aliases are classified defensively for redirects but
+    never authorize HTTP. No DNS lookup is performed; DNS and hosts-file
+    aliases require connection-level rebinding protection outside this helper.
     """
-    parsed = urlparse(url)
-    if not parsed.hostname:
+    parsed = _parse_url(url)
+    if parsed is None:
         return False
-    is_localhost = parsed.hostname in _LOOPBACK_HOSTS
-    return parsed.scheme == "https" or (parsed.scheme == "http" and is_localhost)
+    return parsed.scheme == "https" or (
+        parsed.scheme == "http" and _is_definite_loopback_host(parsed.hostname)
+    )
+
+
+def is_safe_download_redirect(old_url: str, new_url: str) -> bool:
+    """Return whether a redirect preserves the shared download URL policy."""
+    if not is_https_or_localhost_http(new_url):
+        return False
+    return not _is_potential_local_target_url(new_url) or is_loopback_url(old_url)
 
 
 def _raise(error_type: type[ErrorT], message: str) -> NoReturn:
@@ -75,15 +199,20 @@ def read_response_limited(
     explicit value so the intended bound is pinned at the call site rather than
     tracking changes to the shared default.
     """
-    chunks: list[bytes] = []
+    if isinstance(max_bytes, bool) or not isinstance(max_bytes, int):
+        raise TypeError("max_bytes must be an integer")
+    if max_bytes < 0:
+        raise ValueError("max_bytes must be non-negative")
+
+    output = io.BytesIO()
     total = 0
     limit = max_bytes + 1
     while total < limit:
         chunk = response.read(min(READ_CHUNK_SIZE, limit - total))
         if not chunk:
             break
-        chunks.append(chunk)
         total += len(chunk)
-    if total > max_bytes:
-        _raise(error_type, f"{label} exceeds maximum size of {max_bytes} bytes")
-    return b"".join(chunks)
+        if total > max_bytes:
+            _raise(error_type, f"{label} exceeds maximum size of {max_bytes} bytes")
+        output.write(chunk)
+    return output.getvalue()

src/specify_cli/authentication/azure_devops.py

@@ -22,6 +22,16 @@ class _TokenResponseTooLarge(Exception):
     """Raised when an Azure AD token response exceeds the bounded read limit."""
 
 
+def _extract_token(payload: object, key: str) -> str | None:
+    """Return a normalized token from a JSON object, or None for other shapes."""
+    if not isinstance(payload, dict):
+        return None
+    token = payload.get(key)
+    if not isinstance(token, str):
+        return None
+    return token.strip() or None
+
+
 class AzureDevOpsAuth(AuthProvider):
     """Azure DevOps authentication provider.
 
@@ -79,8 +89,7 @@ def _acquire_via_az_cli() -> str | None:
             if result.returncode != 0:
                 return None
             payload = _json.loads(result.stdout)
-            token = payload.get("accessToken", "").strip()
-            return token or None
+            return _extract_token(payload, "accessToken")
         except (
             OSError,
             subprocess.TimeoutExpired,
@@ -146,8 +155,7 @@ def reject_token_redirect(_old_url: str, new_url: str) -> None:
                         label="Azure DevOps token response",
                     ).decode("utf-8")
                 )
-                token = payload.get("access_token", "").strip()
-                return token or None
+                return _extract_token(payload, "access_token")
         except (
             urllib.error.URLError,
             OSError,

src/specify_cli/authentication/http.py

@@ -17,7 +17,7 @@
 from typing import Callable
 from urllib.parse import urlparse
 
-from .._download_security import is_https_or_localhost_http, is_loopback_url
+from .._download_security import is_safe_download_redirect
 from . import get_provider
 from .config import AuthConfigEntry, _default_config_path, find_entries_for_url, load_auth_config
 
@@ -62,15 +62,11 @@ def _hostname_in_hosts(hostname: str, hosts: tuple[str, ...]) -> bool:
 
 
 def _validate_strict_redirect(old_url: str, new_url: str) -> None:
-    target_is_allowed = is_https_or_localhost_http(new_url)
-    remote_to_http_loopback = (
-        urlparse(new_url).scheme == "http"
-        and not is_loopback_url(old_url)
-    )
-    if not target_is_allowed or remote_to_http_loopback:
+    if not is_safe_download_redirect(old_url, new_url):
         raise urllib.error.URLError(
             f"unsafe redirect to {new_url}: target must use HTTPS with a hostname, "
-            "or stay within localhost over HTTP (127.0.0.1, ::1)"
+            "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)"
         )
 
 
@@ -95,6 +91,8 @@ def __init__(
     def redirect_request(self, req, fp, code, msg, headers, newurl):
         try:
             new_parsed = urlparse(newurl)
+            # Force urllib's syntax and range validation before following.
+            new_parsed.port
         except ValueError as exc:
             # Malformed redirect target (e.g. unterminated IPv6 bracket).
             # Surface as URLError so callers' download error handling applies.
@@ -177,9 +175,11 @@ def open_url(
     *redirect_validator*, when provided, is called with ``(old_url, new_url)``
     before following each redirect and may raise to reject the redirect.
 
+    Every attempt uses an isolated opener so a process-wide opener installed
+    with ``urllib.request.install_opener`` cannot replace the redirect guard.
     Redirect scheme safety: every attempt goes through
     ``_StripAuthOnRedirect``, which rejects redirects to non-HTTPS URLs except
-    HTTP between localhost / 127.0.0.1 / ::1 URLs.
+    HTTP between loopback URLs, and rejects remote-to-local redirects.
     """
     entries = find_entries_for_url(url, _load_config())
 

src/specify_cli/presets/_commands.py

@@ -16,6 +16,10 @@
 from rich.markup import escape as _escape_markup
 
 from .._console import console
+from .._download_security import (
+    is_https_or_localhost_http,
+    is_safe_download_redirect,
+)
 
 preset_app = typer.Typer(
     name="preset",
@@ -102,38 +106,25 @@ def preset_add(
 
         elif from_url:
             # Validate URL scheme before downloading
-            from ipaddress import ip_address
             from urllib.parse import urlparse as _urlparse
 
             try:
                 _parsed = _urlparse(from_url)
+                _parsed.port
             except ValueError:
                 console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(from_url)}")
                 raise typer.Exit(1)
 
-            def _is_allowed_download_url(parsed_url):
-                host = parsed_url.hostname
-                if not host:
-                    return False
-                is_loopback = host == "localhost"
-                if not is_loopback:
-                    try:
-                        is_loopback = ip_address(host).is_loopback
-                    except ValueError:
-                        # Host is not an IP literal (e.g., a regular hostname); treat as non-loopback.
-                        pass
-                return parsed_url.scheme == "https" or (parsed_url.scheme == "http" and is_loopback)
-
             def _validate_download_redirect(old_url, new_url):
-                if not _is_allowed_download_url(_urlparse(new_url)):
+                if not is_safe_download_redirect(old_url, new_url):
                     import urllib.error
 
                     raise urllib.error.URLError(
-                        "redirect target must use HTTPS with a hostname, "
-                        "or HTTP for localhost/loopback"
+                        "redirect target must use HTTPS without entering a local "
+                        "target, or stay within loopback over HTTP"
                     )
 
-            if not _is_allowed_download_url(_parsed):
+            if not is_https_or_localhost_http(from_url):
                 console.print(
                     "[red]Error:[/red] URL must use HTTPS with a hostname, "
                     "or HTTP for localhost/loopback."
@@ -167,7 +158,7 @@ def _validate_download_redirect(old_url, new_url):
                         redirect_validator=_validate_download_redirect,
                     ) as response:
                         final_url = response.geturl() if hasattr(response, "geturl") else from_url
-                        if not _is_allowed_download_url(_urlparse(final_url)):
+                        if not is_https_or_localhost_http(final_url):
                             console.print(
                                 "[red]Error:[/red] Preset URL redirected to a disallowed URL: "
                                 f"{final_url}. Redirect targets must use HTTPS with a hostname, "