feat(url_guard): SSRF and path traversal guard for scraper inputs

- Add url_guard.py with validate_url() that rejects:
  - Non-HTTP schemes (file://, ftp://, javascript:)
  - Private/reserved IP ranges (RFC 1918, loopback, CGNAT, link-local)
  - Internal hostnames (localhost, kubernetes, docker, metadata, etc.)
  - Embedded credentials in URLs
  - Path traversal sequences (.., %2e%2e%2f, %%2e%%2e%%2f)
- Integrate into scraper.scrape() and UltimateScraper.scrape()
- Use typed InvalidRequestError for structured API error responses
- Add 24 tests covering all rejection cases and valid URLs
- 560 tests passing, ruff clean

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 {}
 

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]/")

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)
 

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})