fix: address Copilot review on bounded reads and redirect-safety

- read_response_limited: read in a loop until EOF or one byte past the
  limit instead of a single read(max_bytes + 1). A server using chunked
  transfer encoding can return fewer bytes per read() than requested
  while streaming more than max_bytes total, defeating the single-read
  bound. Add regression tests for the short-read and within-limit paths.
- _download_security: annotate _raise / _raise_from as NoReturn so type
  checkers treat call sites as unreachable.
- Extract the duplicated is_https_or_localhost_http redirect-safety
  predicate into _download_security and import it from both _github_http
  and authentication/http so the rule lives in one place.
- azure_devops: stop catching broad ValueError/KeyError around token
  acquisition; give the bounded read a dedicated _TokenResponseTooLarge
  type and catch only URLError, OSError, JSONDecodeError, and that type
  so unrelated programming errors still surface.
- tests: make response mocks faithful streams (advancing cursor, b"" at
  EOF) so the bounded read loop terminates as it would against a real
  http.client.HTTPResponse.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: PascalThuet

src/specify_cli/_download_security.py

@@ -7,7 +7,8 @@
 import stat
 import zipfile
 from pathlib import Path, PurePosixPath
-from typing import TypeVar
+from typing import NoReturn, TypeVar
+from urllib.parse import urlparse
 
 
 ErrorT = TypeVar("ErrorT", bound=Exception)
@@ -20,11 +21,22 @@
 SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$")
 
 
-def _raise(error_type: type[ErrorT], message: str) -> None:
+def is_https_or_localhost_http(url: str) -> bool:
+    """Return True if *url* is HTTPS, or HTTP limited to loopback hosts.
+
+    Shared redirect-safety predicate used by the GitHub and auth HTTP redirect
+    handlers so the rule (and any future tightening of it) lives in one place.
+    """
+    parsed = urlparse(url)
+    is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
+    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 _raise_from(error_type: type[ErrorT], message: str, exc: Exception) -> None:
+def _raise_from(error_type: type[ErrorT], message: str, exc: Exception) -> NoReturn:
     raise error_type(message) from exc
 
 
@@ -35,11 +47,25 @@ def read_response_limited(
     error_type: type[ErrorT] = ValueError,
     label: str = "download",
 ) -> bytes:
-    """Read at most *max_bytes* from a response object."""
-    data = response.read(max_bytes + 1)
-    if len(data) > max_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.
+    """
+    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 data
+    return b"".join(chunks)
 
 
 def normalize_sha256(value: object, *, error_type: type[ErrorT] = ValueError) -> str | None:

src/specify_cli/_github_http.py

@@ -12,6 +12,8 @@
 from typing import Dict
 from urllib.parse import urlparse
 
+from specify_cli._download_security import is_https_or_localhost_http
+
 # GitHub-owned hostnames that should receive the Authorization header.
 # Includes codeload.github.com because GitHub archive URL downloads
 # (e.g. /archive/refs/tags/<tag>.zip) redirect there and require auth
@@ -55,12 +57,6 @@ def build_github_request(url: str) -> urllib.request.Request:
     return urllib.request.Request(url, headers=headers)
 
 
-def _is_https_or_localhost_http(url: str) -> bool:
-    parsed = urlparse(url)
-    is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
-    return parsed.scheme == "https" or (parsed.scheme == "http" and is_localhost)
-
-
 class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
     """Redirect handler that drops the Authorization header when leaving GitHub.
 
