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

harden: bound HTTP reads and enforce strict redirects (#3140)

Browse Source 
* harden: bound HTTP reads and enforce strict redirects

Add a shared _download_security module (read_response_limited,
is_https_or_localhost_http, size constants) and route the GitHub release
and Azure DevOps token network reads through bounded reads so an oversized
response can't exhaust memory.

Add a strict_redirects mode to authentication.open_url: the redirect handler
now rejects any redirect whose target isn't HTTPS (or HTTP to localhost),
composing with the existing per-hop redirect_validator and auth-stripping.
The Azure DevOps token POST is routed through that handler so a 307/308
cannot forward the client_secret body to a non-HTTPS host.

Assisted-by: Codex (model: GPT-5, autonomous)

* test: align HTTP fakes with bounded reads

Assisted-by: Codex (model: GPT-5, autonomous)

* fix: tolerate invalid token response encoding

Assisted-by: Codex (model: GPT-5, autonomous)

* test: align GHES fakes with bounded reads

Assisted-by: Codex (model: GPT-5, autonomous)

* test: reuse shared upgrade HTTP response helper

Assisted-by: Codex (model: GPT-5, autonomous)

* fix: include rejected redirect target in error

Assisted-by: Codex (model: GPT-5, autonomous)

* fix: enforce strict redirects by default

Assisted-by: Codex (model: GPT-5, autonomous)

* fix: close redirect credential and SSRF gaps

Assisted-by: Codex (model: GPT-5, autonomous)
This commit is contained in:
Pascal THUET
2026-07-22 18:08:32 +02:00
committed by GitHub
parent 0f6ea64a03
commit 5601830ba3
18 changed files with 592 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
src/specify_cli/_download_security.py Normal file
View File

@@ -0,0 +1,89 @@
"""Helpers for bounded HTTP downloads."""
from __future__ import annotations
from typing import NoReturn, TypeVar
from urllib.parse import urlparse
ErrorT = TypeVar("ErrorT", bound=Exception)
MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024
READ_CHUNK_SIZE = 1024 * 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
# downloads; JSON metadata responses are far smaller, so capping them close to
# their real size shrinks the memory-DoS surface and keeps the "too large"
# error reachable (rather than only triggering on tens of MiB). Pass it
# explicitly at each JSON call site so the intended bound is pinned there.
# 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 is_loopback_url(url: str) -> bool:
"""Return whether *url* targets an explicitly allowed loopback host."""
return urlparse(url).hostname in _LOOPBACK_HOSTS
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.
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.
"""
parsed = urlparse(url)
if not parsed.hostname:
return False
is_localhost = parsed.hostname in _LOOPBACK_HOSTS
return parsed.scheme == "https" or (parsed.scheme == "http" and is_localhost)
def _raise(error_type: type[ErrorT], message: str) -> NoReturn:
raise error_type(message)
def read_response_limited(
response,
*,
max_bytes: int = MAX_DOWNLOAD_BYTES,
error_type: type[ErrorT] = ValueError,
label: str = "download",
) -> bytes:
"""Read at most *max_bytes* from a response object.
``response.read(n)`` is only guaranteed to return *up to* ``n`` bytes and may
return fewer even when more data is pending (e.g. chunked transfer encoding),
so a single ``read(max_bytes + 1)`` cannot enforce the bound on its own. Read
in a loop until EOF or until one byte past the limit has been accumulated.
*max_bytes* is keyword-only. It defaults to the module-wide
``MAX_DOWNLOAD_BYTES`` (50 MiB) ceiling for archive/payload downloads;
callers with a tighter budget (e.g. small JSON responses) should pass an
explicit value so the intended bound is pinned at the call site rather than
tracking changes to the shared default.
"""
chunks: list[bytes] = []
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)

13
src/specify_cli/_github_http.py
View File

@@ -100,6 +100,8 @@ def resolve_github_release_asset_api_url(
import json
import urllib.error
from specify_cli._download_security import read_response_limited
parsed = urlparse(download_url)
hostname = (parsed.hostname or "").lower()
parts = [unquote(part) for part in parsed.path.strip("/").split("/")]
@@ -158,10 +160,13 @@ def resolve_github_release_asset_api_url(
if redirect_validator is not None:
open_kwargs["redirect_validator"] = redirect_validator
with open_url_fn(release_url, **open_kwargs) as response:
raw_release_data = response.read(max_metadata_bytes + 1)
if len(raw_release_data) > max_metadata_bytes:
raise ValueError("GitHub release metadata exceeds size limit")
release_data = json.loads(raw_release_data)
release_data = json.loads(
read_response_limited(
response,
max_bytes=max_metadata_bytes,
label=f"GitHub release metadata {release_url}",
)
)
except (
urllib.error.URLError,
json.JSONDecodeError,

13
src/specify_cli/_version.py
View File

@@ -4,8 +4,8 @@ Pure helpers for comparing PEP 440 versions and fetching the latest GitHub
release tag. The ``self_app`` Typer sub-command group is co-located here so
all version-related logic lives in one place.
Dependencies: stdlib + packaging + ._console only (no other internal imports
at module level, keeping this layer thin and circular-import-safe).
Dependencies: stdlib + packaging + ._console + ._download_security only
(keeping this layer thin and circular-import-safe).
"""
from __future__ import annotations
@@ -28,6 +28,7 @@ from pathlib import Path
import typer
from packaging.version import InvalidVersion, Version
from ._download_security import MAX_JSON_METADATA_BYTES, read_response_limited
from ._console import console
GITHUB_API_LATEST = "https://api.github.com/repos/github/spec-kit/releases/latest"
@@ -119,7 +120,13 @@ def _fetch_latest_release_tag() -> tuple[str | None, str | None]:
timeout=5,
extra_headers={"Accept": "application/vnd.github+json"},
) as resp:
payload = json.loads(resp.read().decode("utf-8"))
payload = json.loads(
read_response_limited(
resp,
max_bytes=MAX_JSON_METADATA_BYTES,
label="GitHub latest release",
).decode("utf-8")
)
tag = payload.get("tag_name")
if not isinstance(tag, str) or not tag:
raise ValueError("GitHub API response missing valid tag_name")

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

