From 024aac7c3dd1cefb0c19956fdcaf426b0bb19bd3 Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 22 Jul 2026 18:28:25 +0200 Subject: [PATCH] fix: harden bounded reads and redirect validation Assisted-by: OpenAI Codex (model: GPT-5, autonomous) --- src/specify_cli/_download_security.py | 171 +++++++++++++-- .../authentication/azure_devops.py | 16 +- src/specify_cli/authentication/http.py | 18 +- src/specify_cli/presets/_commands.py | 29 +-- src/specify_cli/workflows/_commands.py | 89 ++------ tests/test_authentication.py | 195 +++++++++++++++++- tests/test_download_security.py | 163 ++++++++++++++- tests/test_self_upgrade_guidance.py | 1 + tests/test_upgrade.py | 11 +- tests/test_workflows.py | 7 + 10 files changed, 557 insertions(+), 143 deletions(-) diff --git a/src/specify_cli/_download_security.py b/src/specify_cli/_download_security.py index 0d39baca03..b4f6d28add 100644 --- a/src/specify_cli/_download_security.py +++ b/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`` 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() diff --git a/src/specify_cli/authentication/azure_devops.py b/src/specify_cli/authentication/azure_devops.py index c475fa6fda..907f1b5e24 100644 --- a/src/specify_cli/authentication/azure_devops.py +++ b/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, diff --git a/src/specify_cli/authentication/http.py b/src/specify_cli/authentication/http.py index 532779c1ad..aa643c908e 100644 --- a/src/specify_cli/authentication/http.py +++ b/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()) diff --git a/src/specify_cli/presets/_commands.py b/src/specify_cli/presets/_commands.py index ed8b937600..ba2b85b351 100644 --- a/src/specify_cli/presets/_commands.py +++ b/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, " diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index be8b731d51..459d0e7404 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -20,6 +20,10 @@ from rich.markup import escape as _escape_markup from .._console import console, err_console +from .._download_security import ( + is_https_or_localhost_http, + is_safe_download_redirect, +) from .._project import _resolve_init_dir_override workflow_app = typer.Typer( @@ -383,27 +387,12 @@ def ownership_for(candidate: Path) -> tuple[Path, str] | None: def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None: """Reject insecure redirects before they are followed.""" import urllib.error - from ipaddress import ip_address - from urllib.parse import urlparse - def _is_loopback_http(url: str) -> bool: - parsed = urlparse(url) - if parsed.scheme != "http": - return False - host = parsed.hostname or "" - if host == "localhost": - return True - try: - return ip_address(host).is_loopback - except ValueError: - return False - - if urlparse(new_url).scheme == "https": - return - if _is_loopback_http(old_url) and _is_loopback_http(new_url): + if is_safe_download_redirect(old_url, new_url): return raise urllib.error.URLError( - "redirect target must use HTTPS; loopback HTTP may only redirect from loopback HTTP" + "redirect target must use HTTPS without entering a local target; " + "loopback HTTP may only redirect from another loopback URL" ) @@ -1579,24 +1568,15 @@ def _validate_and_install_local( else (source if source.startswith(("http://", "https://")) else None) ) if download_url is not None: - from ipaddress import ip_address from urllib.parse import urlparse from specify_cli.authentication.http import open_url as _open_url try: - parsed_src = urlparse(download_url) + urlparse(download_url).port except ValueError: console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(download_url)}") raise typer.Exit(1) - src_host = parsed_src.hostname or "" - src_loopback = src_host == "localhost" - if not src_loopback: - try: - src_loopback = ip_address(src_host).is_loopback - except ValueError: - # Host is not an IP literal (e.g., a DNS name); keep default non-loopback. - pass - if parsed_src.scheme != "https" and not (parsed_src.scheme == "http" and src_loopback): + if not is_https_or_localhost_http(download_url): console.print("[red]Error:[/red] Only HTTPS URLs are allowed, except HTTP for localhost.") raise typer.Exit(1) @@ -1647,16 +1627,7 @@ def _validate_and_install_local( redirect_validator=_reject_insecure_download_redirect, ) as resp: final_url = resp.geturl() - final_parsed = urlparse(final_url) - final_host = final_parsed.hostname or "" - final_lb = final_host == "localhost" - if not final_lb: - try: - final_lb = ip_address(final_host).is_loopback - except ValueError: - # Redirect host is not an IP literal; keep loopback as determined above. - pass - if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_lb): + if not is_https_or_localhost_http(final_url): console.print( f"[red]Error:[/red] URL redirected to non-HTTPS: {_escape_markup(final_url)}" ) @@ -1788,27 +1759,17 @@ def versions_match(actual: object, expected: str) -> bool: raise typer.Exit(1) # Validate URL scheme (HTTPS required, HTTP allowed for localhost only) - from ipaddress import ip_address from urllib.parse import urlparse try: parsed_url = urlparse(workflow_url) - url_host = parsed_url.hostname or "" + parsed_url.port except ValueError: console.print( f"[red]Error:[/red] Workflow '{safe_wf_id}' has a malformed install URL." ) raise typer.Exit(1) - is_loopback = False - if url_host == "localhost": - is_loopback = True - else: - try: - is_loopback = ip_address(url_host).is_loopback - except ValueError: - # Host is not an IP literal (e.g., a regular hostname); treat as non-loopback. - pass - if parsed_url.scheme != "https" and not (parsed_url.scheme == "http" and is_loopback): + if not is_https_or_localhost_http(workflow_url): console.print( f"[red]Error:[/red] Workflow '{safe_wf_id}' has an invalid install URL. " "Only HTTPS URLs are allowed, except HTTP for localhost/loopback." @@ -1862,16 +1823,7 @@ def versions_match(actual: object, expected: str) -> bool: ) as response: # Validate final URL after redirects final_url = response.geturl() - final_parsed = urlparse(final_url) - final_host = final_parsed.hostname or "" - final_loopback = final_host == "localhost" - if not final_loopback: - try: - final_loopback = ip_address(final_host).is_loopback - except ValueError: - # Host is not an IP literal (e.g., a regular hostname); treat as non-loopback. - pass - if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_loopback): + if not is_https_or_localhost_http(final_url): _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before) console.print( f"[red]Error:[/red] Workflow '{safe_wf_id}' redirected to non-HTTPS URL: {_escape_markup(final_url)}" @@ -2694,28 +2646,17 @@ def workflow_step_add( ) raise typer.Exit(1) - from urllib.parse import urlparse from specify_cli.authentication.http import open_url as _open_url def _safe_fetch(url: str) -> bytes: - parsed = urlparse(url) - is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") - if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost): + if not is_https_or_localhost_http(url): raise ValueError(f"Refusing to fetch from non-HTTPS URL: {url}") - if not parsed.hostname: - raise ValueError(f"Refusing to fetch from URL with no hostname: {url}") with _open_url( url, timeout=30, redirect_validator=_reject_insecure_download_redirect ) as resp: final_url = resp.geturl() - final_parsed = urlparse(final_url) - final_is_localhost = final_parsed.hostname in ("localhost", "127.0.0.1", "::1") - if final_parsed.scheme != "https" and not ( - final_parsed.scheme == "http" and final_is_localhost - ): + if not is_https_or_localhost_http(final_url): raise ValueError(f"Redirect to non-HTTPS URL: {final_url}") - if not final_parsed.hostname: - raise ValueError(f"Redirect to URL with no hostname: {final_url}") return _read_response_within_limit(resp) _validate_step_id_or_exit(step_id) diff --git a/tests/test_authentication.py b/tests/test_authentication.py index 727a9a37db..06523c6f26 100644 --- a/tests/test_authentication.py +++ b/tests/test_authentication.py @@ -516,6 +516,23 @@ def test_resolve_token_azure_cli_undecodable_output_returns_none(self): with patch("specify_cli.authentication.azure_devops.subprocess.run", side_effect=boom): assert AzureDevOpsAuth().resolve_token(entry) is None + @pytest.mark.parametrize("payload", [[], {"accessToken": None}, {"accessToken": 123}]) + def test_resolve_token_azure_cli_unexpected_json_shape_returns_none( + self, payload + ): + from unittest.mock import MagicMock, patch + + entry = AuthConfigEntry( + hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli", + ) + result = MagicMock(returncode=0, stdout=json.dumps(payload)) + + with patch( + "specify_cli.authentication.azure_devops.subprocess.run", + return_value=result, + ): + assert AzureDevOpsAuth().resolve_token(entry) is None + def test_resolve_token_azure_ad_success(self, monkeypatch): """azure-ad acquires token via OAuth2 client credentials.""" from unittest.mock import patch, MagicMock @@ -559,7 +576,13 @@ def test_resolve_token_azure_ad_network_error_returns_none(self, monkeypatch): with patch("urllib.request.build_opener", return_value=mock_opener): assert AzureDevOpsAuth().resolve_token(entry) is None - def test_resolve_token_azure_ad_rejects_https_redirect(self, monkeypatch): + @pytest.mark.parametrize( + ("status", "reason"), + [(307, "Temporary Redirect"), (308, "Permanent Redirect")], + ) + def test_resolve_token_azure_ad_rejects_https_redirect( + self, monkeypatch, status, reason + ): """The client-secret POST must never be redirected to another host.""" import urllib.error from unittest.mock import MagicMock, patch @@ -577,17 +600,72 @@ def test_resolve_token_azure_ad_rejects_https_redirect(self, monkeypatch): assert AzureDevOpsAuth().resolve_token(entry) is None redirect_handler = build_opener.call_args.args[0] - request = Request("https://login.microsoftonline.com/tid/oauth2/v2.0/token") + request = Request( + "https://login.microsoftonline.com/tid/oauth2/v2.0/token", + data=b"grant_type=client_credentials&client_secret=secret-value", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + assert request.get_method() == "POST" + assert b"client_secret=secret-value" in request.data with pytest.raises(urllib.error.URLError, match="must not be redirected"): redirect_handler.redirect_request( request, io.BytesIO(b""), - 307, - "Temporary Redirect", + status, + reason, {}, "https://evil.example/token", ) + def test_resolve_token_azure_ad_oversized_response_returns_none( + self, monkeypatch + ): + """Oversized token metadata is rejected before JSON parsing.""" + from unittest.mock import MagicMock, patch + + from specify_cli._download_security import MAX_JSON_METADATA_BYTES + + monkeypatch.setenv("MY_SECRET", "secret-value") + entry = AuthConfigEntry( + hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad", + tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET", + ) + mock_resp = MagicMock() + mock_resp.read.side_effect = io.BytesIO( + b"x" * (MAX_JSON_METADATA_BYTES + 1) + ).read + mock_resp.__enter__ = lambda s: s + mock_resp.__exit__ = MagicMock(return_value=False) + mock_opener = MagicMock() + mock_opener.open.return_value = mock_resp + + with patch("urllib.request.build_opener", return_value=mock_opener), patch( + "specify_cli.authentication.azure_devops._json.loads", + side_effect=AssertionError("oversized body must not be parsed"), + ): + assert AzureDevOpsAuth().resolve_token(entry) is None + + @pytest.mark.parametrize("payload", [[], {"access_token": None}, {"access_token": 123}]) + def test_resolve_token_azure_ad_unexpected_json_shape_returns_none( + self, monkeypatch, payload + ): + from unittest.mock import MagicMock, patch + + monkeypatch.setenv("MY_SECRET", "secret-value") + entry = AuthConfigEntry( + hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad", + tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET", + ) + mock_resp = MagicMock() + mock_resp.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read + mock_resp.__enter__ = lambda s: s + mock_resp.__exit__ = MagicMock(return_value=False) + mock_opener = MagicMock() + mock_opener.open.return_value = mock_resp + + with patch("urllib.request.build_opener", return_value=mock_opener): + assert AzureDevOpsAuth().resolve_token(entry) is None + def test_resolve_token_azure_ad_invalid_utf8_returns_none(self, monkeypatch): """azure-ad returns None when the token response is not valid UTF-8.""" from unittest.mock import MagicMock, patch @@ -960,7 +1038,41 @@ def test_redirect_rejects_https_downgrade(self): handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {}, "http://evil.example.com/archive.zip") - def test_redirect_rejects_remote_to_http_loopback(self): + @pytest.mark.parametrize( + "target", + [ + "http://127.0.0.1/internal", + "https://localhost/internal", + "https://localhost./internal", + "https://service.localhost/internal", + "https://service.localhost./internal", + "https://127.0.0.2/internal", + "https://127.1/internal", + "https://2130706433/internal", + "https://0x7f000001/internal", + "https://017700000001/internal", + "https://0177.0.0.1/internal", + "https://[::1]/internal", + "https://[::1%25lo0]/internal", + "https://[::ffff:127.0.0.1]/internal", + "https://127%2e0%2e0%2e1/internal", + "https://%31%32%37.0.0.1/internal", + "https://127%2E1/internal", + "https://local%68ost/internal", + "https://[::ffff:127%2e0.0.1]/internal", + "https://[::ffff:7f00%3a1]/internal", + "https://[::ffff%3a127.0.0.1]/internal", + "https://ℓocalhost/internal", + "https://ｌｏｃａｌｈｏｓｔ/internal", + "https://127。0。0。1/internal", + "https://0.0.0.0/internal", + "https://0/internal", + "https://00.00.00.00/internal", + "https://[::]/internal", + "https://[::ffff:0.0.0.0]/internal", + ], + ) + def test_redirect_rejects_remote_to_loopback(self, target): """A remote response must not redirect a download into loopback.""" import io import urllib.error @@ -978,10 +1090,27 @@ def test_redirect_rejects_remote_to_http_loopback(self): 302, "Found", {}, - "http://127.0.0.1/internal", + target, ) - def test_redirect_allows_loopback_to_http_loopback(self): + @pytest.mark.parametrize( + ("source", "target"), + [ + ( + "http://localhost:8000/archive.zip", + "http://127.0.0.1:8001/archive.zip", + ), + ( + "http://127.0.0.2:8000/archive.zip", + "http://127.255.255.254:8001/archive.zip", + ), + ( + "https://[0:0:0:0:0:0:0:1]/archive.zip", + "http://[::1]:8001/archive.zip", + ), + ], + ) + def test_redirect_allows_loopback_to_http_loopback(self, source, target): """Local development may continue redirecting between loopback URLs.""" import io from urllib.request import Request @@ -989,18 +1118,66 @@ def test_redirect_allows_loopback_to_http_loopback(self): from specify_cli.authentication.http import _StripAuthOnRedirect handler = _StripAuthOnRedirect(()) - request = Request("http://localhost:8000/archive.zip") + request = Request(source) redirected = handler.redirect_request( request, io.BytesIO(b""), 302, "Found", {}, - "http://127.0.0.1:8001/archive.zip", + target, ) assert redirected is not None + def test_multi_hop_remote_to_loopback_chain_is_rejected_at_first_hop(self): + import io + import urllib.error + from urllib.request import Request + + from specify_cli.authentication.http import _StripAuthOnRedirect + + handler = _StripAuthOnRedirect(()) + request = Request("https://example.com/archive.zip") + + with pytest.raises(urllib.error.URLError, match="unsafe redirect"): + handler.redirect_request( + request, + io.BytesIO(b""), + 302, + "Found", + {}, + "https://localhost:4443/hop", + ) + + @pytest.mark.parametrize( + "target", + [ + "https://example.com:notaport/archive.zip", + "https://example.com:+443/archive.zip", + "https://example.com:65536/archive.zip", + ], + ) + def test_malformed_redirect_port_raises_urlerror(self, target): + import io + import urllib.error + from urllib.request import Request + + from specify_cli.authentication.http import _StripAuthOnRedirect + + handler = _StripAuthOnRedirect(()) + request = Request("https://example.com/archive.zip") + + with pytest.raises(urllib.error.URLError, match="malformed redirect URL"): + handler.redirect_request( + request, + io.BytesIO(b""), + 302, + "Found", + {}, + target, + ) + def test_strict_redirect_error_describes_target_and_allowed_localhost(self): from specify_cli.authentication.http import _StripAuthOnRedirect from urllib.request import Request diff --git a/tests/test_download_security.py b/tests/test_download_security.py index 24eeedd959..f5dffe10de 100644 --- a/tests/test_download_security.py +++ b/tests/test_download_security.py @@ -2,10 +2,13 @@ from __future__ import annotations +import weakref + import pytest from specify_cli._download_security import ( is_https_or_localhost_http, + is_loopback_url, read_response_limited, ) @@ -16,20 +19,93 @@ ("https://example.com/preset.zip", True), ("http://localhost:8000/preset.zip", True), ("http://127.0.0.1/preset.zip", True), + ("http://127.0.0.2/preset.zip", True), + ("http://127.255.255.254/preset.zip", True), ("http://[::1]/preset.zip", True), + ("http://[0:0:0:0:0:0:0:1]/preset.zip", True), + ("http://[::ffff:127.0.0.2]/preset.zip", True), + ("http://[::1%25lo0]/preset.zip", True), # Non-loopback HTTP is rejected. ("http://example.com/preset.zip", False), - # Loopback allowance is an exact-string match: 127.0.0.2 is not covered. - ("http://127.0.0.2/preset.zip", False), + ("http://192.0.2.1/preset.zip", False), + ("http://[fe80::1]/preset.zip", False), + ("http://[fe80::1%25lo0]/preset.zip", False), + ("http://0.0.0.0/preset.zip", False), + ("http://0/preset.zip", False), + ("http://[::]/preset.zip", False), + ("http://[::ffff:0.0.0.0]/preset.zip", False), + # Ambiguous/platform-dependent spellings may never authorize HTTP. + ("http://127.1/preset.zip", False), + ("http://2130706433/preset.zip", False), + ("http://0x7f000001/preset.zip", False), + ("http://017700000001/preset.zip", False), + ("http://0177.0.0.1/preset.zip", False), + ("http://00177.0.0.1/preset.zip", False), + ("http://localhost./preset.zip", False), + ("http://ℓocalhost/preset.zip", False), + ("http://127。0。0。1/preset.zip", False), # A hostname is always required, even for HTTPS. ("https:///preset.zip", False), ("https://", False), + # Invalid ports must be rejected before urllib opens the URL. + ("https://example.com:notaport/preset.zip", False), + ("https://example.com:+443/preset.zip", False), + ("https://example.com:65536/preset.zip", False), + # urllib decodes escapes in the authority before connecting; reject + # encoded reg-names so validation and connection cannot disagree. + ("https://127%2e0%2e0%2e1/preset.zip", False), + ("https://%31%32%37.0.0.1/preset.zip", False), + ("https://local%68ost/preset.zip", False), + ("https://example.com%3a443/preset.zip", False), + ("https://[::1%lo0]/preset.zip", False), + ("https://[::ffff:127%2e0.0.1]/preset.zip", False), + ("https://[::ffff:7f00%3a1]/preset.zip", False), + ("https://[::ffff%3a127.0.0.1]/preset.zip", False), ], ) def test_is_https_or_localhost_http(url, allowed): assert is_https_or_localhost_http(url) is allowed +@pytest.mark.parametrize( + "url", + [ + "https://localhost/internal", + "https://127.0.0.2/internal", + "https://[::1]/internal", + "https://[::1%25lo0]/internal", + "https://[::ffff:127.0.0.2]/internal", + ], +) +def test_is_loopback_url_recognizes_effective_loopback_literals(url): + assert is_loopback_url(url) is True + + +@pytest.mark.parametrize( + "url", + [ + "https://localhost./internal", + "https://service.localhost/internal", + "https://service.localhost./internal", + "https://127.1/internal", + "https://2130706433/internal", + "https://0x7f000001/internal", + "https://017700000001/internal", + "https://0177.0.0.1/internal", + "https://ℓocalhost/internal", + "https://127。0。0。1/internal", + "https://127%2e0%2e0%2e1/internal", + "https://0.0.0.0/internal", + "https://0/internal", + "https://00.00.00.00/internal", + "https://[::]/internal", + "https://[::ffff:0.0.0.0]/internal", + ], +) +def test_is_loopback_url_does_not_authorize_ambiguous_spellings(url): + assert is_loopback_url(url) is False + + class _Response: """Faithful stream stand-in: read() advances a cursor and returns b"" at EOF.""" @@ -50,6 +126,41 @@ def read(self, size: int = -1) -> bytes: return out +class _RecordingResponse(_Response): + def __init__(self, data: bytes, *, chunk: int | None = None): + super().__init__(data, chunk=chunk) + self.requested_sizes: list[int] = [] + + def read(self, size: int = -1) -> bytes: + self.requested_sizes.append(size) + return super().read(size) + + +class _TrackedChunk(bytearray): + pass + + +class _OneByteResponse: + """Return distinct weak-referenceable chunks to detect retained fragments.""" + + def __init__(self, count: int): + self.remaining = count + self.refs: list[weakref.ReferenceType[_TrackedChunk]] = [] + self.peak_live = 0 + + def read(self, _size: int = -1) -> bytes | _TrackedChunk: + if self.remaining == 0: + return b"" + self.remaining -= 1 + chunk = _TrackedChunk(b"x") + self.refs.append(weakref.ref(chunk)) + self.peak_live = max( + self.peak_live, + sum(ref() is not None for ref in self.refs), + ) + return chunk + + def test_read_response_limited_rejects_oversized_download(): with pytest.raises(ValueError, match="exceeds maximum size"): read_response_limited(_Response(b"abcde"), max_bytes=4) @@ -66,3 +177,51 @@ def test_read_response_limited_enforces_bound_under_short_reads(): response = _Response(b"x" * 100, chunk=8) with pytest.raises(ValueError, match="exceeds maximum size"): read_response_limited(response, max_bytes=16) + + +def test_read_response_limited_does_not_retain_short_read_fragments(): + response = _OneByteResponse(64) + + assert read_response_limited(response, max_bytes=64) == b"x" * 64 + assert response.peak_live <= 2 + + +def test_read_response_limited_caps_underlying_reads_at_64_kib(): + response = _RecordingResponse(b"x" * (64 * 1024 + 1)) + + with pytest.raises(ValueError, match="exceeds maximum size"): + read_response_limited(response, max_bytes=64 * 1024) + + assert max(response.requested_sizes) <= 64 * 1024 + + +@pytest.mark.parametrize("value", [None, "1", 1.5, True]) +def test_read_response_limited_rejects_non_integer_limits(value): + with pytest.raises(TypeError, match="integer"): + read_response_limited(_Response(b""), max_bytes=value) + + +def test_read_response_limited_rejects_negative_limit_without_reading(): + response = _RecordingResponse(b"") + + with pytest.raises(ValueError, match="non-negative"): + read_response_limited(response, max_bytes=-1) + + assert response.requested_sizes == [] + + +def test_read_response_limited_allows_empty_response_at_zero_limit(): + assert read_response_limited(_Response(b""), max_bytes=0) == b"" + + +class _CustomLimitError(Exception): + pass + + +def test_read_response_limited_rejects_first_byte_at_zero_limit(): + with pytest.raises(_CustomLimitError, match="exceeds maximum size"): + read_response_limited( + _Response(b"x"), + max_bytes=0, + error_type=_CustomLimitError, + ) diff --git a/tests/test_self_upgrade_guidance.py b/tests/test_self_upgrade_guidance.py index 55d6c2bf7b..5fcb6a5274 100644 --- a/tests/test_self_upgrade_guidance.py +++ b/tests/test_self_upgrade_guidance.py @@ -6,6 +6,7 @@ from tests.self_upgrade_helpers import ( mock_urlopen_response, + route_opener_open_through_urlopen, # noqa: F401 (autouse fixture) runner, strip_ansi, ) diff --git a/tests/test_upgrade.py b/tests/test_upgrade.py index b6f3134286..edce927e84 100644 --- a/tests/test_upgrade.py +++ b/tests/test_upgrade.py @@ -2,11 +2,12 @@ Network isolation contract (SC-004 / FR-014): every test that exercises `specify self check` or `_fetch_latest_release_tag()` MUST mock the outbound -urllib path it expects (`urlopen` for unauthenticated requests, `build_opener` -for authenticated requests) so no real outbound call ever reaches api.github.com. -Tests for non-network `self upgrade` behavior should keep that contract explicit -with local mocks. Run this module under `pytest-socket` (if installed) with -`--disable-socket` as an extra safety net. +urllib path so no real call reaches api.github.com. Production always uses an +isolated `build_opener`; this module's autouse fixture routes its `open()` back +through the locally mocked `urlopen`. Tests for non-network `self upgrade` +behavior should keep that contract explicit with local mocks. Run this module +under `pytest-socket` (if installed) with `--disable-socket` as an extra safety +net. """ import urllib.error diff --git a/tests/test_workflows.py b/tests/test_workflows.py index b83514ef95..2c26301cc3 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -11744,6 +11744,10 @@ def test_download_redirect_validator_rejects_http_before_follow(self): _reject_insecure_download_redirect( "https://example.com/wf.yml", "http://localhost:8000/wf.yml" ) + with pytest.raises(urllib.error.URLError): + _reject_insecure_download_redirect( + "https://example.com/wf.yml", "https://127.0.0.2/wf.yml" + ) # Allowed: HTTPS anywhere, or loopback HTTP that stays on loopback HTTP. _reject_insecure_download_redirect( "https://example.com/wf.yml", "https://cdn.example.com/wf.yml" @@ -11754,6 +11758,9 @@ def test_download_redirect_validator_rejects_http_before_follow(self): _reject_insecure_download_redirect( "http://127.0.0.1/source.yml", "http://127.0.0.1/wf.yml" ) + _reject_insecure_download_redirect( + "http://127.0.0.2/source.yml", "http://127.255.255.254/wf.yml" + ) def test_add_from_url_passes_redirect_validator(self, project_dir, monkeypatch): from unittest.mock import patch