from __future__ import annotations

import ipaddress
import errno
import http.client
import os
import socket
import urllib.parse
import urllib.request
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import BinaryIO, Final


DEFAULT_STRUCTURED_RESPONSE_BYTES: Final = 16 * 1024 * 1024
DEFAULT_FILE_TRANSFER_BYTES: Final = 512 * 1024 * 1024
_READ_CHUNK_BYTES: Final = 64 * 1024
_TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"})
_FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"})
_IPV4_LIMITED_BROADCAST: Final = ipaddress.IPv4Address("255.255.255.255")
_IPV4_COMPATIBLE_NETWORK: Final = ipaddress.IPv6Network("::/96")
_NAT64_WELL_KNOWN_NETWORK: Final = ipaddress.IPv6Network("64:ff9b::/96")
_KNOWN_METADATA_ADDRESSES: Final = frozenset(
    {
        ipaddress.ip_address("100.100.100.200"),
        ipaddress.ip_address("169.254.169.254"),
        ipaddress.ip_address("fd00:ec2::254"),
    }
)


class OutboundHttpError(RuntimeError):
    """Base error for deployment-wide outbound HTTP policy failures."""


class OutboundHttpBlocked(OutboundHttpError):
    """Raised when a URL targets an address forbidden by deployment policy."""


class OutboundResponseTooLarge(OutboundHttpError):
    """Raised before a connector can retain an oversized remote response."""


@dataclass(frozen=True, slots=True)
class OutboundHttpPolicy:
    allow_private_networks: bool
    structured_response_bytes: int
    file_transfer_bytes: int


def outbound_http_policy(environ: Mapping[str, str] | None = None) -> OutboundHttpPolicy:
    env = os.environ if environ is None else environ
    return OutboundHttpPolicy(
        allow_private_networks=_private_network_default(env),
        structured_response_bytes=_positive_int(
            env.get("GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES"),
            default=DEFAULT_STRUCTURED_RESPONSE_BYTES,
            name="GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES",
        ),
        file_transfer_bytes=_positive_int(
            env.get("GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES"),
            default=DEFAULT_FILE_TRANSFER_BYTES,
            name="GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES",
        ),
    )


def validate_outbound_http_url(
    value: str,
    *,
    label: str = "Connector URL",
    policy: OutboundHttpPolicy | None = None,
) -> str:
    parsed = urllib.parse.urlparse(str(value).strip())
    if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname:
        raise OutboundHttpBlocked(f"{label} must be an absolute HTTP(S) URL")
    if parsed.username or parsed.password:
        raise OutboundHttpBlocked(f"{label} must not include embedded credentials")
    try:
        port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
    except ValueError as exc:
        raise OutboundHttpBlocked(f"{label} has an invalid port") from exc
    validate_outbound_host(parsed.hostname, port=port, label=label, policy=policy)
    return urllib.parse.urlunparse(parsed)


def validate_unpinned_sdk_http_url(
    value: str,
    *,
    label: str,
    policy: OutboundHttpPolicy | None = None,
) -> str:
    """Fail closed when an SDK cannot connect to a prevalidated DNS answer."""

    active_policy = policy or outbound_http_policy()
    validate_outbound_http_url(value, label=label, policy=active_policy)
    _raise_unpinned_transport(label)


def validate_unpinned_sdk_host(
    hostname: str,
    *,
    port: int,
    label: str,
    policy: OutboundHttpPolicy | None = None,
) -> None:
    """Validate a host and then reject an SDK that may select another peer."""

    active_policy = policy or outbound_http_policy()
    validate_outbound_host(hostname, port=port, label=label, policy=active_policy)
    _raise_unpinned_transport(label)


def _raise_unpinned_transport(label: str) -> None:
    raise OutboundHttpBlocked(
        f"{label} uses an SDK that cannot pin every connection peer or revalidate SDK-managed redirects/referrals; "
        "it is disabled until that transport supports connection-time DNS/IP pinning"
    )


def validate_outbound_host(
    hostname: str,
    *,
    port: int,
    label: str = "Connector host",
    policy: OutboundHttpPolicy | None = None,
) -> tuple[str, ...]:
    active_policy = policy or outbound_http_policy()
    records = _resolved_address_records(hostname, port=port, label=label, policy=active_policy)
    addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in records if item[4]))
    return addresses