@@ -8,6 +8,7 @@ import os
import subprocess
from typing import TYPE_CHECKING
from .._download_security import MAX_JSON_METADATA_BYTES, read_response_limited
from .base import AuthProvider
if TYPE_CHECKING:
@@ -17,6 +18,10 @@ if TYPE_CHECKING:
_ADO_RESOURCE_ID = "499b84ac-1321-427f-aa17-267ca6975798"
class _TokenResponseTooLarge(Exception):
"""Raised when an Azure AD token response exceeds the bounded read limit."""
class AzureDevOpsAuth(AuthProvider):
"""Azure DevOps authentication provider.
@@ -119,9 +124,38 @@ class AzureDevOpsAuth(AuthProvider):
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(req, timeout=30) as resp: # noqa: S310
payload = _json.loads(resp.read().decode("utf-8"))
from specify_cli.authentication.http import _StripAuthOnRedirect
def reject_token_redirect(_old_url: str, new_url: str) -> None:
# A 307/308 redirect preserves this POST body, including the
# client_secret. Refuse every redirect so credentials cannot
# leave the fixed Microsoft token endpoint.
raise urllib.error.URLError(
f"Azure AD token request must not be redirected to {new_url}"
)
opener = urllib.request.build_opener(
_StripAuthOnRedirect((), reject_token_redirect)
)
with opener.open(req, timeout=30) as resp: # noqa: S310
payload = _json.loads(
read_response_limited(
resp,
max_bytes=MAX_JSON_METADATA_BYTES,
error_type=_TokenResponseTooLarge,
label="Azure DevOps token response",
).decode("utf-8")
)
token = payload.get("access_token", "").strip()
return token or None
except (urllib.error.URLError, OSError, _json.JSONDecodeError, KeyError):
except (
urllib.error.URLError,
OSError,
_json.JSONDecodeError,
UnicodeDecodeError,
_TokenResponseTooLarge,
):
# Network failure, malformed JSON, or an oversized response — fall
# through to the next strategy. Unrelated programming errors (other
# ValueErrors, KeyErrors) intentionally propagate so they surface.
return None

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

@@ -17,6 +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 . import get_provider
from .config import AuthConfigEntry, _default_config_path, find_entries_for_url, load_auth_config
@@ -60,8 +61,27 @@ def _hostname_in_hosts(hostname: str, hosts: tuple[str, ...]) -> bool:
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:
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)"
)
class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
"""Drop ``Authorization`` when a redirect leaves trusted hosts or downgrades."""
"""Redirect handler that guards every redirect it is installed for.
1. Run any caller-provided redirect validator.
2. Reject redirects that are not HTTPS with a hostname. HTTP loopback is
allowed only when the previous hop is also loopback.
3. Drop ``Authorization`` when a redirect leaves trusted hosts or downgrades.
"""
def __init__(
self,
@@ -82,6 +102,7 @@ class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
if self._redirect_validator is not None:
self._redirect_validator(req.full_url, newurl)
_validate_strict_redirect(req.full_url, newurl)
original_auth = (
req.get_header("Authorization")
@@ -155,6 +176,10 @@ def open_url(
*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.
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.
"""
entries = find_entries_for_url(url, _load_config())
@@ -188,7 +213,7 @@ def open_url(
# 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
# No auth is attached on this path, so the handler's host list is empty:
# here it runs redirect validation only, not auth stripping.
opener = urllib.request.build_opener(_StripAuthOnRedirect((), redirect_validator))
return opener.open(req, timeout=timeout)

35
tests/http_helpers.py
View File

@@ -1,15 +1,46 @@
"""HTTP test helpers shared by version-related CLI tests."""
"""HTTP test helpers shared by CLI tests."""
import io
import json
import urllib.request
from unittest.mock import MagicMock
import pytest
def mock_urlopen_response(payload: dict) -> MagicMock:
"""Build a urlopen context-manager mock whose read returns JSON."""
body = json.dumps(payload).encode("utf-8")
resp = MagicMock()
resp.read.return_value = body
resp.read.side_effect = io.BytesIO(body).read
cm = MagicMock()
cm.__enter__.return_value = resp
cm.__exit__.return_value = False
return cm
@pytest.fixture(autouse=True)
def route_opener_open_through_urlopen(monkeypatch):
"""Route build_opener().open through urllib.request.urlopen.
``open_url(...)`` fetches via ``build_opener(...).open()``, which bypasses
``urllib.request.urlopen`` — and with it the urlopen patches these test
modules are built on.
Delegating ``open()`` to urlopen at call time keeps those patches
effective; the redirect handler's own behavior is covered by
``TestRedirectStripping`` in test_authentication.py.
Import this fixture into a test module to activate it there.
"""
class _UrlopenDelegatingOpener:
def open(self, req, data=None, timeout=None):
if data is None:
return urllib.request.urlopen(req, timeout=timeout)
return urllib.request.urlopen(req, data=data, timeout=timeout)
monkeypatch.setattr(
urllib.request,
"build_opener",
lambda *handlers: _UrlopenDelegatingOpener(),
)

2
tests/integrations/test_integration_catalog.py
View File

@@ -6,6 +6,8 @@ import os
import pytest
import yaml
from tests.http_helpers import route_opener_open_through_urlopen # noqa: F401
from specify_cli.integrations.catalog import (
IntegrationCatalog,
IntegrationCatalogEntry,

3
tests/self_upgrade_helpers.py
View File

@@ -18,7 +18,7 @@ from specify_cli._version import (
_verify_upgrade,
)
from tests.conftest import strip_ansi
from tests.http_helpers import mock_urlopen_response
from tests.http_helpers import mock_urlopen_response, route_opener_open_through_urlopen
__all__ = (
"SENTINEL_GH_TOKEN",
@@ -31,6 +31,7 @@ __all__ = (
"_verify_upgrade",
"mock_urlopen_response",
"requires_posix",
"route_opener_open_through_urlopen",
"runner",
"strip_ansi",
)

193
tests/test_authentication.py
View File

@@ -14,6 +14,7 @@ Covers:
from __future__ import annotations
import base64
import io
import json
import os
@@ -524,10 +525,15 @@ class TestAzureDevOpsAuth:
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
)
mock_resp = MagicMock()
mock_resp.read.return_value = b'{"access_token": "ad-acquired-token"}'
mock_resp.read.side_effect = io.BytesIO(b'{"access_token": "ad-acquired-token"}').read
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
with patch("urllib.request.urlopen", return_value=mock_resp):
# The token request goes through a strict-redirect opener (so a 307/308
# cannot forward the client_secret body to a non-HTTPS host), not bare
# urlopen; patch the opener it builds.
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) == "ad-acquired-token"
def test_resolve_token_azure_ad_missing_secret_returns_none(self, monkeypatch):
@@ -542,14 +548,62 @@ class TestAzureDevOpsAuth:
def test_resolve_token_azure_ad_network_error_returns_none(self, monkeypatch):
"""azure-ad returns None on network errors."""
import urllib.error
from unittest.mock import patch
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",
)
with patch("urllib.request.urlopen",
side_effect=urllib.error.URLError("connection refused")):
mock_opener = MagicMock()
mock_opener.open.side_effect = urllib.error.URLError("connection refused")
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):
"""The client-secret POST must never be redirected to another host."""
import urllib.error
from unittest.mock import MagicMock, patch
from urllib.request import Request
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_opener = MagicMock()
mock_opener.open.side_effect = urllib.error.URLError("stop after setup")
with patch("urllib.request.build_opener", return_value=mock_opener) as build_opener:
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")
with pytest.raises(urllib.error.URLError, match="must not be redirected"):
redirect_handler.redirect_request(
request,
io.BytesIO(b""),
307,
"Temporary Redirect",
{},
"https://evil.example/token",
)
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
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"\xff").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
@@ -615,13 +669,15 @@ class TestAuthenticatedHttp:
monkeypatch.setenv("GH_TOKEN", "my-token")
self._set_config(monkeypatch, [_github_entry()])
captured = {}
def fake_urlopen(req, timeout=None):
def fake_open(req, timeout=None):
captured["req"] = req
resp = MagicMock()
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
return resp
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_urlopen):
mock_opener = MagicMock()
mock_opener.open.side_effect = fake_open
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
open_url("https://example.com/file.json")
assert captured["req"].get_header("Authorization") is None
@@ -630,13 +686,15 @@ class TestAuthenticatedHttp:
from specify_cli.authentication.http import open_url
self._set_config(monkeypatch, [])
captured = {}
def fake_urlopen(req, timeout=None):
def fake_open(req, timeout=None):
captured["req"] = req
resp = MagicMock()
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
return resp
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_urlopen):
mock_opener = MagicMock()
mock_opener.open.side_effect = fake_open
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
open_url("https://github.com/org/repo")
assert captured["req"].get_header("Authorization") is None
@@ -658,8 +716,7 @@ class TestAuthenticatedHttp:
return resp
mock_opener = MagicMock()
mock_opener.open.side_effect = fake_side_effect
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener), \
patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_side_effect):
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
open_url("https://github.com/org/repo")
assert call_count == 2
@@ -700,21 +757,23 @@ class TestAuthenticatedHttpNegative:
def test_urlerror_propagates(self, monkeypatch):
import urllib.error
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from specify_cli.authentication.http import open_url
self._set_config(monkeypatch, [])
with patch("specify_cli.authentication.http.urllib.request.urlopen",
side_effect=urllib.error.URLError("refused")):
mock_opener = MagicMock()
mock_opener.open.side_effect = urllib.error.URLError("refused")
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
with pytest.raises(urllib.error.URLError):
open_url("https://example.com/file")
def test_timeout_propagates(self, monkeypatch):
import socket
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from specify_cli.authentication.http import open_url
self._set_config(monkeypatch, [])
with patch("specify_cli.authentication.http.urllib.request.urlopen",
side_effect=socket.timeout("timed out")):
mock_opener = MagicMock()
mock_opener.open.side_effect = socket.timeout("timed out")
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
with pytest.raises(socket.timeout):
open_url("https://example.com/file")
@@ -820,17 +879,18 @@ class TestRedirectStripping:
assert new_req.headers.get("Authorization") is None
assert new_req.unredirected_hdrs.get("Authorization") is None
def test_https_to_http_same_host_redirect_strips_auth(self):
def test_https_to_http_same_host_redirect_rejected(self):
from specify_cli.authentication.http import _StripAuthOnRedirect
from urllib.request import Request
import io
import urllib.error
handler = _StripAuthOnRedirect(("github.com",))
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
new_req = handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
"http://github.com/org/repo")
assert new_req is not None
assert new_req.headers.get("Authorization") is None
assert new_req.unredirected_hdrs.get("Authorization") is None
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
"http://github.com/org/repo")
def test_redirect_validator_can_reject_before_following_redirect(self):
import urllib.error
@@ -888,6 +948,78 @@ class TestRedirectStripping:
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
"https://[::1/asset")
def test_redirect_rejects_https_downgrade(self):
"""HTTPS downloads must not follow redirects to non-local HTTP URLs."""
from specify_cli.authentication.http import _StripAuthOnRedirect
from urllib.request import Request
import io
import urllib.error
handler = _StripAuthOnRedirect(("example.com",))
req = Request("https://example.com/archive.zip")
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
"http://evil.example.com/archive.zip")
def test_redirect_rejects_remote_to_http_loopback(self):
"""A remote response must not redirect a download into loopback."""
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",
{},
"http://127.0.0.1/internal",
)
def test_redirect_allows_loopback_to_http_loopback(self):
"""Local development may continue redirecting between loopback URLs."""
import io
from urllib.request import Request
from specify_cli.authentication.http import _StripAuthOnRedirect
handler = _StripAuthOnRedirect(())
request = Request("http://localhost:8000/archive.zip")
redirected = handler.redirect_request(
request,
io.BytesIO(b""),
302,
"Found",
{},
"http://127.0.0.1:8001/archive.zip",
)
assert redirected is not None
def test_strict_redirect_error_describes_target_and_allowed_localhost(self):
from specify_cli.authentication.http import _StripAuthOnRedirect
from urllib.request import Request
import io
import urllib.error
handler = _StripAuthOnRedirect(("example.com",))
req = Request("https://example.com/archive.zip")
with pytest.raises(urllib.error.URLError) as exc_info:
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
"http://evil.example.com/archive.zip")
error_message = str(exc_info.value)
assert "http://evil.example.com/archive.zip" in error_message
assert "localhost" in error_message
assert "127.0.0.1" in error_message
assert "::1" in error_message
# ---------------------------------------------------------------------------
# _fetch_latest_release_tag delegation
@@ -907,7 +1039,7 @@ class TestFetchLatestReleaseTagDelegation:
captured["request"] = req
body = _json.dumps({"tag_name": "v9.9.9"}).encode()
resp = MagicMock()
resp.read.return_value = body
resp.read.side_effect = io.BytesIO(body).read
cm = MagicMock()
cm.__enter__.return_value = resp
cm.__exit__.return_value = False
@@ -927,20 +1059,25 @@ class TestFetchLatestReleaseTagDelegation:
assert captured["request"].get_header("Authorization") == "Bearer forwarded-sentinel"
def test_no_config_means_no_auth(self, monkeypatch):
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from specify_cli._version import _fetch_latest_release_tag
self._set_config(monkeypatch, [])
captured, side_effect = self._capture_request()
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
# The unauthenticated path uses the strict redirect opener too.
mock_opener = MagicMock()
mock_opener.open.side_effect = side_effect
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
_fetch_latest_release_tag()
assert captured["request"].get_header("Authorization") is None
def test_accept_header_present(self, monkeypatch):
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from specify_cli._version import _fetch_latest_release_tag
self._set_config(monkeypatch, [])
captured, side_effect = self._capture_request()
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
mock_opener = MagicMock()
mock_opener.open.side_effect = side_effect
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
_fetch_latest_release_tag()
assert captured["request"].get_header("Accept") == "application/vnd.github+json"

