"""
URL Validator for SSRF Prevention

Validates URLs to prevent Server-Side Request Forgery (SSRF) attacks
by blocking requests to internal/private networks and enforcing safe schemes.
"""

import ipaddress
import re
import socket
from urllib.parse import urlparse
from typing import Optional
from loguru import logger
from urllib3.exceptions import LocationParseError
from urllib3.util import parse_url

from .ip_ranges import PRIVATE_IP_RANGES as BLOCKED_IP_RANGES
from .ip_ranges import NAT64_PREFIXES

# Cloud-provider metadata endpoints — always blocked, even with
# allow_localhost=True or allow_private_ips=True. These IPs expose IAM /
# instance-role credentials and are never legitimate destinations.
# nosec B104 - Hardcoded IPs are intentional for SSRF prevention
ALWAYS_BLOCKED_METADATA_IPS = frozenset(
    {
        "169.254.169.254",  # AWS IMDSv1/v2, Azure, OCI, DigitalOcean
        "169.254.170.2",  # AWS ECS task metadata v3
        "169.254.170.23",  # AWS ECS task metadata v4
        "169.254.0.23",  # Tencent Cloud
        "100.100.100.200",  # AlibabaCloud
    }
)

# Allowed URL schemes
ALLOWED_SCHEMES = {"http", "https"}


def is_nat64_wrapped_metadata_ip(ip: ipaddress._BaseAddress) -> bool:
    """True iff ``ip`` is an IPv6 address inside a NAT64 prefix whose
    embedded IPv4 (low 32 bits) is in ``ALWAYS_BLOCKED_METADATA_IPS``.

    Both ``is_ip_blocked`` and ``NotificationURLValidator._ip_matches_blocked_range``
    consult this before honoring the ``security.allow_nat64`` operator
    opt-in, so cloud-metadata access cannot be re-opened through an
    IPv6-wrapped destination on a NAT64-equipped host. Keeping the
    extraction in one place prevents the two validators from drifting.
    """
    if not isinstance(ip, ipaddress.IPv6Address):
        return False
    for nat64_prefix in NAT64_PREFIXES:
        if ip in nat64_prefix:
            embedded_v4 = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
            return str(embedded_v4) in ALWAYS_BLOCKED_METADATA_IPS
    return False


# RFC 3986 forbids these characters in URLs; their presence in a URL signals
# a parser-differential attempt (GHSA-g23j-2vwm-5c25). \s covers space, \t,
# \n, \r, \v, \f. Backslash is the load-bearing payload — Python's urlparse
# treats it as a literal char while requests/urllib3 treat it as a path
# delimiter, so a crafted URL like ``http://127.0.0.1\@1.1.1.1`` would
# pass the urlparse-based hostname check but actually connect to 127.0.0.1.
RFC_FORBIDDEN_URL_CHARS_RE = re.compile(r"[\\\s\x00-\x1f\x7f]")


