pryscraper/url_guard.py

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