diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 294dd8759a..0072f2e545 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -165,8 +165,20 @@ def _validate_catalog_url(self, url: str) -> None: """Validate that a catalog URL uses HTTPS (localhost HTTP allowed).""" from urllib.parse import urlparse - parsed = urlparse(url) - is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") + # A malformed authority (e.g. an unterminated IPv6 bracket + # "https://[::1") makes urlparse / hostname access raise ValueError. + # This validator's contract is to raise WorkflowValidationError for a + # bad URL, so surface that rather than leaking a raw ValueError past the + # command handler (which only catches WorkflowValidationError). Mirrors + # specify_cli.catalogs (#3435). + try: + parsed = urlparse(url) + hostname = parsed.hostname + except ValueError: + raise WorkflowValidationError( + f"Catalog URL is malformed: {url}" + ) from None + is_localhost = hostname in ("localhost", "127.0.0.1", "::1") if parsed.scheme != "https" and not ( parsed.scheme == "http" and is_localhost ): @@ -174,7 +186,7 @@ def _validate_catalog_url(self, url: str) -> None: f"Catalog URL must use HTTPS (got {parsed.scheme}://). " "HTTP is only allowed for localhost." ) - if not parsed.hostname: + if not hostname: raise WorkflowValidationError( "Catalog URL must be a valid URL with a host." ) @@ -340,15 +352,26 @@ def _fetch_single_catalog( from specify_cli.authentication.http import open_url as _open_url def _validate_catalog_url(url: str) -> None: - parsed = urlparse(url) - is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") + # A malformed authority (e.g. "https://[::1") makes urlparse / + # hostname access raise ValueError; treat it as a refused fetch + # rather than leaking a raw ValueError (this also validates the + # post-redirect resp.geturl(), so a hostile redirect target cannot + # crash the fetch either). + try: + parsed = urlparse(url) + hostname = parsed.hostname + except ValueError: + raise WorkflowCatalogError( + f"Refusing to fetch catalog from malformed URL: {url}" + ) from None + is_localhost = hostname in ("localhost", "127.0.0.1", "::1") if parsed.scheme != "https" and not ( parsed.scheme == "http" and is_localhost ): raise WorkflowCatalogError( f"Refusing to fetch catalog from non-HTTPS URL: {url}" ) - if not parsed.hostname: + if not hostname: raise WorkflowCatalogError( f"Refusing to fetch catalog from URL with no hostname: {url}" ) @@ -782,8 +805,20 @@ def _validate_catalog_url(self, url: str) -> None: """Validate that a catalog URL uses HTTPS (localhost HTTP allowed).""" from urllib.parse import urlparse - parsed = urlparse(url) - is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") + # A malformed authority (e.g. an unterminated IPv6 bracket + # "https://[::1") makes urlparse / hostname access raise ValueError. + # This validator's contract is to raise StepValidationError for a bad + # URL, so surface that rather than leaking a raw ValueError past the + # command handler (which only catches StepValidationError). Mirrors + # specify_cli.catalogs (#3435). + try: + parsed = urlparse(url) + hostname = parsed.hostname + except ValueError: + raise StepValidationError( + f"Catalog URL is malformed: {url}" + ) from None + is_localhost = hostname in ("localhost", "127.0.0.1", "::1") if parsed.scheme != "https" and not ( parsed.scheme == "http" and is_localhost ): @@ -791,7 +826,7 @@ def _validate_catalog_url(self, url: str) -> None: f"Catalog URL must use HTTPS (got {parsed.scheme}://). " "HTTP is only allowed for localhost." ) - if not parsed.hostname: + if not hostname: raise StepValidationError( "Catalog URL must be a valid URL with a host." ) @@ -957,15 +992,26 @@ def _fetch_single_catalog( from specify_cli.authentication.http import open_url as _open_url def _validate_url(url: str) -> None: - parsed = urlparse(url) - is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1") + # A malformed authority (e.g. "https://[::1") makes urlparse / + # hostname access raise ValueError; treat it as a refused fetch + # rather than leaking a raw ValueError (this also validates the + # post-redirect resp.geturl(), so a hostile redirect target cannot + # crash the fetch either). + try: + parsed = urlparse(url) + hostname = parsed.hostname + except ValueError: + raise StepCatalogError( + f"Refusing to fetch catalog from malformed URL: {url}" + ) from None + is_localhost = hostname in ("localhost", "127.0.0.1", "::1") if parsed.scheme != "https" and not ( parsed.scheme == "http" and is_localhost ): raise StepCatalogError( f"Refusing to fetch catalog from non-HTTPS URL: {url}" ) - if not parsed.hostname: + if not hostname: raise StepCatalogError( f"Refusing to fetch catalog from URL with no hostname: {url}" ) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 81aec96d3e..e9d4b33a51 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -5072,6 +5072,83 @@ def test_validate_url_localhost_http_allowed(self, project_dir): # Should not raise catalog._validate_catalog_url("http://localhost:8080/catalog.json") + @pytest.mark.parametrize( + "url", + [ + "https://[::1", # unterminated IPv6 bracket + "https://[not-an-ip]/x", # bracketed non-IP host + ], + ) + def test_validate_url_malformed_raises_validation_error(self, project_dir, url): + """A malformed authority must raise WorkflowValidationError, not leak a + raw ValueError. + + ``urlparse``/``.hostname`` raise ValueError on a malformed IPv6 + authority. The command handler only catches WorkflowValidationError, + so a raw ValueError would surface as an uncaught traceback instead of a + clean 'Error:' message + exit 1. Mirrors specify_cli.catalogs (#3435). + """ + from specify_cli.workflows.catalog import ( + WorkflowCatalog, + WorkflowValidationError, + ) + + catalog = WorkflowCatalog(project_dir) + with pytest.raises(WorkflowValidationError, match="malformed"): + catalog._validate_catalog_url(url) + + def test_fetch_malformed_redirect_target_raises_catalog_error( + self, project_dir, monkeypatch + ): + """A malformed post-redirect URL must raise WorkflowCatalogError, not a + raw ValueError. + + The fetch path re-validates ``resp.geturl()`` after following redirects, + so a hostile/broken redirect to a malformed authority + (``https://[::1``) hits ``urlparse``/``.hostname`` and raises + ``ValueError``. Without the guard that ValueError is re-wrapped by the + broad ``except`` as ``Failed to fetch catalog ...: Invalid IPv6 URL``; + the guard turns it into a clean ``... malformed URL ...`` refusal. The + initial ``entry.url`` is valid so validation only trips on the redirect + target. Mirrors specify_cli.catalogs (#3435). + """ + from specify_cli.workflows.catalog import ( + WorkflowCatalog, + WorkflowCatalogEntry, + WorkflowCatalogError, + ) + from specify_cli.authentication import http as auth_http + + class _FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return b"{}" + + def geturl(self): + # A redirect landing on a malformed IPv6 authority. + return "https://[::1" + + monkeypatch.setattr( + auth_http, "open_url", lambda url, timeout=30: _FakeResponse() + ) + + catalog = WorkflowCatalog(project_dir) + entry = WorkflowCatalogEntry( + url="https://example.com/catalog.json", + name="test", + priority=1, + install_allowed=True, + ) + # A fresh project_dir has no cache to fall back to, so the error + # propagates instead of being masked by a stale-cache read. + with pytest.raises(WorkflowCatalogError, match="malformed"): + catalog._fetch_single_catalog(entry, force_refresh=True) + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import WorkflowCatalog @@ -5518,6 +5595,75 @@ def test_validate_url_localhost_http_allowed(self, project_dir): # Should not raise catalog._validate_catalog_url("http://localhost:8080/step-catalog.json") + @pytest.mark.parametrize( + "url", + [ + "https://[::1", # unterminated IPv6 bracket + "https://[not-an-ip]/x", # bracketed non-IP host + ], + ) + def test_validate_url_malformed_raises_validation_error(self, project_dir, url): + """A malformed authority must raise StepValidationError, not leak a raw + ValueError past the command handler (which only catches + StepValidationError). Mirrors specify_cli.catalogs (#3435). + """ + from specify_cli.workflows.catalog import StepCatalog, StepValidationError + + catalog = StepCatalog(project_dir) + with pytest.raises(StepValidationError, match="malformed"): + catalog._validate_catalog_url(url) + + def test_fetch_malformed_redirect_target_raises_catalog_error( + self, project_dir, monkeypatch + ): + """A malformed post-redirect URL must raise StepCatalogError, not a raw + ValueError. + + The fetch path re-validates ``resp.geturl()`` after redirects, so a + broken redirect to a bracketed non-IP host (``https://[not-an-ip]/x``) + makes ``urlparse``/``.hostname`` raise ``ValueError``. Without the guard + that leaks out as ``... Invalid IPv6 URL`` re-wrapping; the guard turns + it into a clean ``... malformed URL ...`` refusal. The initial + ``entry.url`` is valid so validation only trips on the redirect target. + Mirrors specify_cli.catalogs (#3435). + """ + from specify_cli.workflows.catalog import ( + StepCatalog, + StepCatalogEntry, + StepCatalogError, + ) + from specify_cli.authentication import http as auth_http + + class _FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return b"{}" + + def geturl(self): + # A redirect landing on a bracketed non-IP authority. + return "https://[not-an-ip]/x" + + monkeypatch.setattr( + auth_http, "open_url", lambda url, timeout=30: _FakeResponse() + ) + + catalog = StepCatalog(project_dir) + entry = StepCatalogEntry( + url="https://example.com/steps.json", + name="test", + priority=1, + install_allowed=True, + ) + # A fresh project_dir has no cache to fall back to, so the error + # propagates instead of being masked by a stale-cache read. + with pytest.raises(StepCatalogError, match="malformed"): + catalog._fetch_single_catalog(entry, force_refresh=True) + def test_add_catalog(self, project_dir): from specify_cli.workflows.catalog import StepCatalog