def is_ip_blocked(
    ip_str: str,
    allow_localhost: bool = False,
    allow_private_ips: bool = False,
    allow_nat64: Optional[bool] = None,
) -> bool:
    """
    Check if an IP address is in a blocked range.

    Args:
        ip_str: IP address as string
        allow_localhost: Whether to allow localhost/loopback addresses
        allow_private_ips: Whether to allow all private/internal IPs plus localhost.
            This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
            used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
            private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
            like SearXNG or Ollama in containerized environments.
            Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
            (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
            are ALWAYS blocked regardless of these flags.
        allow_nat64: Override for the ``security.allow_nat64`` carve-out.
            ``None`` (default) reads the env setting — the behavior every
            existing caller relies on. An explicit ``bool`` answers a
            hypothetical ("would enabling NAT64 unblock this?") without
            mutating env; used by the notification "Test" admin hint to
            decide whether to surface ``LDR_SECURITY_ALLOW_NAT64``. The
            cloud-metadata always-block above fires first either way, so
            this can never reopen IMDS.

    Returns:
        True if IP is blocked, False otherwise
    """
    # Loopback ranges that can be allowed for trusted internal services
    # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
    LOOPBACK_RANGES = [
        ipaddress.ip_network("127.0.0.0/8"),  # IPv4 loopback
        ipaddress.ip_network("::1/128"),  # IPv6 loopback
    ]

    # Private/internal network ranges - allowed with allow_private_ips=True
    # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
    PRIVATE_RANGES = [
        # RFC1918 Private Ranges
        ipaddress.ip_network("10.0.0.0/8"),  # Class A private
        ipaddress.ip_network("172.16.0.0/12"),  # Class B private
        ipaddress.ip_network("192.168.0.0/16"),  # Class C private
        # Container/Virtual Network Ranges
        ipaddress.ip_network(
            "100.64.0.0/10"
        ),  # CGNAT - used by Podman/rootless containers
        ipaddress.ip_network(
            "169.254.0.0/16"
        ),  # Link-local (cloud metadata IPs blocked separately via ALWAYS_BLOCKED_METADATA_IPS)
        # IPv6 Private Ranges
        ipaddress.ip_network("fc00::/7"),  # IPv6 Unique Local Addresses
        ipaddress.ip_network("fe80::/10"),  # IPv6 Link-Local
    ]

    try:
        ip = ipaddress.ip_address(ip_str)

        # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1 → 127.0.0.1)
        # These bypass IPv4 range checks if not converted.
        if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
            ip = ip.ipv4_mapped

        # ALWAYS block cloud-metadata endpoints - critical SSRF target
        # for credential theft (AWS IMDS/ECS, Azure, OCI, DigitalOcean,
        # AlibabaCloud, Tencent Cloud). These are never legitimate
        # destinations regardless of allow_localhost / allow_private_ips.
        if str(ip) in ALWAYS_BLOCKED_METADATA_IPS:
            return True

        # Also block metadata IPs reached via NAT64 wrap. NAT64 prefixes
        # embed the IPv4 destination in the low 32 bits; even when the
        # operator has set LDR_SECURITY_ALLOW_NAT64=true the metadata
        # block is "always" — an opt-in for IPv4 reachability does NOT
        # license IMDS exposure.
        if is_nat64_wrapped_metadata_ip(ip):
            return True

        # Operator escape hatch for IPv6-only deployments using DNS64+NAT64.
        # Read lazily (not at import) so test monkeypatching works and so the
        # value is not cached across env mutations. Cloud-metadata IPs are
        # ALWAYS blocked above, so this carve-out cannot reopen IMDS via
        # the IPv6-wrapped form.
        #
        # allow_nat64 overrides the env read: None (default) preserves every
        # existing caller; an explicit bool lets the notification "Test"
        # admin hint ask "would LDR_SECURITY_ALLOW_NAT64=true unblock this?"
        # without touching process env.
        if allow_nat64 is None:
            from ..settings.env_registry import get_env_setting

            nat64_allowed = bool(get_env_setting("security.allow_nat64", False))
        else:
            nat64_allowed = allow_nat64

        # Check if IP is in any blocked range
        for blocked_range in BLOCKED_IP_RANGES:
            if ip in blocked_range:
                # NAT64 carve-out: when the operator has opted in, the two
                # NAT64 prefixes don't block. 6to4 / Teredo / discard remain
                # blocked unconditionally.
                if nat64_allowed and blocked_range in NAT64_PREFIXES:
                    continue
                # If allow_private_ips is True, skip blocking for private + loopback
                if allow_private_ips:
                    is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
                    is_private = any(ip in pr for pr in PRIVATE_RANGES)
                    if is_loopback or is_private:
                        continue
                # If allow_localhost is True, skip blocking for loopback only
                elif allow_localhost:
                    is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
                    if is_loopback:
                        continue
                return True

        return False

    except ValueError:
        # Invalid IP address
        return False


