github/github-spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-08-03 06:26:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden bounded reads and redirect validation (#3671)

Browse Source 
Assisted-by: OpenAI Codex (model: GPT-5, autonomous)
This commit is contained in:
Pascal THUET
2026-07-22 22:18:10 +02:00
committed by GitHub
parent c0f4cee25a
commit 3a7a8758f7
10 changed files with 557 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
src/specify_cli/_download_security.py
View File

@@ -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 @@ READ_CHUNK_SIZE = 1024 * 1024
# 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()

16
src/specify_cli/authentication/azure_devops.py
View File

@@ -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 @@ class AzureDevOpsAuth(AuthProvider):
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 @@ class AzureDevOpsAuth(AuthProvider):
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,

18
src/specify_cli/authentication/http.py
View File

@@ -17,7 +17,7 @@ from fnmatch import fnmatch
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 @@ RedirectValidator = Callable[[str, str], None]
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 @@ class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
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())

29
src/specify_cli/presets/_commands.py
View File

@@ -16,6 +16,10 @@ import yaml
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 preset_add(
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, "

89
src/specify_cli/workflows/_commands.py
View File

@@ -20,6 +20,10 @@ import yaml
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 @@ _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"}
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 workflow_add(
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 workflow_add(
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 _install_workflow_from_catalog(
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 _install_workflow_from_catalog(
) 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)

195
tests/test_authentication.py
View File

@@ -516,6 +516,23 @@ class TestAzureDevOpsAuth:
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 @@ class TestAzureDevOpsAuth:
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 @@ class TestAzureDevOpsAuth:
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 @@ class TestRedirectStripping:
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 @@ class TestRedirectStripping:
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 @@ class TestRedirectStripping:
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

163
tests/test_download_security.py
View File

@@ -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 @@ from specify_cli._download_security import (
("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 @@ class _Response:
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,
)

1
tests/test_self_upgrade_guidance.py
View File

@@ -6,6 +6,7 @@ from specify_cli import app
from tests.self_upgrade_helpers import (
mock_urlopen_response,
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
runner,
strip_ansi,
)

11
tests/test_upgrade.py
View File

@@ -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

7
tests/test_workflows.py
View File

@@ -11744,6 +11744,10 @@ steps:
_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 @@ steps:
_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