"""SSRF (Server-Side Request Forgery) protection for outbound HTTP requests.

Any feature that makes outbound HTTP requests (webhooks, MCP callbacks, avatar
fetch) must validate the target URL through this module before sending.

What is blocked:
    * Non-HTTPS schemes (http://, file://, ftp://, etc.)
    * Loopback: 127.0.0.0/8, ::1
    * RFC-1918 private ranges: 10.x, 172.16–31.x, 192.168.x
    * Link-local: 169.254.0.0/16 (AWS metadata at 169.254.169.254), fe80::/10
    * Unique-local IPv6: fc00::/7
    * Shared address space: 100.64.0.0/10 (carrier-grade NAT, RFC-6598)
    * Reserved / unspecified: 0.0.0.0/8, 240.0.0.0/4, ::/128

Two-layer defence:
    1. ``check_url_safe(url)`` — fast sync check (scheme + bare IP literal).
       Called from Pydantic field validators at request-parse time.  No DNS
       resolution; suitable for the async event loop.
    2. ``validate_outbound_url(url)`` — full async check (scheme + IP literal +
       DNS resolution via asyncio.to_thread).  Called by webhook delivery before
       each HTTP POST attempt.  Blocks SSRF via DNS rebinding.
"""

import asyncio
import ipaddress
import socket
from urllib.parse import urlparse

# ── Blocked IP ranges ─────────────────────────────────────────────────────────

_BLOCKED_NETWORKS: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = [
    ipaddress.ip_network("127.0.0.0/8"),        # Loopback IPv4
    ipaddress.ip_network("::1/128"),             # Loopback IPv6
    ipaddress.ip_network("10.0.0.0/8"),          # RFC-1918 class A
    ipaddress.ip_network("172.16.0.0/12"),       # RFC-1918 class B
    ipaddress.ip_network("192.168.0.0/16"),      # RFC-1918 class C
    ipaddress.ip_network("169.254.0.0/16"),      # Link-local IPv4 (AWS metadata)
    ipaddress.ip_network("fe80::/10"),           # Link-local IPv6
    ipaddress.ip_network("fc00::/7"),            # Unique-local IPv6
    ipaddress.ip_network("100.64.0.0/10"),       # Shared address space (RFC-6598)
    ipaddress.ip_network("0.0.0.0/8"),           # Unspecified IPv4
    ipaddress.ip_network("240.0.0.0/4"),         # Reserved IPv4
    ipaddress.ip_network("::/128"),              # Unspecified IPv6
]

def _is_blocked_ip(addr: str) -> bool:
    """Return True when *addr* falls within a blocked network range."""
    try:
        ip = ipaddress.ip_address(addr)
    except ValueError:
        return True  # unparseable address — block it
    return any(ip in net for net in _BLOCKED_NETWORKS)

# ── Sync check (scheme + bare IP literal) ─────────────────────────────────────

def check_url_safe(url: str) -> str:
    """Fast synchronous SSRF pre-check — no DNS resolution.

    Validates scheme (must be ``https``) and rejects bare IP literals that
    fall in blocked ranges.  Suitable for Pydantic field validators and other
    synchronous call sites.

    Args:
        url: The candidate outbound URL.

    Returns:
        The original *url* string if safe.

    Raises:
        ValueError: When the URL is malformed, uses a non-HTTPS scheme, or
            contains a bare IP literal in a blocked range.
    """
    try:
        parsed = urlparse(url)
    except Exception as exc:  # pragma: no cover — urlparse is very forgiving
        raise ValueError(f"Malformed URL: {exc}") from exc

    if parsed.scheme != "https":
        raise ValueError(
            f"Outbound URL must use https:// — got {parsed.scheme!r}. "
            "Non-HTTPS schemes are blocked to prevent credential leakage and SSRF."
        )

    hostname = parsed.hostname
    if not hostname:
        raise ValueError("URL has no hostname.")

    # If the hostname is a literal IP address, check it immediately.
    # DNS-based hostnames are deferred to the async validate_outbound_url check.
    try:
        ipaddress.ip_address(hostname)
        # It parsed — it is a bare IP literal.
        if _is_blocked_ip(hostname):
            raise ValueError(
                f"Outbound URL targets a private/reserved IP address ({hostname}). "
                "Requests to RFC-1918, loopback, and link-local addresses are blocked."
            )
    except ValueError as exc:
        if "private/reserved" in str(exc) or "blocked" in str(exc):
            raise
        # Not an IP literal — hostname will be resolved at delivery time.

    return url

# ── Async check (scheme + IP literal + DNS resolution) ────────────────────────

async def validate_outbound_url(url: str) -> str:
    """Full async SSRF check including DNS resolution.

    Resolves the hostname and verifies that none of the resulting IP addresses
    fall in a blocked range.  Use this immediately before making an outbound
    HTTP request to guard against DNS rebinding attacks.

    Args:
        url: The candidate outbound URL.

    Returns:
        The original *url* string if safe.

    Raises:
        ValueError: When the URL fails any SSRF check.
    """
    # Perform scheme + bare IP check first (cheap, no I/O).
    check_url_safe(url)

    parsed = urlparse(url)
    hostname = parsed.hostname or ""

    # If it's already a bare IP literal, it passed the sync check above.
    # No DNS needed.
    try:
        ipaddress.ip_address(hostname)
        return url  # bare IP — already validated
    except ValueError:
        pass  # not an IP literal — proceed to DNS resolution

    # Resolve via getaddrinfo in a thread so the event loop is not blocked.
    try:
        infos: list[tuple[int, int, int, str, tuple[str, int]]] = await asyncio.to_thread(socket.getaddrinfo, hostname, None)
    except socket.gaierror as exc:
        raise ValueError(
            f"Outbound URL hostname cannot be resolved: {hostname!r} — {exc}"
        ) from exc

    for _family, _type, _proto, _canonname, sockaddr in infos:
        ip_str = sockaddr[0]
        if _is_blocked_ip(ip_str):
            raise ValueError(
                f"Outbound URL {hostname!r} resolves to a private/reserved address "
                f"({ip_str}). RFC-1918 / loopback / link-local targets are blocked."
            )

    return url