fix: stabilize security rebase follow-ups

Author: PascalThuet

src/specify_cli/authentication/http.py

@@ -72,10 +72,10 @@ def _validate_strict_redirect(_old_url: str, new_url: str) -> None:
 class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
     """Redirect handler that guards every redirect it is installed for.
 
-    1. Reject redirects that are not HTTPS with a hostname, except HTTP to
+    1. Run any caller-provided redirect validator.
+    2. Reject redirects that are not HTTPS with a hostname, except HTTP to
        localhost / 127.0.0.1 / ::1 (the exact hosts allowed by
        ``is_https_or_localhost_http``).
-    2. Run any caller-provided redirect validator.
     3. Drop ``Authorization`` when a redirect leaves trusted hosts or downgrades.
     """
 
@@ -89,9 +89,9 @@ def __init__(
         self._redirect_validator = redirect_validator
 
     def redirect_request(self, req, fp, code, msg, headers, newurl):
-        _validate_strict_redirect(req.full_url, newurl)
         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")
@@ -164,6 +164,7 @@ def open_url(
     entries = find_entries_for_url(url, _load_config())
 
     effective_redirect_validator = redirect_validator
+    use_redirect_handler = strict_redirects or effective_redirect_validator is not None
 
     def _make_req(auth_headers: dict[str, str]) -> urllib.request.Request:
         merged = {}
@@ -195,7 +196,7 @@ def _make_req(auth_headers: dict[str, str]) -> urllib.request.Request:
 
     # No entry worked (or none matched) — unauthenticated fallback
     req = _make_req({})
-    if effective_redirect_validator is not None:
+    if use_redirect_handler:
         # 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((), effective_redirect_validator))

src/specify_cli/presets/__init__.py

@@ -238,6 +238,7 @@ def _validate(self):
                     f"Invalid template file path '{file_path}': "
                     "must be a relative path within the preset directory"
                 )
+            tmpl["file"] = normalized
 
             # Validate strategy field (optional, defaults to "replace")
             strategy = tmpl.get("strategy", "replace")

tests/integrations/test_integration_catalog.py

@@ -340,7 +340,8 @@ def __exit__(self, *_args):
                 pass
 
         def fake_urlopen(url, timeout=10):
-            assert url == entry.url
+            actual_url = url.full_url if hasattr(url, "full_url") else url
+            assert actual_url == entry.url
             assert timeout == 10
             return FakeResponse()
 

tests/test_authentication.py

@@ -799,17 +799,18 @@ def test_redirect_outside_hosts_strips_auth(self):
         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

tests/test_extensions.py

@@ -3270,7 +3270,7 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, temp_dir, payload)
         catalog = self._make_catalog(temp_dir)
 
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -3337,10 +3337,12 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload(
             "schema_version": "1.0",
             "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
         }
-        mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(valid).encode()
-        mock_response.__enter__ = lambda s: s
-        mock_response.__exit__ = MagicMock(return_value=False)
+        def make_response():
+            mock_response = MagicMock()
+            mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read
+            mock_response.__enter__ = lambda s: s
+            mock_response.__exit__ = MagicMock(return_value=False)
+            return mock_response
 
         entry = CatalogEntry(
             url=ExtensionCatalog.DEFAULT_CATALOG_URL,
@@ -3349,7 +3351,7 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload(
             install_allowed=True,
         )
 
-        with patch.object(catalog, "_open_url", return_value=mock_response):
+        with patch.object(catalog, "_open_url", side_effect=lambda *a, **kw: make_response()):
             result = catalog._fetch_single_catalog(entry, force_refresh=False)
 
         # The poisoned cache was discarded and the network payload returned.
@@ -3385,7 +3387,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, temp_dir, payload):
 
         catalog = self._make_catalog(temp_dir)
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -3424,12 +3426,14 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, temp_dir):
             "schema_version": "1.0",
             "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
         }
-        mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(valid).encode()
-        mock_response.__enter__ = lambda s: s
-        mock_response.__exit__ = MagicMock(return_value=False)
+        def make_response():
+            mock_response = MagicMock()
+            mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read
+            mock_response.__enter__ = lambda s: s
+            mock_response.__exit__ = MagicMock(return_value=False)
+            return mock_response
 
-        with patch.object(catalog, "_open_url", return_value=mock_response):
+        with patch.object(catalog, "_open_url", side_effect=lambda *a, **kw: make_response()):
             result = catalog.fetch_catalog(force_refresh=False)
 
         # Recovered via network rather than crashing on the unreadable cache.
@@ -3463,7 +3467,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, temp_dir):
             "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
         }
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(valid).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -3537,7 +3541,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, temp_dir, monkeypatch):
             "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
         }
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(payload).encode("utf-8")
+        mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode("utf-8")).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -3587,10 +3591,12 @@ def test_fetch_catalog_survives_unwritable_cache(self, temp_dir, monkeypatch):
             "schema_version": "1.0",
             "extensions": {"foo": {"name": "Foo", "version": "1.0.0"}},
         }
-        mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(valid).encode()
-        mock_response.__enter__ = lambda s: s
-        mock_response.__exit__ = MagicMock(return_value=False)
+        def make_response():
+            mock_response = MagicMock()
+            mock_response.read.side_effect = io.BytesIO(json.dumps(valid).encode()).read
+            mock_response.__enter__ = lambda s: s
+            mock_response.__exit__ = MagicMock(return_value=False)
+            return mock_response
 
         # Simulate an unwritable cache dir: every write_text under the
         # cache directory raises PermissionError (an OSError subclass).
@@ -3603,7 +3609,7 @@ def failing_write_text(self, data, *args, **kwargs):
 
         monkeypatch.setattr(_PathCls, "write_text", failing_write_text)
 
-        with patch.object(catalog, "_open_url", return_value=mock_response):
+        with patch.object(catalog, "_open_url", side_effect=lambda *a, **kw: make_response()):
             # Legacy single-catalog path.
             assert catalog.fetch_catalog(force_refresh=True) == valid
 
@@ -3639,7 +3645,7 @@ def test_get_merged_extensions_skips_non_mapping_entries(self, temp_dir):
             },
         }
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(payload).encode()
+        mock_response.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
 
@@ -3777,7 +3783,7 @@ def test_download_extension_accepts_direct_github_rest_asset_url(self, temp_dir,
         zip_bytes = zip_buf.getvalue()
 
         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)
 

tests/test_github_http.py

@@ -94,7 +94,7 @@ def _make_open_url_fn(self, release_json):
         @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
 
@@ -148,7 +148,7 @@ def test_returns_none_on_network_error(self):
         @contextmanager
         def failing_open(url, timeout=None, extra_headers=None):
             raise urllib.error.URLError("network error")
-            yield  # noqa: unreachable
+            yield  # pragma: no cover
 
         result = resolve_github_release_asset_api_url(
             "https://github.com/org/repo/releases/download/v1/pack.zip",
@@ -164,7 +164,7 @@ def test_tag_with_special_characters_is_url_encoded(self):
         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(
@@ -183,7 +183,7 @@ def test_tag_with_hash_is_url_encoded(self):
         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(