def create_outbound_connection(
    hostname: str,
    port: int,
    timeout: float | object | None = None,
    source_address: tuple[str, int] | None = None,
    socket_options: Iterable[tuple[object, ...]] | None = None,
    *,
    label: str = "Connector host",
    policy: OutboundHttpPolicy | None = None,
) -> socket.socket:
    """Resolve, validate, and connect to the exact approved address records.

    Hostname resolution happens exactly once for this connection attempt. The
    returned socket connects directly to one of those validated sockaddr
    records, while higher protocol layers retain the original hostname for
    HTTP Host, TLS SNI, and certificate verification.
    """

    active_policy = policy or outbound_http_policy()
    records = _resolved_address_records(hostname, port=port, label=label, policy=active_policy)
    effective_timeout = None if timeout is socket._GLOBAL_DEFAULT_TIMEOUT else timeout  # type: ignore[attr-defined]
    last_error: OSError | None = None
    for family, socktype, proto, _canonname, sockaddr in records:
        sock: socket.socket | None = None
        try:
            sock = socket.socket(family, socktype, proto)
            sock.settimeout(effective_timeout)  # type: ignore[arg-type]
            if source_address is not None:
                bind_address: tuple[object, ...] = source_address
                if family == socket.AF_INET6 and len(source_address) == 2:
                    bind_address = (source_address[0], source_address[1], 0, 0)
                sock.bind(bind_address)
            for option in socket_options or ():
                sock.setsockopt(*option)
            try:
                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            except OSError as exc:
                if exc.errno != errno.ENOPROTOOPT:
                    raise
            sock.connect(sockaddr)
            return sock
        except OSError as exc:
            last_error = exc
            if sock is not None:
                sock.close()
    if last_error is not None:
        raise last_error
    raise OutboundHttpBlocked(f"{label} hostname did not resolve to a usable address")


def build_outbound_http_opener(*handlers: urllib.request.BaseHandler) -> urllib.request.OpenerDirector:
    """Build a proxy-free urllib opener whose sockets use approved addresses."""

    return urllib.request.build_opener(
        urllib.request.ProxyHandler({}),
        _OutboundHTTPHandler(),
        _OutboundHTTPSHandler(),
        *handlers,
    )


def response_limit(kind: str, *, policy: OutboundHttpPolicy | None = None) -> int:
    active_policy = policy or outbound_http_policy()
    if kind == "structured":
        return active_policy.structured_response_bytes
    if kind == "file":
        return active_policy.file_transfer_bytes
    raise ValueError("Response kind must be 'structured' or 'file'")


def bounded_response_bytes(
    stream: BinaryIO,
    *,
    headers: Mapping[str, str] | None = None,
    max_bytes: int | None = None,
    kind: str = "structured",
    label: str = "Connector response",
) -> bytes:
    configured_limit = response_limit(kind)
    limit = configured_limit if max_bytes is None else min(int(max_bytes), configured_limit)
    if limit <= 0:
        raise ValueError("max_bytes must be positive")
    declared_size = _content_length(headers or {})
    if declared_size is not None and declared_size > limit:
        raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes")
    body = bytearray()
    while len(body) <= limit:
        chunk = stream.read(min(_READ_CHUNK_BYTES, limit + 1 - len(body)))
        if not chunk:
            return bytes(body)
        body.extend(chunk)
    raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes")


def bounded_chunks_bytes(
    chunks: Iterable[bytes],
    *,
    headers: Mapping[str, str] | None = None,
    max_bytes: int | None = None,
    kind: str = "structured",
    label: str = "Connector response",
) -> bytes:
    configured_limit = response_limit(kind)
    limit = configured_limit if max_bytes is None else min(int(max_bytes), configured_limit)
    if limit <= 0:
        raise ValueError("max_bytes must be positive")
    declared_size = _content_length(headers or {})
    if declared_size is not None and declared_size > limit:
        raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes")
    body = bytearray()
    for chunk in chunks:
        if len(chunk) > limit - len(body):
            raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes")
        body.extend(chunk)
    return bytes(body)


def _private_network_default(environ: Mapping[str, str]) -> bool:
    configured = environ.get("GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS")
    if configured is not None and configured.strip():
        value = configured.strip().lower()
        if value in _TRUE_VALUES:
            return True
        if value in _FALSE_VALUES:
            return False
        raise ValueError("GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS must be true or false")
    return environ.get("APP_ENV", "dev").strip().lower() in {"dev", "development", "test"}


def _positive_int(value: str | None, *, default: int, name: str) -> int:
    if value is None or not value.strip():
        return default
    try:
        parsed = int(value)
    except ValueError as exc:
        raise ValueError(f"{name} must be a positive integer") from exc
    if parsed <= 0:
        raise ValueError(f"{name} must be a positive integer")
    return parsed


