A03: Manual redirect control with per-hop validation
Replace urlopen automatic redirect following with custom HTTPRedirectHandler that raises on 3xx redirects. Each redirect hop is validated (scheme, host, port) before the request is made using _validate_url_before_io(). Key changes: - _StrictRedirectHandler intercepts 301/302/303/307/308 responses - _extract_redirect_url() extracts Location header from redirect responses - fetch_html() manually follows redirects with hop count limit (MAX_REDIRECT_HOPS=5) - Relative redirect URLs resolved with urljoin() before validation - All redirect targets validated against ALLOWED_HOSTS, ALLOWED_PORTS, HTTPS-only Tests: - test_fetch_html_redirect_to_disallowed_host_rejected (mocked redirect) - test_fetch_html_redirect_chain_limit (exceeds MAX_REDIRECT_HOPS) - test_extract_redirect_url_from_headers (Location/location headers) - test_urljoin_resolves_relative_redirects (relative URL resolution)
This commit is contained in:
@@ -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()}")
|
||||
|
||||
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:
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user