diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 329aaf8192..082189a177 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -523,8 +523,20 @@ def _validate_catalog_url(url: str) -> None: _validate_catalog_url(entry.url) + # Validate EVERY redirect hop, not just the final URL: _open_url follows + # redirects, so an https:// entry that 30x-redirects through http:// (or + # to a non-HTTPS host mid-chain) could otherwise let a network attacker + # rewrite the next hop and slip a payload past a final-URL-only check. + # redirect_validator runs before each hop; the geturl() check below is + # retained as a defense-in-depth backstop. Mirrors the presets/extensions + # catalog fix (#3523 / #3524). + def _validate_redirect(_old_url: str, new_url: str) -> None: + _validate_catalog_url(new_url) + try: - with _open_url(entry.url, timeout=30) as resp: + with _open_url( + entry.url, timeout=30, redirect_validator=_validate_redirect + ) as resp: _validate_catalog_url(resp.geturl()) data = json.loads(resp.read().decode("utf-8")) except Exception as exc: @@ -1180,8 +1192,20 @@ def _validate_url(url: str) -> None: _validate_url(entry.url) + # Validate EVERY redirect hop, not just the final URL: _open_url follows + # redirects, so an https:// entry that 30x-redirects through http:// (or + # to a non-HTTPS host mid-chain) could otherwise let a network attacker + # rewrite the next hop and slip a payload past a final-URL-only check. + # redirect_validator runs before each hop; the geturl() check below is + # retained as a defense-in-depth backstop. Mirrors the presets/extensions + # catalog fix (#3523 / #3524). + def _validate_redirect(_old_url: str, new_url: str) -> None: + _validate_url(new_url) + try: - with _open_url(entry.url, timeout=30) as resp: + with _open_url( + entry.url, timeout=30, redirect_validator=_validate_redirect + ) as resp: _validate_url(resp.geturl()) data = json.loads(resp.read().decode("utf-8")) except Exception as exc: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index cd1b235411..20baa1b6ec 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -6161,7 +6161,9 @@ def geturl(self): return "https://[::1" monkeypatch.setattr( - auth_http, "open_url", lambda url, timeout=30: _FakeResponse() + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(), ) catalog = WorkflowCatalog(project_dir) @@ -6176,6 +6178,41 @@ def geturl(self): with pytest.raises(WorkflowCatalogError, match="malformed"): catalog._fetch_single_catalog(entry, force_refresh=True) + def test_fetch_validates_every_redirect_hop(self, project_dir, monkeypatch): + """A redirect_validator is passed to open_url and rejects a non-HTTPS + INTERMEDIATE hop — closing the https -> http -> attacker-https chain a + terminal-URL-only check would miss. Mirrors presets/extensions + (#3523 / #3524).""" + from specify_cli.workflows.catalog import ( + WorkflowCatalog, + WorkflowCatalogEntry, + WorkflowCatalogError, + ) + from specify_cli.authentication import http as auth_http + + captured = {} + + def fake_open(url, timeout=30, 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") + + monkeypatch.setattr(auth_http, "open_url", fake_open) + + catalog = WorkflowCatalog(project_dir) + entry = WorkflowCatalogEntry( + url="https://good.example/catalog.json", + name="test", + priority=1, + install_allowed=True, + ) + with pytest.raises(WorkflowCatalogError, match="HTTPS"): + catalog._fetch_single_catalog(entry, force_refresh=True) + assert captured["rv"] is not None + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import WorkflowCatalog @@ -6720,7 +6757,9 @@ def geturl(self): return "https://[not-an-ip]/x" monkeypatch.setattr( - auth_http, "open_url", lambda url, timeout=30: _FakeResponse() + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(), ) catalog = StepCatalog(project_dir) @@ -6735,6 +6774,41 @@ def geturl(self): with pytest.raises(StepCatalogError, match="malformed"): catalog._fetch_single_catalog(entry, force_refresh=True) + def test_fetch_validates_every_redirect_hop(self, project_dir, monkeypatch): + """A redirect_validator is passed to open_url and rejects a non-HTTPS + INTERMEDIATE hop — closing the https -> http -> attacker-https chain a + terminal-URL-only check would miss. Mirrors presets/extensions + (#3523 / #3524).""" + from specify_cli.workflows.catalog import ( + StepCatalog, + StepCatalogEntry, + StepCatalogError, + ) + from specify_cli.authentication import http as auth_http + + captured = {} + + def fake_open(url, timeout=30, redirect_validator=None): + captured["rv"] = redirect_validator + # Simulate the hop urllib validates before following the redirect. + redirect_validator( + "https://good.example/steps.json", "http://evil.test/hop" + ) + raise AssertionError("redirect_validator should have raised") + + monkeypatch.setattr(auth_http, "open_url", fake_open) + + catalog = StepCatalog(project_dir) + entry = StepCatalogEntry( + url="https://good.example/steps.json", + name="test", + priority=1, + install_allowed=True, + ) + with pytest.raises(StepCatalogError, match="HTTPS"): + catalog._fetch_single_catalog(entry, force_refresh=True) + assert captured["rv"] is not None + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import StepCatalog