68
tests/test_download_security.py Normal file
View File

@@ -0,0 +1,68 @@
"""Tests for bounded HTTP download helpers."""
from __future__ import annotations
import pytest
from specify_cli._download_security import (
is_https_or_localhost_http,
read_response_limited,
)
@pytest.mark.parametrize(
"url, allowed",
[
("https://example.com/preset.zip", True),
("http://localhost:8000/preset.zip", True),
("http://127.0.0.1/preset.zip", True),
("http://[::1]/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),
# A hostname is always required, even for HTTPS.
("https:///preset.zip", False),
("https://", False),
],
)
def test_is_https_or_localhost_http(url, allowed):
assert is_https_or_localhost_http(url) is allowed
class _Response:
"""Faithful stream stand-in: read() advances a cursor and returns b"" at EOF."""
def __init__(self, data: bytes, *, chunk: int | None = None):
self.data = data
self.pos = 0
# When set, never return more than *chunk* bytes per call even if more is
# requested - simulates short reads (e.g. chunked transfer encoding).
self.chunk = chunk
def read(self, size: int = -1) -> bytes:
if size < 0:
size = len(self.data) - self.pos
if self.chunk is not None:
size = min(size, self.chunk)
out = self.data[self.pos : self.pos + size]
self.pos += len(out)
return out
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)
def test_read_response_limited_returns_full_body_within_limit():
assert read_response_limited(_Response(b"abcde"), max_bytes=10) == b"abcde"
def test_read_response_limited_enforces_bound_under_short_reads():
# A server that streams more than max_bytes total while every read() returns
# fewer bytes than requested (chunked encoding) must still be rejected - a
# single read(max_bytes + 1) could be fooled, the accumulating loop cannot.
response = _Response(b"x" * 100, chunk=8)
with pytest.raises(ValueError, match="exceeds maximum size"):
read_response_limited(response, max_bytes=16)

