Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/specify_cli/extensions/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2279,6 +2279,16 @@ def _fetch_single_catalog(
# Fetch from network
try:
with self._open_url(entry.url, timeout=10) as response:
# Re-validate the URL after any redirects: _open_url follows
# redirects (stripping auth only on an HTTPS->HTTP downgrade), so
# without this an https:// catalog entry that 30x-redirects to
# http://attacker/... would be fetched and trusted. The payload
# supplies each extension's download_url + sha256, so a redirected
# payload defeats sha256 verification. Mirrors the
# integrations/presets/workflows catalog fetchers.
final_url = response.geturl()
if final_url != entry.url:
self._validate_catalog_url(final_url)
catalog_data = json.loads(response.read())

self._validate_catalog_payload(catalog_data, entry.url)
Expand Down
38 changes: 38 additions & 0 deletions tests/test_extensions.py
Original file line number Diff line number Diff line change
Expand Up @@ -3650,6 +3650,35 @@ def fake_open(req, timeout=None):

assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken"

def test_fetch_single_catalog_revalidates_redirected_url(self, temp_dir):
"""An HTTPS catalog URL that redirects to http:// must be rejected AFTER
the redirect. _open_url follows redirects (auth stripped on downgrade),
so without re-validating response.geturl() the http payload would still
be fetched and trusted — and it supplies each extension's download_url +
sha256, defeating sha256 verification. Parity with the
integrations/presets/workflows catalog fetchers."""
from unittest.mock import patch, MagicMock

catalog = self._make_catalog(temp_dir)

mock_response = MagicMock()
mock_response.read.return_value = json.dumps(
{"schema_version": "1.0", "extensions": {}}
).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "http://evil.test/catalog.json"

entry = CatalogEntry(
url="https://good.example/catalog.json",
name="c",
priority=1,
install_allowed=True,
)
with patch.object(catalog, "_open_url", return_value=mock_response):
with pytest.raises(ExtensionError, match="HTTPS"):
catalog._fetch_single_catalog(entry, force_refresh=True)

@pytest.mark.parametrize(
"payload",
[
Expand Down Expand Up @@ -3683,6 +3712,7 @@ def test_fetch_single_catalog_rejects_malformed_payload(self, temp_dir, payload)
mock_response.read.return_value = json.dumps(payload).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

entry = CatalogEntry(
url="https://example.com/catalog.json",
Expand Down Expand Up @@ -3751,6 +3781,7 @@ def test_fetch_single_catalog_rejects_malformed_cached_payload(
mock_response.read.return_value = json.dumps(valid).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

entry = CatalogEntry(
url=ExtensionCatalog.DEFAULT_CATALOG_URL,
Expand Down Expand Up @@ -3798,6 +3829,7 @@ def test_fetch_catalog_rejects_malformed_payload(self, temp_dir, payload):
mock_response.read.return_value = json.dumps(payload).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

with patch.object(catalog, "_open_url", return_value=mock_response):
with pytest.raises(ExtensionError, match="Invalid catalog format"):
Expand Down Expand Up @@ -3838,6 +3870,7 @@ def test_fetch_catalog_recovers_from_unreadable_cache(self, temp_dir):
mock_response.read.return_value = json.dumps(valid).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

with patch.object(catalog, "_open_url", return_value=mock_response):
result = catalog.fetch_catalog(force_refresh=False)
Expand Down Expand Up @@ -3876,6 +3909,7 @@ def test_fetch_catalog_recovers_from_unreadable_metadata(self, temp_dir):
mock_response.read.return_value = json.dumps(valid).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

with patch.object(catalog, "_open_url", return_value=mock_response):
result = catalog.fetch_catalog(force_refresh=False)
Expand Down Expand Up @@ -3950,6 +3984,7 @@ def test_fetch_catalog_writes_cache_as_utf8(self, temp_dir, monkeypatch):
mock_response.read.return_value = json.dumps(payload).encode("utf-8")
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

# Record every ``write_text`` call's encoding kwarg so the
# assertion observes the production writer's argument directly.
Expand Down Expand Up @@ -4001,6 +4036,7 @@ def test_fetch_catalog_survives_unwritable_cache(self, temp_dir, monkeypatch):
mock_response.read.return_value = json.dumps(valid).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

# Simulate an unwritable cache dir: every write_text under the
# cache directory raises PermissionError (an OSError subclass).
Expand Down Expand Up @@ -4052,6 +4088,7 @@ def test_get_merged_extensions_skips_non_mapping_entries(self, temp_dir):
mock_response.read.return_value = json.dumps(payload).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

entry = CatalogEntry(
url="https://example.com/catalog.json",
Expand Down Expand Up @@ -6030,6 +6067,7 @@ def test_download_extension_allows_bundled_with_url(self, temp_dir):
mock_response.read.return_value = b"fake zip data"
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "https://example.com/catalog.json"

with patch.object(catalog, "get_extension_info", return_value=bundled_with_url), \
patch.object(urllib.request, "urlopen", return_value=mock_response):
Expand Down