def validate_url(
    url: str,
    allow_localhost: bool = False,
    allow_private_ips: bool = False,
) -> bool:
    """
    Validate URL to prevent SSRF attacks.

    Checks:
    1. URL scheme is allowed (http/https only)
    2. Hostname is not an internal/private IP address
    3. Hostname does not resolve to an internal/private IP

    Args:
        url: URL to validate
        allow_localhost: Whether to allow localhost/loopback addresses.
            Set to True for trusted internal services like self-hosted
            search engines (e.g., searxng). Default False.
        allow_private_ips: Whether to allow all private/internal IPs plus localhost.
            This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
            used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
            private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
            like SearXNG or Ollama in containerized environments.
            Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
            (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
            are ALWAYS blocked regardless of these flags.

    Returns:
        True if URL is safe, False otherwise
    """
    if not isinstance(url, str):
        return False
    try:
        url = url.strip()
        # Layer 1: reject RFC-illegal characters that drive parser-differential
        # attacks (backslash, whitespace, control bytes). The URL is omitted
        # from this log line because userinfo (RFC 3986 §3.2.1) may contain
        # credentials and rejected URLs are by definition adversarial-shaped.
        if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
            logger.warning("Blocked URL containing RFC-illegal characters")
            return False

        parsed = urlparse(url)

        # Check scheme
        if parsed.scheme.lower() not in ALLOWED_SCHEMES:
            logger.warning(
                f"Blocked URL with invalid scheme: {parsed.scheme} - {redact_url_for_log(url)}"
            )
            return False

        # Layer 2: extract host using urllib3, the same parser ``requests``
        # uses internally. ``urlparse`` and urllib3 disagree on URLs like
        # ``http://127.0.0.1\@1.1.1.1`` — urlparse says ``1.1.1.1``,
        # urllib3 says ``127.0.0.1``. Validating against urllib3 means the
        # validator and the HTTP client cannot disagree on destination.
        try:
            u3 = parse_url(url)
        except LocationParseError:
            logger.warning("Blocked URL: urllib3 parser rejected it")
            return False
        hostname = u3.host
        # Authority must be ASCII printable. urllib3 currently rejects
        # non-ASCII via LocationParseError, but this guard keeps us
        # independent of that staying constant — CVE-2019-9636 showed
        # Python's stdlib loosened a similar restriction previously.
        # Brackets/colon used in IPv6 hosts are within 0x20-0x7e, so this
        # runs cleanly before bracket-strip.
        if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname):
            logger.warning("Blocked URL with non-ASCII / control bytes in host")
            return False
        # Strip IPv6 brackets so ipaddress.ip_address can parse the host.
        if hostname and hostname.startswith("[") and hostname.endswith("]"):
            hostname = hostname[1:-1]
        # rstrip(".") matches getaddrinfo behaviour — trailing dots are
        # ignored at resolution time.
        if hostname:
            hostname = hostname.rstrip(".")
        if not hostname:
            logger.warning(
                f"Blocked URL with no hostname: {redact_url_for_log(url)}"
            )
            return False

        # Check if hostname is an IP address
        try:
            ip = ipaddress.ip_address(hostname)
            if is_ip_blocked(
                str(ip),
                allow_localhost=allow_localhost,
                allow_private_ips=allow_private_ips,
            ):
                logger.warning(
                    f"Blocked URL with internal/private IP: {hostname} - {redact_url_for_log(url)}"
                )
                return False
        except ValueError:
            # Not an IP address, it's a hostname - need to resolve it
            pass

        # Resolve hostname to IP and check.
        #
        # NOTE: This is a best-effort, validation-time check. The caller
        # (typically safe_requests) hands the URL to requests/urllib3
        # afterwards, which resolves the hostname AGAIN at connect time --
        # a DNS rebinding TOCTOU window. Closing it would require pinning
        # the resolved IP into the outbound connection (HTTPAdapter shim
        # with server_hostname for SNI), which is HTTPS-only and doesn't
        # follow redirects cleanly. See SECURITY.md "Notification Webhook
        # SSRF" subsection for the accepted-risk rationale (the same
        # caveat applies here).
        try:
            # Get all IP addresses for hostname
            # nosec B104 - DNS resolution is intentional for SSRF prevention (checking if hostname resolves to private IP)
            addr_info = socket.getaddrinfo(
                hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
            )

            for info in addr_info:
                ip_str = str(
                    info[4][0]
                )  # Extract IP address from addr_info tuple

                if is_ip_blocked(
                    ip_str,
                    allow_localhost=allow_localhost,
                    allow_private_ips=allow_private_ips,
                ):
                    logger.warning(
                        f"Blocked URL - hostname {hostname} resolves to "
                        f"internal/private IP: {ip_str} - {redact_url_for_log(url)}"
                    )
                    return False

        except socket.gaierror:
            logger.warning(f"Failed to resolve hostname {hostname}")
            return False
        except Exception:
            logger.exception("Error during hostname resolution")
            return False

        # URL passes all checks
        return True

    except Exception:
        logger.exception(f"Error validating URL {redact_url_for_log(url)}")
        return False