@@ -70,7 +66,7 @@ class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
     """
 
     def redirect_request(self, req, fp, code, msg, headers, newurl):
-        if not _is_https_or_localhost_http(newurl):
+        if not is_https_or_localhost_http(newurl):
             raise urllib.error.URLError(
                 f"Refusing unsafe redirect to non-HTTPS URL: {newurl}"
             )

src/specify_cli/authentication/azure_devops.py

@@ -17,6 +17,10 @@
 _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.
 
@@ -115,6 +119,7 @@ def _acquire_via_client_credentials(entry: AuthConfigEntry) -> str | None:
                 payload = _json.loads(
                     read_response_limited(
                         resp,
+                        error_type=_TokenResponseTooLarge,
                         label="Azure DevOps token response",
                     ).decode("utf-8")
                 )
@@ -123,7 +128,10 @@ def _acquire_via_client_credentials(entry: AuthConfigEntry) -> str | None:
         except (
             urllib.error.URLError,
             OSError,
-            ValueError,
-            KeyError,
+            _json.JSONDecodeError,
+            _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

src/specify_cli/authentication/http.py

@@ -16,6 +16,7 @@
 from fnmatch import fnmatch
 from urllib.parse import urlparse
 
+from .._download_security import is_https_or_localhost_http
 from . import get_provider
 from .config import AuthConfigEntry, _default_config_path, find_entries_for_url, load_auth_config
 
@@ -56,12 +57,6 @@ def _hostname_in_hosts(hostname: str, hosts: tuple[str, ...]) -> bool:
     return any(p == hostname or fnmatch(hostname, p) for p in hosts)
 
 
-def _is_https_or_localhost_http(url: str) -> bool:
-    parsed = urlparse(url)
-    is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
-    return parsed.scheme == "https" or (parsed.scheme == "http" and is_localhost)
-
-
 class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
     """Drop ``Authorization`` when a redirect leaves the entry's declared hosts."""
 
@@ -70,7 +65,7 @@ def __init__(self, hosts: tuple[str, ...]) -> None:
         self._hosts = hosts
 
     def redirect_request(self, req, fp, code, msg, headers, newurl):
-        if not _is_https_or_localhost_http(newurl):
+        if not is_https_or_localhost_http(newurl):
             raise urllib.error.URLError(
                 f"Refusing unsafe redirect to non-HTTPS URL: {newurl}"
             )

tests/integrations/test_integration_catalog.py

@@ -177,9 +177,16 @@ class FakeResponse:
             def __init__(self, data, url=""):
                 self._data = json.dumps(data).encode()
                 self._url = url if isinstance(url, str) else url.full_url
+                self._pos = 0
 
-            def read(self, _size=-1):
-                return self._data
+            def read(self, size=-1):
+                # Advance a cursor and return b"" at EOF like a real stream, so
+                # read_response_limited's bounded loop terminates.
+                if size is None or 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
@@ -553,9 +560,16 @@ class FakeResponse:
             def __init__(self, data, url=""):
                 self._data = json.dumps(data).encode()
                 self._url = url if isinstance(url, str) else url.full_url
-
-            def read(self, _size=-1):
-                return self._data
+                self._pos = 0
+
+            def read(self, size=-1):
+                # Advance a cursor and return b"" at EOF like a real stream, so
+                # read_response_limited's bounded loop terminates.
+                if size is None or 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

tests/test_authentication.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import base64
+import io
 import json
 import os
 
@@ -497,7 +498,7 @@ def test_resolve_token_azure_ad_success(self, monkeypatch):
             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):
@@ -825,7 +826,7 @@ def _capture_request(self):
         def side_effect(req, timeout=None):
             captured["request"] = req
             body = _json.dumps({"tag_name": "v9.9.9"}).encode()
-            resp = MagicMock(); resp.read.return_value = body
+            resp = MagicMock(); resp.read.side_effect = io.BytesIO(body).read
             cm = MagicMock(); cm.__enter__.return_value = resp; cm.__exit__.return_value = False
             return cm
         return captured, side_effect

tests/test_download_security.py

@@ -25,11 +25,23 @@
 
 
 class _Response:
-    def __init__(self, data: bytes):
+    """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:
-        return self.data if size < 0 else self.data[:size]
+        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
 
 
 class _CustomZipError(ValueError):
@@ -93,6 +105,19 @@ def test_read_response_limited_rejects_oversized_download():
         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)
+
+
 def test_remote_downloads_do_not_use_unbounded_response_reads():
     offenders = []
     for path in (REPO_ROOT / "src" / "specify_cli").rglob("*.py"):

tests/test_extensions.py

@@ -10,6 +10,7 @@
 """
 
 import pytest
+import io
 import json
 import hashlib
 import os