12
tests/test_extensions.py
View File

@@ -9,6 +9,7 @@ Tests cover:
- Catalog stack (multi-catalog support)
"""
import io
import pytest
import json
import os
@@ -22,6 +23,7 @@ from datetime import datetime, timezone
from unittest.mock import MagicMock
from tests.conftest import strip_ansi
from tests.http_helpers import route_opener_open_through_urlopen # noqa: F401
from specify_cli import extensions as _ext_module
from specify_cli.extensions import (
CatalogEntry,
@@ -4978,7 +4980,7 @@ class TestExtensionCatalog:
zip_bytes = zip_buf.getvalue()
release_response = MagicMock()
release_response.read.return_value = json.dumps(
release_response.read.side_effect = io.BytesIO(json.dumps(
{
"assets": [
{
@@ -4987,12 +4989,12 @@ class TestExtensionCatalog:
}
]
}
).encode()
).encode()).read
release_response.__enter__ = lambda s: s
release_response.__exit__ = MagicMock(return_value=False)
asset_response = MagicMock()
asset_response.read.return_value = zip_bytes
asset_response.read.side_effect = io.BytesIO(zip_bytes).read
asset_response.__enter__ = lambda s: s
asset_response.__exit__ = MagicMock(return_value=False)
@@ -8761,10 +8763,10 @@ def test_extension_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, mo
def fake_open(url, timeout=None, extra_headers=None):
captured.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({
resp.read.side_effect = io.BytesIO(json.dumps({
"assets": [{"name": "ext.zip",
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/7"}]
}).encode()
}).encode()).read
yield resp
monkeypatch.setattr(catalog, "_open_url", fake_open)

60
tests/test_github_http.py
View File

@@ -1,16 +1,20 @@
"""Tests for GitHub-authenticated HTTP request helpers."""
import io
import json
import os
from contextlib import contextmanager
from unittest.mock import MagicMock, patch
from urllib.request import Request
import pytest
from specify_cli._github_http import (
GITHUB_HOSTS,
build_github_request,
resolve_github_release_asset_api_url,
)
from specify_cli.authentication.http import _StripAuthOnRedirect
class TestBuildGitHubRequest:
@@ -90,7 +94,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
@contextmanager
def fake_open(url, timeout=None, extra_headers=None):
resp = MagicMock()
resp.read.return_value = json.dumps(release_json).encode()
resp.read.side_effect = io.BytesIO(json.dumps(release_json).encode()).read
yield resp
return fake_open
@@ -198,7 +202,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
def capturing_open(url, timeout=None, extra_headers=None):
captured_urls.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({"assets": []}).encode()
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
yield resp
resolve_github_release_asset_api_url(
@@ -217,7 +221,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
def capturing_open(url, timeout=None, extra_headers=None):
captured_urls.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({"assets": []}).encode()
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
yield resp
resolve_github_release_asset_api_url(
@@ -260,7 +264,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
def recording_open(url, timeout=None, extra_headers=None):
called.append(url)
resp = MagicMock()
resp.read.return_value = b"{}"
resp.read.side_effect = io.BytesIO(b"{}").read
yield resp
result = resolve_github_release_asset_api_url(
@@ -299,7 +303,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
def recording_open(url, timeout=None, extra_headers=None):
called.append(url)
resp = MagicMock()
resp.read.return_value = b"{}"
resp.read.side_effect = io.BytesIO(b"{}").read
yield resp
url = "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
@@ -317,7 +321,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
def capturing_open(url, timeout=None, extra_headers=None):
captured.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({"assets": []}).encode()
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
yield resp
resolve_github_release_asset_api_url(
@@ -344,10 +348,10 @@ class TestResolveGitHubReleaseAssetApiUrl:
def capturing_open(url, timeout=None, extra_headers=None):
captured.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({
resp.read.side_effect = io.BytesIO(json.dumps({
"assets": [{"name": "pack.zip",
"url": "https://api.github.com/repos/org/repo/releases/assets/99"}]
}).encode()
}).encode()).read
yield resp
result = resolve_github_release_asset_api_url(
@@ -357,3 +361,43 @@ class TestResolveGitHubReleaseAssetApiUrl:
)
assert result == "https://api.github.com/repos/org/repo/releases/assets/99"
assert captured == ["https://api.github.com/repos/org/repo/releases/tags/v1.0"]
class TestGitHubRedirectAuth:
"""Tests for GitHub-owned redirect auth handling."""
def test_multi_hop_github_redirect_preserves_unredirected_auth(self):
"""Auth survives a multi-hop redirect chain within GitHub hosts."""
handler = _StripAuthOnRedirect(tuple(GITHUB_HOSTS))
req1 = Request(
"https://github.com/org/repo",
headers={"Authorization": "Bearer tok"},
)
req2 = handler.redirect_request(
req1,
io.BytesIO(b""),
302,
"Found",
{},
"https://codeload.github.com/org/repo/zip",
)
assert req2 is not None
auth2 = req2.get_header("Authorization") or req2.unredirected_hdrs.get(
"Authorization"
)
assert auth2 == "Bearer tok"
req3 = handler.redirect_request(
req2,
io.BytesIO(b""),
302,
"Found",
{},
"https://raw.githubusercontent.com/org/repo/main/file",
)
assert req3 is not None
auth3 = req3.get_header("Authorization") or req3.unredirected_hdrs.get(
"Authorization"
)
assert auth3 == "Bearer tok"

10
tests/test_presets.py
View File

@@ -2303,7 +2303,7 @@ class TestPresetCatalog:
zip_bytes = zip_buf.getvalue()
release_response = MagicMock()
release_response.read.return_value = json.dumps(
release_response.read.side_effect = io.BytesIO(json.dumps(
{
"assets": [
{
@@ -2312,12 +2312,12 @@ class TestPresetCatalog:
}
]
}
).encode()
).encode()).read
release_response.__enter__ = lambda s: s
release_response.__exit__ = MagicMock(return_value=False)
asset_response = MagicMock()
asset_response.read.return_value = zip_bytes
asset_response.read.side_effect = io.BytesIO(zip_bytes).read
asset_response.__enter__ = lambda s: s
asset_response.__exit__ = MagicMock(return_value=False)
@@ -7458,10 +7458,10 @@ def test_preset_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, monke
def fake_open(url, timeout=None, extra_headers=None):
captured.append(url)
resp = MagicMock()
resp.read.return_value = json.dumps({
resp.read.side_effect = io.BytesIO(json.dumps({
"assets": [{"name": "pack.zip",
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/9"}]
}).encode()
}).encode()).read
yield resp
monkeypatch.setattr(catalog, "_open_url", fake_open)

1
tests/test_self_upgrade_detection.py
View File

@@ -13,6 +13,7 @@ import specify_cli
from specify_cli import app
from tests.self_upgrade_helpers import (
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
_InstallMethod,
_assemble_installer_argv,
_completed_process,

1
tests/test_self_upgrade_execution.py
View File

@@ -7,6 +7,7 @@ from unittest.mock import patch
from specify_cli import app
from tests.self_upgrade_helpers import (
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
_completed_process,
mock_urlopen_response,
requires_posix,

1
tests/test_self_upgrade_verification.py
View File

@@ -8,6 +8,7 @@ import specify_cli
from specify_cli import app
from tests.self_upgrade_helpers import (
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
SENTINEL_GH_TOKEN,
SENTINEL_GITHUB_TOKEN,
_InstallMethod,

46
tests/test_upgrade.py
View File

@@ -17,6 +17,7 @@ import pytest
from typer.testing import CliRunner
from specify_cli import app
from specify_cli._download_security import read_response_limited as _real_read_response_limited
from specify_cli._version import (
_fetch_latest_release_tag,
_get_installed_version,
@@ -24,7 +25,10 @@ from specify_cli._version import (
_normalize_tag,
)
from tests.conftest import strip_ansi
from tests.http_helpers import mock_urlopen_response
from tests.http_helpers import (
mock_urlopen_response,
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
)
runner = CliRunner()
@@ -235,6 +239,46 @@ class TestFailureCategorization:
_fetch_latest_release_tag()
class TestBoundedRead:
"""Regression test for the read_response_limited hardening.
A future refactor could silently revert `_fetch_latest_release_tag` to
`resp.read()` (the unbounded form) — this test pins the contract that
the response body is read through ``read_response_limited`` with a
bounded ``max_bytes``.
"""
def test_response_body_is_bounded(self):
recorded: dict[str, int | str] = {}
def _spy(response, *, max_bytes: int, label: str, **kwargs):
# max_bytes and label are keyword-only with no defaults: if the
# caller forgets to pass either, the call raises TypeError here
# (instead of recording a misleading None).
recorded["max_bytes"] = max_bytes
recorded["label"] = label
# Forward to the real implementation so the function under test
# still gets a parseable body.
return _real_read_response_limited(
response, max_bytes=max_bytes, label=label, **kwargs
)
with patch(
"specify_cli.authentication.http.urllib.request.urlopen",
return_value=mock_urlopen_response({"tag_name": "v9.9.9"}),
), patch("specify_cli._version.read_response_limited", side_effect=_spy):
tag, reason = _fetch_latest_release_tag()
assert tag == "v9.9.9"
assert reason is None
# The cap (1 MiB) is a deliberate ceiling for the GitHub release
# JSON — keep it explicit so a future refactor that drops the
# `max_bytes=` argument fails this test instead of regressing
# silently to the default.
assert recorded["max_bytes"] == 1024 * 1024
assert recorded["label"] == "GitHub latest release"
_FAILURE_CASES = [
("offline or timeout", urllib.error.URLError("down")),
(_RATE_LIMITED_REASON, _http_error(403)),

85
tests/test_workflows.py
View File

@@ -8658,18 +8658,15 @@ steps:
class FakeResponse:
def __init__(self, data, url=None):
self._data = data
self._pos = 0
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/42"
def read(self, amt=None):
if not hasattr(self, "_pos"):
self._pos = 0
if amt is None:
chunk = self._data[self._pos :]
self._pos = len(self._data)
return chunk
chunk = self._data[self._pos : self._pos + amt]
self._pos += len(chunk)
return chunk
def read(self, size=-1):
if size < 0:
size = len(self._data) - self._pos
out = self._data[self._pos : self._pos + size]
self._pos += len(out)
return out
def geturl(self):
return self._url
@@ -8729,18 +8726,15 @@ steps:
class FakeResponse:
def __init__(self, data, url=None):
self._data = data
self._pos = 0
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/42"
def read(self, amt=None):
if not hasattr(self, "_pos"):
self._pos = 0
if amt is None:
chunk = self._data[self._pos :]
self._pos = len(self._data)
return chunk
chunk = self._data[self._pos : self._pos + amt]
self._pos += len(chunk)
return chunk
def read(self, size=-1):
if size < 0:
size = len(self._data) - self._pos
out = self._data[self._pos : self._pos + size]
self._pos += len(out)
return out
def geturl(self):
return self._url
@@ -8780,18 +8774,15 @@ steps:
class FakeResponse:
def __init__(self, data, url=None):
self._data = data
self._pos = 0
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/55"
def read(self, amt=None):
if not hasattr(self, "_pos"):
self._pos = 0
if amt is None:
chunk = self._data[self._pos :]
self._pos = len(self._data)
return chunk
chunk = self._data[self._pos : self._pos + amt]
self._pos += len(chunk)
return chunk
def read(self, size=-1):
if size < 0:
size = len(self._data) - self._pos
out = self._data[self._pos : self._pos + size]
self._pos += len(out)
return out
def geturl(self):
return self._url
@@ -8873,18 +8864,15 @@ steps:
class FakeResponse:
def __init__(self, data, url=None):
self._data = data
self._pos = 0
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"
def read(self, amt=None):
if not hasattr(self, "_pos"):
self._pos = 0
if amt is None:
chunk = self._data[self._pos :]
self._pos = len(self._data)
return chunk
chunk = self._data[self._pos : self._pos + amt]
self._pos += len(chunk)
return chunk
def read(self, size=-1):
if size < 0:
size = len(self._data) - self._pos
out = self._data[self._pos : self._pos + size]
self._pos += len(out)
return out
def geturl(self):
return self._url
@@ -8936,18 +8924,15 @@ steps:
class FakeResponse:
def __init__(self, data, url=None):
self._data = data
self._pos = 0
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/55"
def read(self, amt=None):
if not hasattr(self, "_pos"):
self._pos = 0
if amt is None:
chunk = self._data[self._pos :]
self._pos = len(self._data)
return chunk
chunk = self._data[self._pos : self._pos + amt]
self._pos += len(chunk)
return chunk
def read(self, size=-1):
if size < 0:
size = len(self._data) - self._pos
out = self._data[self._pos : self._pos + size]
self._pos += len(out)
return out
def geturl(self):
return self._url