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(presets): re-validate catalog URL after redirects (HTTPS parity/security) (#3523)

Browse Source 
* fix(presets): re-validate catalog URL after redirects (HTTPS parity)

PresetCatalog._fetch_single_catalog opened the catalog URL and trusted the
payload without re-validating response.geturl() after redirects. _open_url
follows redirects (stripping auth only on an HTTPS->HTTP downgrade), so an
https:// catalog entry that 30x-redirects to http://attacker/... was still
fetched and trusted. The catalog payload supplies each preset's download_url +
sha256, so a redirected payload can drive install of an arbitrary archive that
passes verify_archive_sha256.

Add the post-redirect geturl() re-validation via _validate_catalog_url,
mirroring integrations/catalog.py, workflows/catalog.py, and bundler adapters —
and presets/_commands.py, which already does this on its --from download path.
This is the lone preset catalog-fetch site missing the guard.

Test: an HTTPS URL whose response.geturl() reports http:// is rejected
(PresetValidationError). Completed four existing fetch-test mocks that predated
this behavior to report geturl() like a real urllib response.

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

* fix(presets): validate every redirect hop + guard the legacy fetch_catalog path

Two follow-ups to the catalog redirect hardening:

1. Validate every redirect hop, not just the terminal URL. A final-geturl-only
   check passes an https -> http -> attacker-controlled-https chain: the insecure
   intermediate hop lets a network attacker rewrite the next redirect. _open_url
   now forwards a redirect_validator to open_url (called before each hop), and
   _fetch_single_catalog passes _validate_catalog_url through it while retaining
   the final geturl() check — mirroring bundler/services/adapters.py.

2. The legacy public fetch_catalog() single-catalog path parsed response.read()
   with no redirect check at all. Give it the same redirect_validator + final
   geturl() validation.

Tests: a stubbed intermediate http hop is rejected (redirect_validator), and the
legacy fetch_catalog() rejects an HTTPS->http redirected payload (fail before:
no raise). Existing fetch-test mocks updated to accept the redirect_validator
kwarg and report geturl() like a real response. Full test_presets.py (365) green.

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

* test(presets): drop duplicate geturl mock; add legacy fetch_catalog per-hop test

- Remove the duplicate mock_response.geturl.return_value assignment left by the
  geturl mock-completion pass (the explanatory comment was stranded between the
  two identical assignments); keep a single assignment after the comment.
- Add test_fetch_catalog_legacy_validates_every_redirect_hop so the legacy
  fetch_catalog() path is verified to supply the redirect_validator (rejecting an
  insecure intermediate hop), not just the terminal geturl() — parity with
  _fetch_single_catalog and the #3524 sibling.

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ali jawwad
2026-07-21 22:00:53 +05:00
committed by GitHub
parent 2a0ada9a6a
commit 1f7290c975
2 changed files with 144 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
src/specify_cli/presets/__init__.py
View File

@@ -2131,13 +2131,22 @@ class PresetCatalog:
url: str,
timeout: int = 10,
extra_headers: Optional[Dict[str, str]] = None,
redirect_validator=None,
):
"""Open a URL with provider-based auth, trying each configured provider.
Delegates to :func:`specify_cli.authentication.http.open_url`.
*redirect_validator*, when provided, is invoked as ``(old_url, new_url)``
before EACH redirect hop, so an HTTPS host guarantee can be enforced on
every intermediate URL, not just the terminal one.
"""
from specify_cli.authentication.http import open_url
return open_url(url, timeout, extra_headers=extra_headers)
return open_url(
url,
timeout,
extra_headers=extra_headers,
redirect_validator=redirect_validator,
)
def _resolve_github_release_asset_api_url(
self,
@@ -2427,7 +2436,21 @@ class PresetCatalog:
pass
try:
with self._open_url(entry.url, timeout=10) as response:
# Validate EVERY redirect hop (not just the terminal URL): an
# https -> http -> attacker-controlled-https chain would pass a
# final-URL-only check while the insecure intermediate hop lets a
# network attacker rewrite the next redirect. redirect_validator runs
# before each hop; the final geturl() check is retained as a
# belt-and-braces guard. Mirrors bundler/services/adapters.py.
def _validate_redirect(_old_url: str, new_url: str) -> None:
self._validate_catalog_url(new_url)
with self._open_url(
entry.url, timeout=10, redirect_validator=_validate_redirect
) as response:
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)
@@ -2578,7 +2601,18 @@ class PresetCatalog:
pass
try:
with self._open_url(catalog_url, timeout=10) as response:
# Same redirect hardening as _fetch_single_catalog: validate every
# redirect hop AND the final URL so this legacy single-catalog path
# is not vulnerable to an HTTPS->HTTP redirected payload either.
def _validate_redirect(_old_url: str, new_url: str) -> None:
self._validate_catalog_url(new_url)
with self._open_url(
catalog_url, timeout=10, redirect_validator=_validate_redirect
) as response:
final_url = response.geturl()
if final_url != catalog_url:
self._validate_catalog_url(final_url)
catalog_data = json.loads(response.read())
# Validate catalog structure. Reuses the same helper as