@@ -2820,7 +2821,7 @@ def test_fetch_single_catalog_sends_auth_header(self, temp_dir, monkeypatch):
 
         catalog_data = {"schema_version": "1.0", "extensions": {}}
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(catalog_data).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(catalog_data).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         mock_response.geturl.return_value = "https://raw.githubusercontent.com/org/repo/main/catalog.json"
@@ -2904,7 +2905,7 @@ def test_download_extension_sends_auth_header(self, temp_dir, monkeypatch):
         zip_bytes = zip_buf.getvalue()
 
         mock_response = MagicMock()
-        mock_response.read.return_value = zip_bytes
+        mock_response.read.side_effect = io.BytesIO(zip_bytes).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -2937,7 +2938,7 @@ def test_download_extension_verifies_sha256(self, temp_dir):
         catalog = self._make_catalog(temp_dir)
         zip_bytes = b"fake zip data"
         mock_response = MagicMock()
-        mock_response.read.return_value = zip_bytes
+        mock_response.read.side_effect = io.BytesIO(zip_bytes).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         ext_info = {
@@ -2960,7 +2961,7 @@ def test_download_extension_rejects_sha256_mismatch(self, temp_dir):
 
         catalog = self._make_catalog(temp_dir)
         mock_response = MagicMock()
-        mock_response.read.return_value = b"fake zip data"
+        mock_response.read.side_effect = io.BytesIO(b"fake zip data").read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         ext_info = {
@@ -4034,7 +4035,7 @@ def test_download_extension_allows_bundled_with_url(self, temp_dir):
         }
 
         mock_response = MagicMock()
-        mock_response.read.return_value = b"fake zip data"
+        mock_response.read.side_effect = io.BytesIO(b"fake zip data").read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 

tests/test_presets.py

@@ -11,6 +11,7 @@
 """
 
 import pytest
+import io
 import json
 import hashlib
 import tempfile
@@ -1522,7 +1523,7 @@ def test_fetch_single_catalog_sends_auth_header(self, project_dir, monkeypatch):
 
         catalog_data = {"schema_version": "1.0", "presets": {}}
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(catalog_data).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(catalog_data).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         mock_response.geturl.return_value = "https://raw.githubusercontent.com/org/repo/main/presets/catalog.json"
@@ -1563,7 +1564,7 @@ def test_download_pack_sends_auth_header(self, project_dir, monkeypatch):
         zip_bytes = zip_buf.getvalue()
 
         mock_response = MagicMock()
-        mock_response.read.return_value = zip_bytes
+        mock_response.read.side_effect = io.BytesIO(zip_bytes).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -1638,7 +1639,7 @@ def test_download_pack_verifies_sha256(self, project_dir):
         catalog = PresetCatalog(project_dir)
         zip_bytes = b"fake zip data"
         mock_response = MagicMock()
-        mock_response.read.return_value = zip_bytes
+        mock_response.read.side_effect = io.BytesIO(zip_bytes).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         pack_info = {
@@ -1662,7 +1663,7 @@ def test_download_pack_rejects_sha256_mismatch(self, project_dir):
 
         catalog = PresetCatalog(project_dir)
         mock_response = MagicMock()
-        mock_response.read.return_value = b"fake zip data"
+        mock_response.read.side_effect = io.BytesIO(b"fake zip data").read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         pack_info = {

tests/test_upgrade.py

@@ -8,6 +8,7 @@
 (if installed) with `--disable-socket` as an extra safety net.
 """
 
+import io
 import json
 import urllib.error
 import importlib.metadata
@@ -39,7 +40,10 @@
 def _mock_urlopen_response(payload: dict) -> MagicMock:
     body = json.dumps(payload).encode("utf-8")
     resp = MagicMock()
-    resp.read.return_value = body
+    # Back read() with a real stream so it advances and returns b"" at EOF,
+    # matching http.client.HTTPResponse (a fixed return_value would loop forever
+    # under read_response_limited's bounded read loop).
+    resp.read.side_effect = io.BytesIO(body).read
     cm = MagicMock()
     cm.__enter__.return_value = resp
     cm.__exit__.return_value = False