"""Outbound URL validation to mitigate SSRF attacks.

User-controlled URLs (provider ``url``, webhook target ``url``,
shared-link base URLs, image downloads) must be validated before any
HTTP request is issued. This module rejects schemes other than
http/https and blocks destinations that resolve to private, loopback,
link-local, unspecified, CGNAT (100.64.0.0/10), or IPv4-mapped IPv6
ranges.

DNS rebinding mitigation
~~~~~~~~~~~~~~~~~~~~~~~~
``avalidate_outbound_url`` returns the original URL on success, but
also returns the resolved IP it actually validated. Callers that pass
the validated URL straight into ``aiohttp`` are vulnerable to a
DNS-rebinding attack: the validator's ``getaddrinfo`` returns a public
IP; aiohttp's connect-time resolution returns ``127.0.0.1``. To close
that gap, use :func:`build_ssrf_safe_session` (or
:class:`PinnedResolver`) so the resolved IP from the validation step is
the one aiohttp connects to.

Set ``NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS=1`` in the environment for
development against localhost services.
"""

from __future__ import annotations

import asyncio
import ipaddress
import logging
import os
import socket
from dataclasses import dataclass
from urllib.parse import urlparse

import aiohttp

_LOGGER = logging.getLogger(__name__)

_ALLOW_PRIVATE = os.environ.get("NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS") == "1"
_ALLOWED_SCHEMES = frozenset({"http", "https"})

# Carrier-grade NAT range. Not in stdlib's ``is_private``; an attacker
# pointing a domain at a CGNAT IP could reach the operator's ISP-side
# routing infrastructure. RFC 6598.
_CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10")

if _ALLOW_PRIVATE:  # pragma: no cover — operator-visible banner
    _LOGGER.warning(
        "SSRF guard: private-URL bypass ENABLED "
        "(NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS=1). Requests to RFC1918 / "
        "loopback / link-local hosts will be permitted."
    )


class UnsafeURLError(ValueError):
    """Raised when a URL targets a disallowed network destination."""


@dataclass(frozen=True)
class ValidatedURL:
    """Result of validating an outbound URL.

    Attributes:
        url: The original URL string (unchanged).
        host: Hostname extracted from the URL (lower-cased, IDN-encoded).
        ip: Resolved IP address that passed the block-range check, as a
            string. Pass to :class:`PinnedResolver` to defeat DNS
            rebinding by reusing this exact IP at connect time.
    """

    url: str
    host: str
    ip: str


def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
    # An IPv4-mapped IPv6 like ``::ffff:127.0.0.1`` is NOT considered
    # ``is_private`` etc. by stdlib — the v4 view holds those flags. So
    # we unwrap before checking.
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        ip = ip.ipv4_mapped
    return (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
        or (isinstance(ip, ipaddress.IPv4Address) and ip in _CGNAT_NETWORK)
    )


def _safe_host_repr(host: str) -> str:
    """Return ``host`` shortened/escaped for safe inclusion in error text."""
    h = host[:64].replace("\r", "").replace("\n", "")
    return h


def _normalize_host(parsed_host: str) -> str:
    """Normalize a hostname: lowercase, strip trailing dot, IDN-encode."""
    host = parsed_host.lower()
    if host.endswith("."):
        host = host[:-1]
    # Strip IPv6 zone id ("fe80::1%eth0") — must not reach the resolver.
    if "%" in host:
        host = host.split("%", 1)[0]
    # IDN-encode unicode hostnames so we don't downgrade to confusables
    # in any later log/output and so getaddrinfo gets the ascii form.
    try:
        if any(ord(c) > 127 for c in host):
            host = host.encode("idna").decode("ascii")
    except UnicodeError:
        # Caller will fail on resolution; leave as-is so the error path
        # surfaces a "DNS resolution failed" rather than a stack trace.
        pass
    return host


def _check_scheme_host(url: str) -> tuple[str, str]:
    if not isinstance(url, str) or not url:
        raise UnsafeURLError("URL is empty")
    parsed = urlparse(url)
    scheme = parsed.scheme.lower()
    if scheme not in _ALLOWED_SCHEMES:
        raise UnsafeURLError(f"Scheme '{scheme[:16]}' not allowed")
    host = parsed.hostname
    if not host:
        raise UnsafeURLError("URL has no host")
    return scheme, _normalize_host(host)


def _select_addresses(
    host: str, infos: list[tuple],
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
    """Return parsed, non-blocked IPs from ``getaddrinfo`` results."""
    addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
    for info in infos:
        sockaddr = info[4]
        try:
            ip = ipaddress.ip_address(sockaddr[0])
        except ValueError:
            continue
        if _is_blocked_ip(ip):
            raise UnsafeURLError(
                f"Host {_safe_host_repr(host)} resolves to blocked address {ip}"
            )
        addrs.append(ip)
    if not addrs:
        raise UnsafeURLError(f"Host {_safe_host_repr(host)} has no usable address")
    return addrs


def validate_outbound_url(url: str) -> str:
    """Validate ``url`` is safe to fetch; returns the URL on success.

    .. deprecated::
        Synchronous; uses blocking ``socket.getaddrinfo``. Prefer
        :func:`avalidate_outbound_url` from async code paths so the
        event loop isn't blocked, and use :func:`build_ssrf_safe_session`
        to defeat DNS rebinding.
    """
    _, host = _check_scheme_host(url)

    if _ALLOW_PRIVATE:
        return url

    try:
        ip = ipaddress.ip_address(host)
        if _is_blocked_ip(ip):
            raise UnsafeURLError(f"Host {_safe_host_repr(host)} is in a blocked range")
        return url
    except ValueError:
        pass

    try:
        infos = socket.getaddrinfo(host, None)
    except (socket.gaierror, UnicodeError, OSError) as exc:
        # ``UnicodeError`` covers IDNA failures (labels >63 chars, malformed
        # unicode) which getaddrinfo surfaces as encoding errors rather than
        # gaierror. ``OSError`` covers transient resolver failures on some
        # platforms.
        raise UnsafeURLError(f"DNS resolution failed for {_safe_host_repr(host)}") from exc
    _select_addresses(host, infos)
    return url


async def avalidate_outbound_url(url: str) -> str:
    """Async variant — returns the URL on success.

    For DNS-rebinding-safe usage, prefer :func:`avalidate_outbound_url_full`
    which also returns the resolved IP for connect-time pinning.
    """
    result = await avalidate_outbound_url_full(url)
    return result.url


async def avalidate_outbound_url_full(url: str) -> ValidatedURL:
    """Validate ``url`` and return a :class:`ValidatedURL` on success.

    The returned ``ip`` field is the IP that passed the block-range
    check. Pair with :class:`PinnedResolver` so aiohttp connects to that
    exact IP and a malicious DNS server can't swap in a private address
    after validation.
    """
    _, host = _check_scheme_host(url)

    if _ALLOW_PRIVATE:
        # In dev mode we still resolve to give a usable IP, but we don't
        # gate on the result.
        try:
            ip = str(ipaddress.ip_address(host))
        except ValueError:
            try:
                loop = asyncio.get_running_loop()
                infos = await loop.getaddrinfo(host, None)
                ip = infos[0][4][0] if infos else host
            except (socket.gaierror, OSError):
                ip = host
        return ValidatedURL(url=url, host=host, ip=ip)

    # Literal IP host
    try:
        ip_obj = ipaddress.ip_address(host)
        if _is_blocked_ip(ip_obj):
            raise UnsafeURLError(f"Host {_safe_host_repr(host)} is in a blocked range")
        return ValidatedURL(url=url, host=host, ip=str(ip_obj))
    except ValueError:
        pass

    loop = asyncio.get_running_loop()
    try:
        infos = await loop.getaddrinfo(host, None)
    except (socket.gaierror, UnicodeError, OSError) as exc:
        raise UnsafeURLError(f"DNS resolution failed for {_safe_host_repr(host)}") from exc
    addrs = _select_addresses(host, infos)
    return ValidatedURL(url=url, host=host, ip=str(addrs[0]))


class PinnedResolver(aiohttp.abc.AbstractResolver):
    """aiohttp resolver that returns a fixed (host, ip) mapping.

    Used to pin the resolved IP from :func:`avalidate_outbound_url_full`
    so aiohttp's connect-time resolution can't be tricked by DNS
    rebinding into using a different IP than the one we validated.

    Falls back to :class:`aiohttp.AsyncResolver` (or default) for any
    host not explicitly pinned, so a single resolver instance can be
    reused across multiple validated URLs.
    """

    def __init__(self, mapping: dict[str, str] | None = None) -> None:
        self._map: dict[str, str] = dict(mapping or {})
        self._fallback: aiohttp.abc.AbstractResolver | None = None

    def pin(self, host: str, ip: str) -> None:
        self._map[host.lower()] = ip

    async def resolve(
        self, host: str, port: int = 0, family: int = socket.AF_INET,
    ) -> list[dict]:
        ip = self._map.get(host.lower())
        if ip is not None:
            try:
                ip_obj = ipaddress.ip_address(ip)
            except ValueError:
                ip_obj = None
            if ip_obj is not None:
                fam = socket.AF_INET6 if ip_obj.version == 6 else socket.AF_INET
                return [{
                    "hostname": host,
                    "host": ip,
                    "port": port,
                    "family": fam,
                    "proto": 0,
                    "flags": socket.AI_NUMERICHOST,
                }]
        if self._fallback is None:
            self._fallback = aiohttp.ThreadedResolver()
        return await self._fallback.resolve(host, port, family)

    async def close(self) -> None:
        if self._fallback is not None:
            await self._fallback.close()
