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

from .errors import ComputerRuntimeError

ALLOWED_SCHEMES = {"http", "https"}
INTERNAL_BROWSER_SCHEMES = {"about", "blob", "data"}
BLOCKED_HOSTS = {
    "localhost",
    "localhost.localdomain",
    "metadata",
    "metadata.google.internal",
}


def _address_is_blocked(address: str) -> bool:
    ip = ipaddress.ip_address(address.split("%", 1)[0])
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
        ip = ip.ipv4_mapped
    return any((
        ip.is_private,
        ip.is_loopback,
        ip.is_link_local,
        ip.is_multicast,
        ip.is_reserved,
        ip.is_unspecified,
    ))


async def validate_public_url(url: str, *, allow_internal_browser_url: bool = False) -> str:
    if not isinstance(url, str) or len(url) > 4096:
        raise ComputerRuntimeError("BROWSER_URL_INVALID", "The browser URL is invalid.", 422)
    try:
        parsed = urlsplit(url)
        port = parsed.port
    except ValueError as error:
        raise ComputerRuntimeError("BROWSER_URL_INVALID", "The browser URL is invalid.", 422) from error
    scheme = parsed.scheme.lower()
    if allow_internal_browser_url and scheme in INTERNAL_BROWSER_SCHEMES:
        return url
    if scheme not in ALLOWED_SCHEMES or not parsed.hostname:
        raise ComputerRuntimeError(
            "BROWSER_URL_SCHEME_BLOCKED", "Only public HTTP and HTTPS URLs are allowed.", 422
        )
    if parsed.username is not None or parsed.password is not None:
        raise ComputerRuntimeError(
            "BROWSER_URL_CREDENTIALS_BLOCKED", "Credentials in browser URLs are not allowed.", 422
        )
    host = parsed.hostname.rstrip(".").lower()
    if host in BLOCKED_HOSTS or host.endswith((".localhost", ".local", ".internal")):
        raise ComputerRuntimeError("BROWSER_DESTINATION_BLOCKED", "The destination is blocked.", 403)
    try:
        literal = ipaddress.ip_address(host)
    except ValueError:
        literal = None
    if literal is not None:
        if _address_is_blocked(str(literal)):
            raise ComputerRuntimeError(
                "BROWSER_DESTINATION_BLOCKED", "The destination is blocked.", 403
            )
        return url
    loop = asyncio.get_running_loop()
    try:
        results = await loop.run_in_executor(
            None,
            lambda: socket.getaddrinfo(
                host, port or (443 if scheme == "https" else 80),
                type=socket.SOCK_STREAM,
            ),
        )
    except socket.gaierror as error:
        raise ComputerRuntimeError(
            "BROWSER_DNS_FAILED", "The destination could not be resolved.", 422
        ) from error
    addresses = {item[4][0] for item in results}
    if not addresses or any(_address_is_blocked(address) for address in addresses):
        raise ComputerRuntimeError("BROWSER_DESTINATION_BLOCKED", "The destination is blocked.", 403)
    return url