diff --git a/rf4_research/community_cli.py b/rf4_research/community_cli.py index 697866d..1a67de9 100644 --- a/rf4_research/community_cli.py +++ b/rf4_research/community_cli.py @@ -93,13 +93,30 @@ 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() + 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 + + 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) request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"}) with urlopen(request, timeout=timeout) as response: final_url = response.url - hostname = (urlsplit(final_url).hostname or "").lower() - if hostname not in ALLOWED_HOSTS: - raise ValueError(f"URL hostname {hostname} not in allowlist") + # Also validate redirect target + final_hostname = (urlsplit(final_url).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 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 e18d406..47b5198 100644 --- a/tests/test_community_cli.py +++ b/tests/test_community_cli.py @@ -31,3 +31,29 @@ def test_failed_fetch_still_reserves_site_cooldown(tmp_path: Path, monkeypatch: monkeypatch.setattr(community_cli, "fetch_html", fail) assert community_cli.main(["rf4db", "--state-file", str(state_file)]) == 1 assert "download.rf4db.com" in json.loads(state_file.read_text(encoding="utf-8")) + + +def test_validate_url_host_rejects_disallowed_hosts() -> None: + from rf4_research.community_cli import _validate_url_host + + # Allowed hosts pass + assert _validate_url_host("https://download.rf4db.com/ru/catches") == "download.rf4db.com" + assert _validate_url_host("https://www.rf4-stat.ru/posts/") == "rf4-stat.ru" + assert _validate_url_host("https://rf4map.ru/point/123") == "rf4map.ru" + + # Disallowed hosts raise ValueError before network I/O + with pytest.raises(ValueError, match="not in allowlist"): + _validate_url_host("http://localhost:8080/admin") + 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") + + +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("")