diff --git a/scraper.py b/scraper.py index ff872dc..143bf42 100644 --- a/scraper.py +++ b/scraper.py @@ -19,8 +19,10 @@ import httpx import trafilatura from trafilatura.settings import use_config +from errors import InvalidRequestError from settings import settings from stealth_engine import StealthEngine +from url_guard import validate_url USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", @@ -281,6 +283,10 @@ class PryScraper: async def scrape(self, url: str, options: dict | None = None) -> dict[str, Any]: if not url or not url.startswith(("http://", "https://")): return {"status": "error", "url": url, "error": "Invalid URL", "content": ""} + try: + validate_url(url) + except InvalidRequestError as e: + return {"status": "error", "url": url, "error": str(e), "content": ""} opts = options or {} diff --git a/tests/test_url_guard.py b/tests/test_url_guard.py new file mode 100644 index 0000000..8b5ce1a --- /dev/null +++ b/tests/test_url_guard.py @@ -0,0 +1,107 @@ +"""Tests for SSRF and path-traversal guard.""" + +from __future__ import annotations + +import pytest + +from errors import InvalidRequestError +from url_guard import validate_url + + +class TestValidateURL: + def test_accepts_normal_https(self): + result = validate_url("https://example.com/page") + assert result == "https://example.com/page" + + def test_accepts_normal_http(self): + result = validate_url("http://example.com/page") + assert result == "http://example.com/page" + + def test_accepts_https_with_path_and_query(self): + url = "https://api.example.com/v1/data?key=val&page=2#section" + result = validate_url(url) + assert result == url + + def test_rejects_empty_url(self): + with pytest.raises(InvalidRequestError): + validate_url("") + + def test_rejects_ftp_scheme(self): + with pytest.raises(InvalidRequestError): + validate_url("ftp://example.com/file") + + def test_rejects_file_scheme(self): + with pytest.raises(InvalidRequestError): + validate_url("file:///etc/passwd") + + def test_rejects_javascript_scheme(self): + with pytest.raises(InvalidRequestError): + validate_url("javascript:alert(1)") + + def test_rejects_localhost_hostname(self): + with pytest.raises(InvalidRequestError): + validate_url("http://localhost:8080/admin") + + def test_rejects_localhost_ip(self): + with pytest.raises(InvalidRequestError): + validate_url("http://127.0.0.1:8080/admin") + + def test_rejects_private_ip_10_dot(self): + with pytest.raises(InvalidRequestError): + validate_url("http://10.0.0.1/admin") + + def test_rejects_private_ip_192_168(self): + with pytest.raises(InvalidRequestError): + validate_url("http://192.168.1.1/admin") + + def test_rejects_private_ip_172(self): + with pytest.raises(InvalidRequestError): + validate_url("http://172.16.0.1/admin") + + def test_rejects_internal_hostname(self): + with pytest.raises(InvalidRequestError): + validate_url("http://internal.corp/admin") + + def test_rejects_kubernetes_metadata(self): + with pytest.raises(InvalidRequestError): + validate_url("http://kubernetes.default.svc.cluster.local") + + def test_rejects_embedded_credentials(self): + with pytest.raises(InvalidRequestError): + validate_url("http://user:pass@example.com/") + + def test_rejects_path_traversal_simple(self): + with pytest.raises(InvalidRequestError): + validate_url("https://example.com/../../etc/passwd") + + def test_rejects_path_traversal_encoded(self): + with pytest.raises(InvalidRequestError): + validate_url("https://example.com/%2e%2e%2fetc/passwd") + + def test_rejects_path_traversal_double_encoded(self): + with pytest.raises(InvalidRequestError): + validate_url("https://example.com/%252e%252e%252fetc") + + def test_rejects_tailscale_ip(self): + with pytest.raises(InvalidRequestError): + validate_url("http://100.64.0.1/") + + def test_rejects_empty_hostname(self): + with pytest.raises(InvalidRequestError): + validate_url("http:///path") + + def test_rejects_docker_host(self): + with pytest.raises(InvalidRequestError): + validate_url("http://docker/container") + + def test_rejects_metadata_google(self): + with pytest.raises(InvalidRequestError): + validate_url("http://metadata.google.internal/") + + def test_rejects_link_local_ipv6(self): + with pytest.raises(InvalidRequestError): + validate_url("http://[fe80::1]/") + + def test_rejects_loopback_ipv6(self): + with pytest.raises(InvalidRequestError): + validate_url("http://[::1]/") diff --git a/ultimate_scraper.py b/ultimate_scraper.py index c59a209..086924f 100644 --- a/ultimate_scraper.py +++ b/ultimate_scraper.py @@ -18,7 +18,9 @@ import time from typing import Any, ClassVar from urllib.parse import urlparse +from errors import InvalidRequestError from resilience import registry +from url_guard import validate_url logger = logging.getLogger(__name__) @@ -111,6 +113,11 @@ class UltimateScraper: Will try up to 10 strategies until one succeeds.""" opts = options or {} errors: list[str] = [] + try: + validate_url(url) + except InvalidRequestError as e: + return {"status": "error", "url": url, "error": str(e)} + proxy_url = self._resolve_proxy_url(opts) diff --git a/url_guard.py b/url_guard.py new file mode 100644 index 0000000..3b12588 --- /dev/null +++ b/url_guard.py @@ -0,0 +1,155 @@ +"""SSRF and path traversal guard for Pry scrapers. + +Validates URLs before scraping to prevent: +- SSRF attacks (private IPs, loopback, internal hosts) +- Path traversal (../, encoded variants) +- File:// or other non-HTTP schemes +- DNS rebinding risks (common internal hostnames) +""" + +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Rug Munch Media LLC +# +# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper +# Licensed under MIT. See LICENSE. + +from __future__ import annotations + +import ipaddress +import logging +import re +import socket +from urllib.parse import urlparse + +logger = logging.getLogger(__name__) + +# Private/reserved IP ranges (RFC 1918, RFC 6598, RFC 4193, link-local, loopback) +_PRIVATE_NETS: list[str] = [ + "127.0.0.0/8", # loopback + "10.0.0.0/8", # RFC 1918 + "172.16.0.0/12", # RFC 1918 + "192.168.0.0/16", # RFC 1918 + "100.64.0.0/10", # RFC 6598 (CGNAT) + "169.254.0.0/16", # link-local + "fc00::/7", # RFC 4193 (ULA) + "fe80::/10", # link-local IPv6 + "::1/128", # IPv6 loopback +] + +# Common internal hostnames that suggest SSRF targeting +_INTERNAL_HOST_RE = re.compile( + r"^(.+\.)?" + r"(localhost|internal|intranet|private|corp|home|router|gateway|" + r"metadata|kubernetes|kube|docker|consul|nomad|vault|" + r"elasticsearch|redis|mysql|postgres|memcached|" + r"grafana|prometheus|alertmanager|" + r"minio|s3|storage|db|database|" + r"admin|dashboard|management|" + r"apiserver|etcd|" + r"tailscale|100\.\d+\.\d+\.\d+)" + r"(\..+)?$", + re.IGNORECASE, +) + +# Blocked URL schemes +_ALLOWED_SCHEMES = {"http", "https"} + +# Path traversal patterns +_TRAVERSAL_RE = re.compile( + r"(?:^|/)\.\.(?:/|%2[fF]|\\|$)" + r"|%2[eE]%2[eE]" + r"|%252[eE]%252[eE]%252[fF]" + r"|\.\.(?:/|\\|$)" +) + + +def validate_url(url: str, *, resolve_dns: bool = False) -> str: + """Validate a URL for SSRF and path-traversal safety. + + Args: + url: The URL to validate. + resolve_dns: If True, perform a DNS lookup to check that the resolved + IP is not in a private range. This adds latency and cannot + be async here, so it defaults to False. + + Returns: + The validated URL (unchanged). + + Raises: + InvalidRequestError: If the URL is unsafe. + """ + if not url: + raise _error("URL is empty", url) + + parsed = urlparse(url) + + # Reject non-HTTP schemes + if parsed.scheme not in _ALLOWED_SCHEMES: + raise _error( + f"URL scheme {parsed.scheme} is not allowed (only http/https)", + url, + ) + + # Reject URLs with embedded credentials + if parsed.username or parsed.password: + raise _error("URL must not contain embedded credentials (user:password@)", url) + + hostname = parsed.hostname or "" + + # Reject empty hostnames + if not hostname: + raise _error("URL must have a valid hostname", url) + + # Reject raw IP: check if it is a private/reserved address + try: + ip = ipaddress.ip_address(hostname) + for net in _PRIVATE_NETS: + if ip in ipaddress.ip_network(net): + raise _error( + f"URL resolves to a private/reserved IP range ({net})", + url, + ) + except ValueError: + # Not an IP address — continue with hostname checks + pass + + # Reject internal/sensitive hostnames + if _INTERNAL_HOST_RE.match(hostname): + raise _error( + f"URL hostname {hostname} appears to be an internal resource", + url, + ) + + # Reject DNS resolution to private IPs (opt-in) + if resolve_dns: + try: + addrs = socket.getaddrinfo(hostname, None) + for _family, _, _, _, sockaddr in addrs: + addr = sockaddr[0] + try: + ip = ipaddress.ip_address(addr) + for net in _PRIVATE_NETS: + if ip in ipaddress.ip_network(net): + raise _error( + f"URL resolves to private IP {addr} ({net})", + url, + ) + except ValueError: + continue + except OSError as e: + logger.debug("dns_lookup_failed", extra={"hostname": hostname, "error": str(e)}) + + # Check path for traversal attempts + path = parsed.path + parsed.params + parsed.query + parsed.fragment + if _TRAVERSAL_RE.search(path): + raise _error("URL contains path traversal sequences", url) + + return url + + +def _error(message: str, url: str) -> Exception: + """Build a structured InvalidRequestError.""" + from errors import InvalidRequestError + + logger.warning("url_guard_rejected", extra={"url": url, "reason": message}) + return InvalidRequestError(message, details={"url": url, "reason": message})