def _content_length(headers: Mapping[str, str]) -> int | None:
    value = next((item for key, item in headers.items() if key.casefold() == "content-length"), None)
    if value is None:
        return None
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return None
    return parsed if parsed >= 0 else None


def _is_public_address(value: str) -> bool:
    try:
        address = ipaddress.ip_address(value)
    except ValueError:
        return False
    if address.is_reserved or getattr(address, "is_site_local", False) or not address.is_global:
        return False
    return all(
        embedded.is_global and not embedded.is_reserved
        for embedded in _embedded_ipv4_addresses(address)
    )


def _is_forbidden_special_address(value: str) -> bool:
    """Keep connector access away from host-local and non-unicast address space.

    Enabling private-network connectors deliberately permits internal and
    loopback destinations, but it must not expose link-local metadata services
    or addresses that cannot identify a single remote peer.
    """

    try:
        address = ipaddress.ip_address(value)
    except ValueError:
        return True
    candidates = (address, *_embedded_ipv4_addresses(address))
    return any(
        candidate in _KNOWN_METADATA_ADDRESSES
        or candidate == _IPV4_LIMITED_BROADCAST
        or candidate.is_link_local
        or candidate.is_multicast
        or candidate.is_unspecified
        for candidate in candidates
    )


def _embedded_ipv4_addresses(
    address: ipaddress.IPv4Address | ipaddress.IPv6Address,
) -> tuple[ipaddress.IPv4Address, ...]:
    """Return IPv4 destinations encoded by standard IPv6 transition forms."""

    if isinstance(address, ipaddress.IPv4Address):
        return ()
    candidates: list[ipaddress.IPv4Address] = []
    if address.ipv4_mapped is not None:
        candidates.append(address.ipv4_mapped)
    elif address in _IPV4_COMPATIBLE_NETWORK:
        candidates.append(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
    if address in _NAT64_WELL_KNOWN_NETWORK:
        candidates.append(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
    if address.sixtofour is not None:
        candidates.append(address.sixtofour)
    if address.teredo is not None:
        candidates.extend(address.teredo)
    return tuple(dict.fromkeys(candidates))


def _resolved_address_records(
    hostname: str,
    *,
    port: int,
    label: str,
    policy: OutboundHttpPolicy,
) -> tuple[tuple[int, int, int, str, tuple[object, ...]], ...]:
    host = str(hostname).strip().rstrip(".")
    if not host:
        raise OutboundHttpBlocked(f"{label} must include a hostname")
    if not 1 <= int(port) <= 65535:
        raise OutboundHttpBlocked(f"{label} has an invalid port")
    try:
        results = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
    except socket.gaierror as exc:
        raise OutboundHttpBlocked(f"{label} hostname could not be resolved") from exc
    records = tuple(results)
    if not records:
        raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address")
    addresses = tuple(str(item[4][0]).split("%", 1)[0] for item in records if item[4])
    if not addresses:
        raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address")
    if any(_is_forbidden_special_address(address) for address in addresses):
        raise OutboundHttpBlocked(f"{label} resolves to a forbidden special-purpose network")
    if not policy.allow_private_networks and any(not _is_public_address(address) for address in addresses):
        raise OutboundHttpBlocked(
            f"{label} resolves to a non-public network; "
            "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it"
        )
    return records


def _create_outbound_connection_compat(
    address: tuple[str, int],
    timeout: float | object | None = socket._GLOBAL_DEFAULT_TIMEOUT,  # type: ignore[attr-defined]
    source_address: tuple[str, int] | None = None,
) -> socket.socket:
    return create_outbound_connection(
        address[0],
        address[1],
        timeout=timeout,
        source_address=source_address,
        label="Outbound HTTP connection",
    )


class _OutboundHTTPConnection(http.client.HTTPConnection):
    def __init__(self, *args: object, **kwargs: object) -> None:
        super().__init__(*args, **kwargs)  # type: ignore[arg-type]
        self._create_connection = _create_outbound_connection_compat


class _OutboundHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, *args: object, **kwargs: object) -> None:
        super().__init__(*args, **kwargs)  # type: ignore[arg-type]
        self._create_connection = _create_outbound_connection_compat


class _OutboundHTTPHandler(urllib.request.HTTPHandler):
    def http_open(self, req):  # type: ignore[no-untyped-def]
        return self.do_open(_OutboundHTTPConnection, req)


class _OutboundHTTPSHandler(urllib.request.HTTPSHandler):
    def https_open(self, req):  # type: ignore[no-untyped-def]
        return self.do_open(_OutboundHTTPSConnection, req, context=self._context)
