from __future__ import annotations import hashlib import ipaddress from collections.abc import Mapping from dataclasses import dataclass from datetime import datetime from datetime import timezone as utc from email.utils import parsedate_to_datetime from urllib.parse import urljoin import httpx from django.conf import settings from apps.sources.models import Source from .policy import assess_url from .tls import trusted_tls_context from .url_security import ValidatedUrl, validate_public_url ALLOWED_CONTENT_TYPES = ( "text/html", "application/xhtml+xml", "application/xml", "text/xml", "application/json", "application/ld+json", "text/plain", "application/rss+xml", "application/atom+xml", ) class FetchError(RuntimeError): pass class PolicyBlockedError(FetchError): pass class ContentRejectedError(FetchError): pass class RateLimitedError(FetchError): def __init__(self, retry_after_seconds: int | None, message: str = "HTTP 429") -> None: self.retry_after_seconds = retry_after_seconds suffix = f"; Retry-After={retry_after_seconds}s" if retry_after_seconds else "" super().__init__(f"{message}{suffix}".strip()) class FetchTimeoutError(FetchError): pass def _connected_peer_ip( response: httpx.Response, ) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None: """Return the actual remote peer when httpcore exposes its network stream.""" stream = response.extensions.get("network_stream") if stream is None or not hasattr(stream, "get_extra_info"): return None try: server_address = stream.get_extra_info("server_addr") raw_address = server_address[0] if server_address else None return ipaddress.ip_address(raw_address) if raw_address else None except (IndexError, TypeError, ValueError, OSError): return None def _validate_connected_peer( response: httpx.Response, *, before: ValidatedUrl, after: ValidatedUrl, require_peer: bool, ) -> None: peer_ip = _connected_peer_ip(response) if peer_ip is not None: if not peer_ip.is_global: raise FetchError(f"Niet-publiek verbonden IP-adres geblokkeerd: {peer_ip}") return if require_peer: raise FetchError( "Verbonden IP-adres kon niet veilig worden vastgesteld; aanvraag geblokkeerd." ) # Mock/custom transports do not always expose the peer socket. Keep the conservative # DNS-overlap fallback only for explicitly injected test/custom clients. if not set(before.addresses).intersection(after.addresses): raise FetchError("DNS-rebindcontrole faalde bij het benaderen van bron.") def parse_retry_after(value: str | None) -> int | None: if not value: return None value = value.strip() if not value: return None try: return max(0, int(value)) except ValueError: pass try: retry_datetime = parsedate_to_datetime(value) except (TypeError, ValueError): return None if retry_datetime is None: return None if retry_datetime.tzinfo is None: retry_datetime = retry_datetime.replace(tzinfo=utc) retry_aware = retry_datetime.astimezone(utc) now = datetime.now(utc) delta = (retry_aware - now).total_seconds() return max(0, int(delta)) @dataclass(slots=True) class FetchedDocument: requested_url: str final_url: str status_code: int headers: dict[str, str] content: bytes @property def text(self) -> str: encoding = "utf-8" content_type = self.headers.get("content-type", "") if "charset=" in content_type: encoding = content_type.split("charset=", 1)[1].split(";", 1)[0].strip() return self.content.decode(encoding, errors="replace") @property def sha256(self) -> str: return hashlib.sha256(self.content).hexdigest() def fetch_url( url: str, *, source: Source | None = None, conditional_headers: Mapping[str, str] | None = None, client: httpx.Client | None = None, ) -> FetchedDocument: decision = assess_url(url, source=source) if not decision.allowed: raise PolicyBlockedError(decision.reason) headers = { "User-Agent": settings.FETCHER_USER_AGENT, "Accept": ( "text/html,application/xhtml+xml,application/xml,application/json," "text/plain;q=0.8,*/*;q=0.1" ), } if conditional_headers: headers.update(conditional_headers) own_client = client is None http_client = client or httpx.Client( timeout=httpx.Timeout(settings.FETCHER_TIMEOUT_SECONDS), follow_redirects=False, headers=headers, verify=trusted_tls_context(), ) current_url = url try: for _ in range(settings.FETCHER_MAX_REDIRECTS + 1): validation = validate_public_url( current_url, allow_nonstandard_ports=settings.FETCHER_ALLOW_NONSTANDARD_PORTS, ) response = http_client.get(current_url, headers=headers) post_validation = validate_public_url( str(response.url) if response.url else current_url, allow_nonstandard_ports=settings.FETCHER_ALLOW_NONSTANDARD_PORTS, ) _validate_connected_peer( response, before=validation, after=post_validation, require_peer=own_client, ) if response.status_code in {301, 302, 303, 307, 308}: location = response.headers.get("location") if not location: raise FetchError("Redirect zonder Location-header.") current_url = urljoin(current_url, location) next_decision = assess_url(current_url, source=source) if not next_decision.allowed: raise PolicyBlockedError(next_decision.reason) continue if response.status_code == 304: return FetchedDocument(url, current_url, 304, dict(response.headers), b"") if response.status_code == 429: retry_after = parse_retry_after(response.headers.get("retry-after")) raise RateLimitedError(retry_after) if response.status_code >= 400: raise FetchError(f"HTTP {response.status_code}") content_type = response.headers.get("content-type", "").lower() if content_type and not any( allowed in content_type for allowed in ALLOWED_CONTENT_TYPES ): raise ContentRejectedError(f"Content-Type niet toegestaan: {content_type}") content = response.content if len(content) > settings.FETCHER_MAX_BYTES: raise ContentRejectedError("Document overschrijdt de ingestelde groottebeperking.") return FetchedDocument( url, current_url, response.status_code, dict(response.headers), content ) raise FetchError("Te veel redirects.") except httpx.TimeoutException as exc: raise FetchTimeoutError("Timeout tijdens HTTP-opvraag") from exc except httpx.RequestError as exc: raise FetchError(f"Netwerkfout tijdens HTTP-opvraag: {exc}") from exc finally: if own_client: http_client.close()