diff --git a/rf4_research/community_cli.py b/rf4_research/community_cli.py index e8e7b4e..71fc71f 100644 --- a/rf4_research/community_cli.py +++ b/rf4_research/community_cli.py @@ -7,10 +7,11 @@ import os import sys import time import tempfile +import urllib.error from dataclasses import asdict from pathlib import Path -from urllib.parse import urlsplit -from urllib.request import Request, urlopen +from urllib.parse import urlsplit, urljoin +from urllib.request import Request, urlopen, HTTPRedirectHandler, build_opener from .community_sources import ( parse_rf4db_catches, @@ -148,6 +149,43 @@ MAX_REDIRECT_HOPS = 5 ALLOWED_PORTS = frozenset({443, 80}) +class _StrictRedirectHandler(HTTPRedirectHandler): + """A03: Raise on redirect instead of following automatically. + + Returns the HTTPError (3xx) so the caller can validate each hop + before deciding whether to follow. + """ + + def http_error_302(self, req, fp, code, msg, headers): + """A03: Intercept 302 — don't follow automatically.""" + return None + + def http_error_301(self, req, fp, code, msg, headers): + """A03: Intercept 301 — don't follow automatically.""" + return None + + def http_error_303(self, req, fp, code, msg, headers): + """A03: Intercept 303 — don't follow automatically.""" + return None + + def http_error_307(self, req, fp, code, msg, headers): + """A03: Intercept 307 — don't follow automatically.""" + return None + + def http_error_308(self, req, fp, code, msg, headers): + """A03: Intercept 308 — don't follow automatically.""" + return None + + +def _extract_redirect_url(headers) -> str | None: + """A03: Extract Location header from redirect response.""" + # Try 'Location' first (RFC standard), then 'location' (lowercase) + for key in ("Location", "location"): + if key in headers: + return headers[key] + return None + + def _validate_url_before_io(url: str) -> tuple[str, str]: """A03: Validate scheme, host, port before any network I/O. @@ -181,34 +219,43 @@ def _validate_url_host(url: str) -> str: def fetch_html(url: str, *, timeout: float = 30, _redirects: int = 0) -> str: - """A03: Fetch with pre-I/O validation and redirect hop limit.""" + """A03: Manual redirect control with per-hop validation. + + Each redirect hop is validated (scheme, host, port) before the request + is made. urlopen's automatic redirect following is bypassed. + """ 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) + # Build opener that does NOT follow redirects automatically + opener = build_opener(_StrictRedirectHandler()) request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"}) - with urlopen(request, timeout=timeout) as response: - final_url = response.url - # 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) - if len(data) > MAX_RESPONSE_BYTES: - raise ValueError("response exceeded 5MB limit") - return data.decode(response.headers.get_content_charset() or "utf-8") + + try: + with opener.open(request, timeout=timeout) as response: + # If we get here, no redirect occurred (or final destination reached) + data = response.read(MAX_RESPONSE_BYTES + 1) + if len(data) > MAX_RESPONSE_BYTES: + raise ValueError("response exceeded 5MB limit") + content_type = response.headers.get_content_type() or "" + if "text/html" not in content_type: + raise ValueError(f"expected text/html, got {content_type}") + return data.decode(response.headers.get_content_charset() or "utf-8") + except urllib.error.HTTPError as exc: + # Check if this is a redirect (3xx status) + if exc.code in (301, 302, 303, 307, 308): + redirect_url = _extract_redirect_url(exc.headers) + if redirect_url is None: + raise ValueError(f"Redirect {exc.code} with no Location header") from exc + # Resolve relative URLs + redirect_url = urljoin(url, redirect_url) + # Validate this hop BEFORE following + _validate_url_before_io(redirect_url) + # Recursively follow with hop count + return fetch_html(redirect_url, timeout=timeout, _redirects=_redirects + 1) + else: + raise def main(argv: list[str] | None = None) -> int: diff --git a/tests/test_community_cli.py b/tests/test_community_cli.py index 65ba805..8603aab 100644 --- a/tests/test_community_cli.py +++ b/tests/test_community_cli.py @@ -141,3 +141,135 @@ def test_check_and_reserve_atomic_under_concurrent_access(tmp_path: Path) -> Non assert ok_count == 1, f"Expected exactly 1 ok, got {ok_count}: {results}" denied_count = sum(1 for _, r in results if "cooldown" in r) assert denied_count == 2, f"Expected 2 denied, got {denied_count}: {results}" + + +# A03: Manual redirect control tests + +def test_validate_url_before_io_rejects_http(tmp_path: Path) -> None: + """A03: HTTP scheme rejected even for allowed hosts.""" + from rf4_research.community_cli import _validate_url_before_io + + with pytest.raises(ValueError, match="Only HTTPS"): + _validate_url_before_io("http://rf4-stat.ru/fishing/") + + +def test_validate_url_before_io_rejects_bad_ports(tmp_path: Path) -> None: + """A03: Non-standard ports rejected.""" + from rf4_research.community_cli import _validate_url_before_io + + with pytest.raises(ValueError, match="not in allowed ports"): + _validate_url_before_io("https://rf4-stat.ru:8080/path") + with pytest.raises(ValueError, match="not in allowed ports"): + _validate_url_before_io("https://rf4-stat.ru:4443/path") + + +def test_validate_url_before_io_rejects_disallowed_hosts(tmp_path: Path) -> None: + """A03: Disallowed hosts rejected before network I/O.""" + from rf4_research.community_cli import _validate_url_before_io + + with pytest.raises(ValueError, match="not in allowlist"): + _validate_url_before_io("https://evil.com/phishing") + with pytest.raises(ValueError, match="not in allowlist"): + _validate_url_before_io("https://169.254.169.254/metadata") + + +def test_validate_url_before_io_allows_valid_urls() -> None: + """A03: Valid allowed hosts pass validation.""" + from rf4_research.community_cli import _validate_url_before_io + + hostname, scheme = _validate_url_before_io("https://rf4-stat.ru/fishing/") + assert hostname == "rf4-stat.ru" + assert scheme == "https" + + hostname, scheme = _validate_url_before_io("https://download.rf4db.com/ru/catches") + assert hostname == "download.rf4db.com" + assert scheme == "https" + + +def test_validate_url_before_io_normalizes_www() -> None: + """A03: www. prefix is stripped from hostname.""" + from rf4_research.community_cli import _validate_url_before_io + + hostname, _ = _validate_url_before_io("https://www.rf4-stat.ru/fishing/") + assert hostname == "rf4-stat.ru" + + +def test_fetch_html_redirect_to_disallowed_host_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + """A03: Redirect to disallowed host raises before making request.""" + from rf4_research.community_cli import fetch_html + import urllib.error + + # Mock build_opener to return an opener that raises 302 redirect + from unittest.mock import Mock + from rf4_research.community_cli import _StrictRedirectHandler + + opener_instance = Mock() + + def mock_open(request, timeout=None): + # Simulate a redirect response + exc = urllib.error.HTTPError( + url=str(request.full_url), + code=302, + msg="Found", + hdrs=Mock(), + fp=None + ) + exc.headers = {"Location": "https://evil.com/phishing"} + raise exc + + opener_instance.open = mock_open + + def mock_build_opener(*args, **kwargs): + return opener_instance + + monkeypatch.setattr("rf4_research.community_cli.build_opener", mock_build_opener) + + # Should raise during redirect validation (evil.com not in allowlist) + with pytest.raises(ValueError, match="not in allowlist"): + fetch_html("https://rf4-stat.ru/redirect-to-evil", _redirects=0) + + +def test_fetch_html_redirect_chain_limit() -> None: + """A03: Redirect chain exceeding MAX_REDIRECT_HOPS raises ValueError.""" + from rf4_research.community_cli import fetch_html, MAX_REDIRECT_HOPS + + with pytest.raises(ValueError, match=f"Redirect chain exceeds {MAX_REDIRECT_HOPS} hops"): + fetch_html("https://rf4-stat.ru/", _redirects=MAX_REDIRECT_HOPS + 1) + + +def test_fetch_html_allows_valid_https() -> None: + """A03: Valid HTTPS URLs pass scheme validation.""" + from rf4_research.community_cli import _validate_url_before_io + + hostname, scheme = _validate_url_before_io("https://rf4-stat.ru/fishing/") + assert scheme == "https" + assert hostname == "rf4-stat.ru" + + +def test_extract_redirect_url_from_headers() -> None: + """A03: _extract_redirect_url handles both Location and location headers.""" + from rf4_research.community_cli import _extract_redirect_url + + # Standard capitalization + headers = {"Location": "https://rf4-stat.ru/new-path"} + assert _extract_redirect_url(headers) == "https://rf4-stat.ru/new-path" + + # Lowercase (some servers use this) + headers = {"location": "https://rf4-stat.ru/other-path"} + assert _extract_redirect_url(headers) == "https://rf4-stat.ru/other-path" + + # Missing header + headers = {} + assert _extract_redirect_url(headers) is None + + +def test_urljoin_resolves_relative_redirects() -> None: + """A03: Relative redirect URLs are resolved against the base URL.""" + from urllib.parse import urljoin + + # Relative path + assert urljoin("https://rf4-stat.ru/old", "/new") == "https://rf4-stat.ru/new" + # Relative without leading slash + assert urljoin("https://rf4-stat.ru/old/path", "new") == "https://rf4-stat.ru/old/new" + # Absolute URL + assert urljoin("https://rf4-stat.ru/old", "https://rf4-stat.ru/absolute") == "https://rf4-stat.ru/absolute"