Skip to content
Merged
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
70 changes: 58 additions & 12 deletions src/specify_cli/workflows/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,16 +165,28 @@ 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
):
raise WorkflowValidationError(
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."
)
Expand Down Expand Up @@ -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(
Comment thread
mnriem marked this conversation as resolved.
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}"
)
Expand Down Expand Up @@ -782,16 +805,28 @@ 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
):
raise StepValidationError(
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."
)
Expand Down Expand Up @@ -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(
Comment thread
mnriem marked this conversation as resolved.
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}"
)
Expand Down
146 changes: 146 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down
Loading