def assert_base_url_safe(base_url: str, *, setting_key: str) -> str:
    """Validate an LLM provider base_url. Raises ValueError on SSRF.

    Args:
        base_url: The URL to validate.
        setting_key: The settings dot-path that produced this URL
            (e.g. ``"llm.ollama.url"``). Embedded into the error message
            so operators know which setting to fix. Pass ``cls.url_setting``
            from the OpenAI-compat parent or ``"llm.ollama.url"`` from
            the Ollama provider — NEVER ``cls.provider_name`` which is a
            display string ("xAI Grok", "llama.cpp") not a settings key.

    Uses ``allow_localhost=True, allow_private_ips=True`` because the
    legitimate destinations for LLM SDKs are localhost (Ollama, LM Studio,
    llama.cpp) and RFC1918 (Docker / private network deployments). The
    ``ALWAYS_BLOCKED_METADATA_IPS`` set still fires under those flags and
    prevents the auth-gated SSRF that would otherwise reach cloud-credential
    endpoints (AWS IMDS / ECS, Azure, OCI, DigitalOcean, AlibabaCloud,
    Tencent).

    Caveat: this guard validates once at provider construction. The SDK
    re-resolves the hostname on every inference call, so a hostile DNS
    authority can rebind between guard and connect — same TOCTOU as
    ``validate_url``. See SECURITY.md for the accepted-risk rationale.
    """
    if not validate_url(base_url, allow_localhost=True, allow_private_ips=True):
        raise ValueError(
            f"base_url failed SSRF validation: refusing to send "
            f"inference traffic. Check {setting_key} config."
        )
    return base_url


def get_safe_url(
    url: Optional[str], default: Optional[str] = None
) -> Optional[str]:
    """
    Get URL if it's safe, otherwise return default.

    Args:
        url: URL to validate
        default: Default value if URL is unsafe

    Returns:
        URL if safe, default otherwise
    """
    if not url:
        return default

    if validate_url(url):
        return url

    logger.warning(f"Unsafe URL rejected: {redact_url_for_log(url)}")
    return default


def redact_url_for_log(url: str) -> str:
    """Return ``scheme://host:port`` (no userinfo, path, query, fragment).

    For log output only. Drops everything except scheme + authority host
    + port to minimise the chance of leaking credentials, tokens, or
    sensitive paths into logs while still giving operators enough to
    distinguish ``http://10.0.0.1:80`` from ``https://10.0.0.1:443``.

    RFC 3986 §3.2.1 allows credentials in URL userinfo
    (``http://user:pass@host/``). A rejected URL is by definition
    adversarial-shaped, but it may still carry the operator's real
    credentials if a misconfiguration produced it.
    """
    try:
        u = parse_url(url)
        scheme = u.scheme or "?"
        host = u.host or "<no-host>"
        host_port = f"{host}:{u.port}" if u.port else host
        return f"{scheme}://{host_port}"
    except (LocationParseError, ValueError):
        return "<unparseable>"
