diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 03d5a4b8a..71422b623 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -108,6 +108,26 @@ _WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"}) +def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None: + """Reject a redirect before it is followed unless HTTPS (or loopback HTTP).""" + import urllib.error + from ipaddress import ip_address + from urllib.parse import urlparse + + parsed = urlparse(new_url) + host = parsed.hostname or "" + loopback = host == "localhost" + if not loopback: + try: + loopback = ip_address(host).is_loopback + except ValueError: + pass + if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback): + raise urllib.error.URLError( + "redirect target must use HTTPS, or HTTP for localhost/loopback" + ) + + def _validate_workflow_id_or_exit(workflow_id: str) -> None: """Validate that ``workflow_id`` is a safe installed-workflow directory name.""" if ( @@ -756,7 +776,12 @@ def workflow_add( import tempfile try: - with _open_url(download_url, timeout=30, extra_headers=_wf_url_extra_headers) as resp: + with _open_url( + download_url, + timeout=30, + extra_headers=_wf_url_extra_headers, + redirect_validator=_reject_insecure_download_redirect, + ) as resp: final_url = resp.geturl() final_parsed = urlparse(final_url) final_host = final_parsed.hostname or "" @@ -903,7 +928,12 @@ def _install_workflow_from_catalog( _wf_cat_extra_headers = {"Accept": "application/octet-stream"} workflow_dir.mkdir(parents=True, exist_ok=True) - with _open_url(workflow_url, timeout=30, extra_headers=_wf_cat_extra_headers) as response: + with _open_url( + workflow_url, + timeout=30, + extra_headers=_wf_cat_extra_headers, + redirect_validator=_reject_insecure_download_redirect, + ) as response: # Validate final URL after redirects final_url = response.geturl() final_parsed = urlparse(final_url) @@ -1621,7 +1651,9 @@ def workflow_step_add( raise ValueError(f"Refusing to fetch from non-HTTPS URL: {url}") if not parsed.hostname: raise ValueError(f"Refusing to fetch from URL with no hostname: {url}") - with _open_url(url, timeout=30) as resp: + with _open_url( + url, timeout=30, redirect_validator=_reject_insecure_download_redirect + ) as resp: final_url = resp.geturl() final_parsed = urlparse(final_url) final_is_localhost = final_parsed.hostname in ("localhost", "127.0.0.1", "::1") diff --git a/tests/test_workflows.py b/tests/test_workflows.py index d8c702b18..cad52b148 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -6579,7 +6579,7 @@ steps: def __exit__(self, *a): return False - def fake_open_url(url, timeout=None, extra_headers=None): + def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None): captured_urls.append((url, extra_headers, timeout)) if "releases/tags/" in url: return FakeResponse(json.dumps({ @@ -6631,7 +6631,7 @@ steps: def __exit__(self, *a): return False - def fake_open_url(url, timeout=None, extra_headers=None): + def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None): captured_urls.append((url, extra_headers)) return FakeResponse(self.VALID_WORKFLOW_YAML.encode()) @@ -6674,7 +6674,7 @@ steps: def __exit__(self, *a): return False - def fake_open_url(url, timeout=None, extra_headers=None): + def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None): captured_urls.append((url, extra_headers)) if "releases/tags/" in url: return FakeResponse(json.dumps({ @@ -6750,7 +6750,7 @@ steps: def __exit__(self, *a): return False - def fake_open_url(url, timeout=None, extra_headers=None): + def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None): captured_urls.append((url, extra_headers)) if "releases/tags/" in url: return FakeResponse(json.dumps({ @@ -6818,7 +6818,7 @@ steps: run: "echo hello" """ - def fake_open_url(url, timeout=None, extra_headers=None): + def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None): captured_urls.append((url, extra_headers)) if "releases/tags/" in url: return FakeResponse(json.dumps({ @@ -7364,7 +7364,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), ): result = runner.invoke( app, @@ -7384,7 +7384,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), ): result = runner.invoke( app, @@ -7418,7 +7418,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", redirected_url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(b"", redirected_url), ): result = runner.invoke( app, @@ -7436,7 +7436,7 @@ steps: monkeypatch.chdir(project_dir) calls: list[str] = [] - def _fake_open(url, timeout=None, extra_headers=None): + def _fake_open(url, timeout=None, extra_headers=None, redirect_validator=None): calls.append(url) raise AssertionError(f"network fetch attempted: {url}") @@ -7601,6 +7601,51 @@ steps: assert result.exception is None or isinstance(result.exception, SystemExit) assert "No workflow.yml found" in result.output + def test_download_redirect_validator_rejects_http_before_follow(self): + import urllib.error + + from specify_cli.workflows._commands import _reject_insecure_download_redirect + + with pytest.raises(urllib.error.URLError): + _reject_insecure_download_redirect( + "https://example.com/wf.yml", "http://evil.example.com/wf.yml" + ) + # Allowed: HTTPS anywhere, HTTP on loopback. + _reject_insecure_download_redirect( + "https://example.com/wf.yml", "https://cdn.example.com/wf.yml" + ) + _reject_insecure_download_redirect( + "https://example.com/wf.yml", "http://localhost:8000/wf.yml" + ) + _reject_insecure_download_redirect( + "https://example.com/wf.yml", "http://127.0.0.1/wf.yml" + ) + + def test_add_from_url_passes_redirect_validator(self, project_dir, monkeypatch): + from unittest.mock import patch + + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + data = self.WORKFLOW_YAML.format(version="1.0.0").encode() + seen: dict[str, object] = {} + + def fake_open(url, timeout=None, extra_headers=None, redirect_validator=None): + seen["validator"] = redirect_validator + return self._FakeResponse(data, url) + + runner = CliRunner() + with patch("specify_cli.authentication.http.open_url", side_effect=fake_open): + result = runner.invoke( + app, + ["workflow", "add", "align-wf", "--from", "https://example.com/workflow.yml"], + ) + assert result.exit_code == 0, result.output + from specify_cli.workflows._commands import _reject_insecure_download_redirect + + assert seen["validator"] is _reject_insecure_download_redirect + def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch): """A failed dump must not truncate the persisted registry.""" from specify_cli.workflows.catalog import WorkflowRegistry @@ -7712,7 +7757,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), ): result = runner.invoke(app, ["workflow", "update"], input="y\n") assert result.exit_code == 0, result.output @@ -7760,7 +7805,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(b"", url), ), patch.object( WorkflowDefinition, "from_yaml", @@ -7901,7 +7946,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), ): result = runner.invoke(app, ["workflow", "update"], input="y\n") assert "does not match the catalog version" in result.output @@ -7939,7 +7984,7 @@ steps: runner = CliRunner() with patch( "specify_cli.authentication.http.open_url", - side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), ): result = runner.invoke(app, ["workflow", "update"], input="y\n") assert result.exit_code == 0, result.output @@ -8187,7 +8232,7 @@ steps: }, ) - def boom(url, timeout=None, extra_headers=None): + def boom(url, timeout=None, extra_headers=None, redirect_validator=None): raise OSError("network down") runner = CliRunner()