diff --git a/rf4_research/community_cli.py b/rf4_research/community_cli.py index aebae21..ac99d50 100644 --- a/rf4_research/community_cli.py +++ b/rf4_research/community_cli.py @@ -102,30 +102,65 @@ def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> No _write_state(state_file, state) -def _validate_url_host(url: str) -> str: - """Validate URL hostname is in allowlist before making network call.""" - hostname = (urlsplit(url).hostname or "").lower() +MAX_REDIRECT_HOPS = 5 +ALLOWED_PORTS = frozenset({443, 80}) + + +def _validate_url_before_io(url: str) -> tuple[str, str]: + """A03: Validate scheme, host, port before any network I/O. + + Returns (normalized_hostname, scheme). Raises ValueError for disallowed + schemes, ports, or hosts. + """ + parsed = urlsplit(url) + # Validate scheme + if parsed.scheme not in ("http", "https"): + raise ValueError(f"URL scheme {parsed.scheme!r} not allowed") + if parsed.scheme != "https": + raise ValueError("Only HTTPS URLs are allowed") + # Validate port + if parsed.port is not None and parsed.port not in ALLOWED_PORTS: + raise ValueError(f"Port {parsed.port} not in allowed ports {ALLOWED_PORTS}") + # Validate hostname + hostname = (parsed.hostname or "").lower() if hostname.startswith("www."): hostname = hostname[4:] if not hostname: raise ValueError("URL must include a valid hostname") if hostname not in ALLOWED_HOSTS: raise ValueError(f"URL hostname {hostname} not in allowlist") + return hostname, parsed.scheme + + +def _validate_url_host(url: str) -> str: + """Legacy alias for _validate_url_before_io (returns hostname only).""" + hostname, _ = _validate_url_before_io(url) return hostname -def fetch_html(url: str, *, timeout: float = 30) -> str: - # Validate host BEFORE network I/O to prevent SSRF to internal endpoints - _validate_url_host(url) +def fetch_html(url: str, *, timeout: float = 30, _redirects: int = 0) -> str: + """A03: Fetch with pre-I/O validation and redirect hop limit.""" + if _redirects > MAX_REDIRECT_HOPS: + raise ValueError(f"Redirect chain exceeds {MAX_REDIRECT_HOPS} hops") + # Validate BEFORE any network I/O + hostname, scheme = _validate_url_before_io(url) request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"}) with urlopen(request, timeout=timeout) as response: final_url = response.url - # Also validate redirect target - final_hostname = (urlsplit(final_url).hostname or "").lower() + # Validate redirect target + final_parsed = urlsplit(final_url) + final_hostname = (final_parsed.hostname or "").lower() if final_hostname.startswith("www."): final_hostname = final_hostname[4:] if final_hostname not in ALLOWED_HOSTS: raise ValueError(f"Redirect hostname {final_hostname} not in allowlist") + if final_parsed.scheme != "https": + raise ValueError("Redirect to non-HTTPS not allowed") + if final_parsed.port is not None and final_parsed.port not in ALLOWED_PORTS: + raise ValueError(f"Redirect port {final_parsed.port} not allowed") + # Recursively follow if redirect (urlopen follows automatically, but we check) + if final_url != url: + return fetch_html(final_url, timeout=timeout, _redirects=_redirects + 1) if response.headers.get_content_type() != "text/html": raise ValueError(f"expected text/html, got {response.headers.get_content_type()}") data = response.read(MAX_RESPONSE_BYTES + 1) diff --git a/tests/test_community_cli.py b/tests/test_community_cli.py index 07bf935..400e95b 100644 --- a/tests/test_community_cli.py +++ b/tests/test_community_cli.py @@ -43,20 +43,23 @@ def test_validate_url_host_rejects_disallowed_hosts() -> None: # Disallowed hosts raise ValueError before network I/O with pytest.raises(ValueError, match="not in allowlist"): - _validate_url_host("http://localhost:8080/admin") + _validate_url_host("https://169.254.169.254/latest/meta-data/") with pytest.raises(ValueError, match="not in allowlist"): - _validate_url_host("http://169.254.169.254/latest/meta-data/") - with pytest.raises(ValueError, match="not in allowlist"): - _validate_url_host("http://internal-service.corp/api") + _validate_url_host("https://internal-service.corp/api") + # Port validation also works + with pytest.raises(ValueError, match="not in allowed ports"): + _validate_url_host("https://download.rf4db.com:9999/admin") def test_validate_url_host_rejects_missing_hostname() -> None: from rf4_research.community_cli import _validate_url_host with pytest.raises(ValueError, match="valid hostname"): - _validate_url_host("not-a-valid-url") - with pytest.raises(ValueError, match="valid hostname"): - _validate_url_host("") + _validate_url_host("https://") + with pytest.raises(ValueError, match="scheme.*not allowed"): + _validate_url_host("ftp://rf4db.com/file") + with pytest.raises(ValueError, match="HTTPS"): + _validate_url_host("http://rf4db.com/file") def test_write_state_is_atomic_with_flush(tmp_path: Path) -> None: