"""URL, DNS and SSRF validation shared by HTTP, browser and downloader layers."""

from __future__ import annotations

import asyncio
import ipaddress
import logging
import re
import socket
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from urllib.parse import urlsplit, urlunsplit

import aiohttp

from app.config import normalize_domain
from app.errors import DomainNotAllowedError, InvalidUrlError, NetworkError, SsrfBlockedError

logger = logging.getLogger(__name__)

_HOST_RE = re.compile(r"^(?!-)[a-z0-9-]{1,63}(?<!-)(\.(?!-)[a-z0-9-]{1,63}(?<!-))*$")
_BAD_HOST_SUFFIXES = (".localhost", ".local", ".internal", ".home.arpa", ".lan")
_BAD_HOSTS = {"localhost", "local", "internal", "metadata.google.internal"}
# Resolvers and URL libraries disagree about shorthand/hex/octal IPv4 forms
# (for example ``127.1`` or ``0x7f000001``). Treat every numeric-looking host
# that is not a canonical IP literal as ambiguous and block it.
_AMBIGUOUS_NUMERIC_HOST_RE = re.compile(
    r"^(?:\d+|\d+(?:\.\d+){1,4}|0[xX][0-9a-fA-F]+(?:\.[0-9a-fA-F]+)*)$"
)
_MAX_URL_LENGTH = 4096

AddressResolver = Callable[[str, int], Awaitable[list[tuple[str, int, int, str]]]]


@dataclass(frozen=True, slots=True)
class ValidatedUrl:
    original: str
    url: str
    host: str
    port: int
    addresses: tuple[str, ...]


def _default_resolver(host: str, port: int) -> Awaitable[list[tuple[str, int, int, str]]]:
    async def resolve() -> list[tuple[str, int, int, str]]:
        loop = asyncio.get_running_loop()
        try:
            return await asyncio.wait_for(
                loop.getaddrinfo(host, port, type=socket.SOCK_STREAM),
                timeout=8.0,
            )
        except (OSError, TimeoutError) as exc:
            raise NetworkError(f"DNS resolution failed for {host}") from exc

    return resolve()


class NetworkGuard:
    """Reject non-public destinations before every network operation.

    DNS answers are cached briefly and every returned address is checked. The
    actual aiohttp connection uses the validated addresses through
    :class:`ValidatingResolver`, closing the usual DNS-rebinding gap.
    """

    def __init__(
        self,
        allowed_domains: list[str] | tuple[str, ...],
        *,
        resolver: AddressResolver | None = None,
        dns_cache_ttl: float = 30.0,
    ) -> None:
        self.allowed_domains = frozenset(normalize_domain(domain) for domain in allowed_domains)
        if not self.allowed_domains:
            raise ValueError("at least one allowed domain is required")
        self._resolver = resolver or _default_resolver
        self._cache_ttl = dns_cache_ttl
        self._cache: dict[tuple[str, int], tuple[float, tuple[str, ...]]] = {}
        self._locks: dict[tuple[str, int], asyncio.Lock] = {}
        self._lock = asyncio.Lock()

    @staticmethod
    def _normalize_host(host: str) -> str:
        value = host.strip().strip(".").lower()
        if not value:
            raise InvalidUrlError("URL has no hostname")
        if value.startswith("[") and value.endswith("]"):
            value = value[1:-1]
        try:
            ip = ipaddress.ip_address(value)
        except ValueError:
            pass
        else:
            return ip.compressed
        if value in _BAD_HOSTS or value.endswith(_BAD_HOST_SUFFIXES):
            raise SsrfBlockedError("internal hostname is blocked")
        if _AMBIGUOUS_NUMERIC_HOST_RE.fullmatch(value):
            raise SsrfBlockedError("ambiguous numeric hostname is blocked")
        try:
            ascii_value = value.encode("idna").decode("ascii")
        except UnicodeError as exc:
            raise InvalidUrlError("invalid internationalized hostname") from exc
        if not _HOST_RE.fullmatch(ascii_value):
            raise InvalidUrlError("hostname contains invalid characters")
        return ascii_value

    def is_domain_allowed(self, host: str) -> bool:
        normalized = self._normalize_host(host)
        return any(
            normalized == allowed or normalized.endswith("." + allowed)
            for allowed in self.allowed_domains
        )

    @staticmethod
    def _validate_ip(address: str) -> str:
        try:
            ip = ipaddress.ip_address(address.split("%", 1)[0])
        except ValueError as exc:
            raise SsrfBlockedError("DNS returned an invalid address") from exc
        if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
            ip = ip.ipv4_mapped
        if (
            not ip.is_global
            or ip.is_private
            or ip.is_loopback
            or ip.is_link_local
            or ip.is_multicast
            or ip.is_reserved
            or ip.is_unspecified
        ):
            raise SsrfBlockedError(f"non-public IP address blocked: {ip}")
        return ip.compressed

    async def resolve_addresses(self, host: str, port: int) -> tuple[str, ...]:
        normalized = self._normalize_host(host)
        try:
            literal = ipaddress.ip_address(normalized)
        except ValueError:
            literal = None
        if literal is not None:
            return (self._validate_ip(normalized),)

        key = (normalized, port)
        now = time.monotonic()
        async with self._lock:
            cached = self._cache.get(key)
            if cached and now - cached[0] < self._cache_ttl:
                return cached[1]
            lock = self._locks.setdefault(key, asyncio.Lock())
        async with lock:
            now = time.monotonic()
            cached = self._cache.get(key)
            if cached and now - cached[0] < self._cache_ttl:
                return cached[1]
            try:
                infos = await self._resolver(normalized, port)
            except NetworkError:
                raise
            except Exception as exc:  # noqa: BLE001 - normalize resolver failures
                raise NetworkError(f"DNS resolution failed for {normalized}") from exc
            addresses: list[str] = []
            for info in infos:
                sockaddr = info[4]
                if not sockaddr:
                    continue
                addresses.append(self._validate_ip(str(sockaddr[0])))
            unique = tuple(dict.fromkeys(addresses))
            if not unique:
                raise NetworkError(f"DNS returned no usable addresses for {normalized}")
            async with self._lock:
                self._cache[key] = (now, unique)
                if len(self._cache) > 512:
                    oldest = min(self._cache, key=lambda item: self._cache[item][0])
                    self._cache.pop(oldest, None)
            return unique

    async def validate(self, raw_url: str, *, resolve_dns: bool = True) -> ValidatedUrl:
        if not isinstance(raw_url, str):
            raise InvalidUrlError("URL must be text")
        value = raw_url.strip()
        if not value or len(value) > _MAX_URL_LENGTH:
            raise InvalidUrlError("URL is empty or too long")
        if any(ord(char) < 0x20 or ord(char) == 0x7F for char in value):
            raise InvalidUrlError("URL contains control characters")
        if "\\" in value:
            raise InvalidUrlError("backslashes are not allowed in URLs")
        try:
            parts = urlsplit(value)
        except ValueError as exc:
            raise InvalidUrlError("URL cannot be parsed") from exc
        scheme = parts.scheme.lower()
        if scheme not in {"http", "https"}:
            raise InvalidUrlError("only http and https URLs are accepted")
        if parts.username is not None or parts.password is not None or "@" in parts.netloc:
            raise InvalidUrlError("credentials in URLs are not allowed")
        if not parts.hostname:
            raise InvalidUrlError("URL has no hostname")
        try:
            host = self._normalize_host(parts.hostname)
            explicit_port = parts.port
        except ValueError as exc:
            raise InvalidUrlError("URL has an invalid port") from exc
        default_port = 443 if scheme == "https" else 80
        if explicit_port is not None and not 1 <= explicit_port <= 65535:
            raise InvalidUrlError("URL port is out of range")
        port = explicit_port if explicit_port is not None else default_port
        try:
            ip_literal = ipaddress.ip_address(host)
        except ValueError:
            ip_literal = None
        if ip_literal is not None:
            self._validate_ip(str(ip_literal))
        if not self.is_domain_allowed(host):
            raise DomainNotAllowedError(
                f"domain {host!r} is not present in ALLOWED_DOMAINS"
            )

        netloc = f"[{host}]" if ":" in host else host
        if explicit_port is not None and explicit_port != default_port:
            netloc = f"{netloc}:{explicit_port}"
        path = parts.path or "/"
        normalized_url = urlunsplit((scheme, netloc, path, parts.query, ""))

        if resolve_dns:
            addresses = await self.resolve_addresses(host, port)
        else:
            addresses = ()
        return ValidatedUrl(
            original=raw_url,
            url=normalized_url,
            host=host,
            port=port,
            addresses=addresses,
        )

    def validate_host_only(self, raw_url: str) -> ValidatedUrl:
        """Synchronous structural/allow-list check; DNS must be checked separately."""

        return self._validate_without_dns(raw_url)

    def _validate_without_dns(self, raw_url: str) -> ValidatedUrl:
        value = raw_url.strip()
        if not value or len(value) > _MAX_URL_LENGTH:
            raise InvalidUrlError("URL is empty or too long")
        if any(ord(char) < 0x20 or ord(char) == 0x7F for char in value):
            raise InvalidUrlError("URL contains control characters")
        if "\\" in value:
            raise InvalidUrlError("backslashes are not allowed in URLs")
        try:
            parts = urlsplit(value)
        except ValueError as exc:
            raise InvalidUrlError("URL cannot be parsed") from exc
        scheme = parts.scheme.lower()
        if scheme not in {"http", "https"}:
            raise InvalidUrlError("only http and https URLs are accepted")
        if parts.username is not None or parts.password is not None or "@" in parts.netloc:
            raise InvalidUrlError("credentials in URLs are not allowed")
        if not parts.hostname:
            raise InvalidUrlError("URL has no hostname")
        try:
            host = self._normalize_host(parts.hostname)
            explicit_port = parts.port
        except ValueError as exc:
            raise InvalidUrlError("URL has an invalid port") from exc
        default_port = 443 if scheme == "https" else 80
        if explicit_port is not None and not 1 <= explicit_port <= 65535:
            raise InvalidUrlError("URL port is out of range")
        port = explicit_port if explicit_port is not None else default_port
        try:
            ip_literal = ipaddress.ip_address(host)
        except ValueError:
            ip_literal = None
        if ip_literal is not None:
            self._validate_ip(str(ip_literal))
        if not self.is_domain_allowed(host):
            raise DomainNotAllowedError("domain is not present in ALLOWED_DOMAINS")
        netloc = f"[{host}]" if ":" in host else host
        if explicit_port is not None and explicit_port != default_port:
            netloc = f"{netloc}:{explicit_port}"
        return ValidatedUrl(
            original=raw_url,
            url=urlunsplit((scheme, netloc, parts.path or "/", parts.query, "")),
            host=host,
            port=port,
            addresses=(),
        )


class ValidatingResolver(aiohttp.abc.AbstractResolver):
    """aiohttp resolver that only returns globally routable, checked addresses."""

    def __init__(self, guard: NetworkGuard) -> None:
        self.guard = guard

    async def resolve(
        self,
        host: str,
        port: int = 0,
        family: int = socket.AF_INET,
    ) -> list[dict[str, object]]:
        addresses = await self.guard.resolve_addresses(host, port)
        results: list[dict[str, object]] = []
        for address in addresses:
            ip = ipaddress.ip_address(address)
            resolved_family = socket.AF_INET6 if ip.version == 6 else socket.AF_INET
            sockaddr: tuple[object, ...]
            if resolved_family == socket.AF_INET6:
                sockaddr = (address, port, 0, 0)
            else:
                sockaddr = (address, port)
            results.append(
                {
                    "hostname": host,
                    "host": address,
                    "port": port,
                    "family": resolved_family,
                    "proto": socket.IPPROTO_TCP,
                    "flags": 0,
                    "sockaddr": sockaddr,
                }
            )
        return results

    async def close(self) -> None:
        return None