R11: Validate URL host before network I/O — prevent SSRF

- Add _validate_url_host() to check allowlist before urlopen()
- Validate both original URL and redirect target
- Reject localhost, internal IPs, and non-allowlisted hosts
- Add 2 unit tests for disallowed host rejection
- Prevents SSRF attacks via malicious source URLs
This commit is contained in:
ik
2026-09-10 05:45:57 +07:00
parent 6e0077bd4d
commit 39f66481be
2 changed files with 46 additions and 3 deletions
+20 -3
View File
@@ -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)