107
tests/test_presets.py
View File

@@ -1766,6 +1766,103 @@ class TestPresetCatalog:
assert captured["req"].get_header("Authorization") == "Bearer ghp_testtoken"
def test_fetch_single_catalog_revalidates_redirected_url(self, project_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 preset's download_url +
sha256, defeating verify_archive_sha256. Parity with the
integrations/workflows catalog fetchers."""
catalog = PresetCatalog(project_dir)
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps({"schema_version": "1.0", "presets": {}}).encode()
def geturl(self):
return "http://evil.test/catalog.json" # downgraded via redirect
catalog._open_url = lambda url, timeout=None, redirect_validator=None: _Resp()
entry = PresetCatalogEntry(
url="https://good.example/catalog.json",
name="c",
priority=1,
install_allowed=True,
)
with pytest.raises(PresetValidationError, match="HTTPS"):
catalog._fetch_single_catalog(entry, force_refresh=True)
def test_fetch_single_catalog_validates_every_redirect_hop(self, project_dir):
"""A redirect_validator is passed to _open_url and rejects a non-HTTPS
INTERMEDIATE hop — closing the https -> http -> attacker-https chain that
a terminal-URL-only check would miss."""
catalog = PresetCatalog(project_dir)
captured = {}
def fake_open(url, timeout=None, redirect_validator=None):
captured["rv"] = redirect_validator
# Simulate the hop urllib validates before following the redirect.
redirect_validator("https://good.example/catalog.json", "http://evil.test/hop")
raise AssertionError("redirect_validator should have raised")
catalog._open_url = fake_open
entry = PresetCatalogEntry(
url="https://good.example/catalog.json",
name="c",
priority=1,
install_allowed=True,
)
with pytest.raises(PresetValidationError, match="HTTPS"):
catalog._fetch_single_catalog(entry, force_refresh=True)
assert captured["rv"] is not None
def test_fetch_catalog_legacy_revalidates_redirected_url(self, project_dir):
"""The legacy single-catalog fetch_catalog() path also rejects an
HTTPS -> http redirected payload (final geturl() check), matching
_fetch_single_catalog — it previously parsed the body with no check."""
catalog = PresetCatalog(project_dir)
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps({"schema_version": "1.0", "presets": {}}).encode()
def geturl(self):
return "http://evil.test/catalog.json"
catalog._open_url = lambda url, timeout=None, redirect_validator=None: _Resp()
with pytest.raises(PresetError, match="HTTPS"):
catalog.fetch_catalog(force_refresh=True)
def test_fetch_catalog_legacy_validates_every_redirect_hop(self, project_dir):
"""The legacy fetch_catalog() path also validates every INTERMEDIATE hop
(not just the terminal URL): it must supply a redirect_validator that
rejects an insecure hop, so an https -> http -> https chain is caught."""
catalog = PresetCatalog(project_dir)
captured = {}
def fake_open(url, timeout=None, redirect_validator=None):
captured["rv"] = redirect_validator
redirect_validator(url, "http://evil.test/hop")
raise AssertionError("redirect_validator should have raised")
catalog._open_url = fake_open
with pytest.raises(PresetError, match="HTTPS"):
catalog.fetch_catalog(force_refresh=True)
assert captured["rv"] is not None
@pytest.mark.parametrize(
"payload",
[
@@ -1799,6 +1896,9 @@ class TestPresetCatalog:
mock_response.read.return_value = json.dumps(payload).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
# A real urllib response reports the final URL (== request URL with no
# redirect); the fetcher re-validates it after redirects.
mock_response.geturl.return_value = "https://example.com/catalog.json"
entry = PresetCatalogEntry(
url="https://example.com/catalog.json",
@@ -1868,6 +1968,7 @@ class TestPresetCatalog:
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 = catalog.DEFAULT_CATALOG_URL
entry = PresetCatalogEntry(
url=catalog.DEFAULT_CATALOG_URL,
@@ -1915,6 +2016,7 @@ class TestPresetCatalog:
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(PresetError, match="Invalid preset catalog format"):
@@ -1956,6 +2058,7 @@ class TestPresetCatalog:
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)
@@ -1994,6 +2097,7 @@ class TestPresetCatalog:
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)
@@ -2064,6 +2168,7 @@ class TestPresetCatalog:
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.
@@ -2113,6 +2218,7 @@ class TestPresetCatalog:
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 = catalog.DEFAULT_CATALOG_URL
# Simulate an unwritable cache dir: every write_text under the
# cache directory raises PermissionError (an OSError subclass).
@@ -2165,6 +2271,7 @@ class TestPresetCatalog:
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 = PresetCatalogEntry(
url="https://example.com/catalog.json",