"""SSRF defense for admin-configurable vendor URLs.

Some integration drivers let the IT admin paste a base URL (Slack
SCIM endpoint, GitHub Enterprise Server host, Jira site host, alert
webhook URL). The admin role is trusted in general — but the URL is
also a payload that travels with the vendor's auth token in the
``Authorization`` header. A misconfigured (or maliciously configured)
URL can:

  • Steal the production token by aiming at attacker.example
  • Probe internal services (``http://127.0.0.1:6379``, ``http://
    169.254.169.254/latest/meta-data/``, internal admin panels)
  • Surface response bodies back into ``test_connection`` /
    ``ProvisioningResult.error_message`` / ``scheduled_jobs.last_
    result_summary`` — turning the admin into an SSRF gateway.

The approach here is *defense-in-depth, not a hard allowlist*:

  1. Scheme MUST be ``https``. Plain HTTP leaks tokens, full stop.
  2. Host MUST NOT resolve to a loopback / private / link-local
     address. This is the SSRF block — even a trusted-domain hostname
     is rejected if it actually points at internal infra.
  3. Per-driver, optional public allowlist (``allowed_suffixes``) for
     "this driver only ever talks to *.slack.com". When the admin
     pastes ``api.attacker.com``, the bind fails at config-save time
     with a clear error rather than at first request.

Credential-bearing HTTP callers use the pinned transports in this module:
the origin hostname remains available to HTTP/TLS for Host and SNI, while
TCP connects only to the public addresses returned by the one validated DNS
lookup. ``trust_env=False`` and cross-origin rejection close the ambient
proxy and redirect escape hatches as well.
"""
from __future__ import annotations

import asyncio
import contextvars
import ipaddress
import socket
import ssl
import threading
import time
from collections.abc import Iterable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Any
from urllib.parse import ParseResult, urlparse

import httpcore
import httpx

from app.core.exceptions import AppException

_PRIVATE_NETWORKS = (
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("169.254.0.0/16"),  # link-local + AWS/GCP IMDS
    ipaddress.ip_network("127.0.0.0/8"),     # loopback
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("100.64.0.0/10"),   # CGNAT
    ipaddress.ip_network("100.100.100.200/32"),  # Aliyun ECS metadata
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),        # IPv6 unique-local
    ipaddress.ip_network("fe80::/10"),       # IPv6 link-local
    ipaddress.ip_network("::ffff:0:0/96"),   # IPv4-mapped IPv6
)

_V4_PRIVATE_NETWORKS = tuple(
    n for n in _PRIVATE_NETWORKS if isinstance(n, ipaddress.IPv4Network)
)

_BLOCKED_HOSTNAMES = {
    "localhost",
    "metadata.google.internal",
    "metadata.goog",
    "metadata.azure.internal",
    "metadata",
    "instance-data",
    "instance-data.ec2.internal",
    # Pre-launch hardening 2026-05-06: defense-in-depth additions
    # spotted by the security audit. The IPv6 metadata addresses
    # below are technically already covered by the fc00::/7 range
    # block, and the "metadata.aws.internal" alias only resolves
    # inside AWS regions, but listing them by name catches a
    # hostname-only block path before DNS resolution.
    "metadata.aws.internal",
    "ip6-localhost",
    "ip6-loopback",
}

# DNS is blocking and operating-system resolvers do not offer reliable
# cancellation. Keep it off both the event loop and asyncio's shared default
# executor: a slow/malicious resolver must not consume every worker used by
# unrelated ``asyncio.to_thread`` jobs. The semaphore prevents the executor's
# otherwise-unbounded work queue from becoming a memory sink after timeouts.
_DNS_RESOLUTION_TIMEOUT_SECONDS = 10.0
_DNS_RESOLUTION_WORKERS = 8
_DNS_RESOLUTION_EXECUTOR = ThreadPoolExecutor(
    max_workers=_DNS_RESOLUTION_WORKERS,
    thread_name_prefix="helix-url-dns",
)
_DNS_RESOLUTION_SLOTS = threading.BoundedSemaphore(_DNS_RESOLUTION_WORKERS)


class UnsafeURLError(AppException):
    def __init__(self, url: str, reason: str) -> None:
        # Round-51 S1-Cred: do NOT embed the full URL in the exception
        # message. Feishu / Slack bot webhook URLs include a bearer-
        # equivalent token in the PATH (e.g.
        # ``https://open.feishu.cn/open-apis/bot/v2/hook/<TOKEN>``).
        # When an admin pastes a typo'd URL that the SSRF guard
        # rejects, the AppException message lands in stdout JSON logs
        # which are shipped to CloudWatch / Datadog. Any operator with
        # log read access could then exfiltrate the bot token. Show
        # only the host (the part the operator actually needs to
        # diagnose) and stash the full URL on a private attribute so
        # local debugging is still possible without log leakage.
        try:
            parsed = urlparse(url)
            hostname = parsed.hostname
            if not hostname:
                host = "<unparseable>"
            else:
                # Never render userinfo. ``netloc`` may contain a username or
                # bearer-equivalent password from an invalid pasted URL.
                try:
                    port = parsed.port
                except ValueError:
                    port = None
                host = f"{hostname}:{port}" if port is not None else hostname
        except Exception:  # noqa: BLE001 — never raise from an exception ctor
            host = "<unparseable>"
        self.host = host
        self.reason = reason
        # Stash but do NOT include in the rendered message. Audit
        # trails / observability layers that need the full URL can
        # read this attribute explicitly.
        self._url_redacted = url
        super().__init__(
            code="UNSAFE_URL",
            message=(
                f"URL with host {host!r} rejected: {reason}. Vendor "
                "base URLs must be public HTTPS hosts; loopback / "
                "private-network / metadata-IP destinations are "
                "blocked to prevent credential exfiltration and SSRF."
            ),
            status_code=400,
        )


@dataclass(frozen=True)
class PinnedURLTarget:
    """An origin plus the public IPs approved for its next HTTP client.

    The hostname remains the HTTP/TLS origin (and therefore the Host header
    and SNI value). Only the TCP destination is replaced by one of
    ``addresses``. Keeping the snapshot immutable prevents a caller from
    accidentally validating one DNS answer and connecting with another.
    """

    host: str
    port: int
    addresses: tuple[str, ...]


def _normalized_url_authority(url: str, parsed: ParseResult) -> tuple[str, int | None]:
    """Return canonical ASCII hostname + validated explicit port.

    ``urlparse().hostname`` correctly excludes userinfo, but it preserves
    Unicode and a trailing root dot. Canonicalize with IDNA before suffix
    matching so allowlists compare DNS names rather than display strings.
    """
    raw_host = parsed.hostname or ""
    if not raw_host:
        raise UnsafeURLError(url, "missing host")
    try:
        host = raw_host.rstrip(".").encode("idna").decode("ascii").lower()
    except UnicodeError as exc:
        raise UnsafeURLError(url, "hostname is not valid IDNA") from exc
    if not host:
        raise UnsafeURLError(url, "missing host")
    try:
        port = parsed.port
    except ValueError as exc:
        raise UnsafeURLError(url, "port is invalid") from exc
    return host, port


def _is_blocked_ip(host: str) -> bool:
    """Return True if ``host`` (a literal IP or a resolvable name) sits
    in a private/loopback/link-local range. Hostnames that fail to
    resolve are treated as blocked — better to fail safe-ish than to
    rely on the operating-system resolver re-trying later under a
    different network view.

    Synchronous version — only call from non-async contexts (tests,
    migrations). Async callers must use ``_is_blocked_ip_async`` to
    avoid blocking the event loop on the ``getaddrinfo`` syscall.
    """
    if host.lower() in _BLOCKED_HOSTNAMES:
        return True
    # If host is a literal IP, check directly. Else resolve.
    try:
        ip = ipaddress.ip_address(host)
        return _ip_is_blocked(ip)
    except ValueError:
        pass
    try:
        infos = socket.getaddrinfo(host, None)
    except (socket.gaierror, OSError):
        return True
    return _resolved_addresses_are_blocked(infos)


def _resolved_addresses_are_blocked(infos: list[tuple]) -> bool:
    """Fail closed unless every resolver candidate is a valid public IP.

    Resolver output is untrusted boundary data. An empty answer or malformed
    sockaddr must not be interpreted as approval: the eventual HTTP client may
    perform another lookup and obtain a destination that was never validated.
    """
    if not infos:
        return True
    for info in infos:
        try:
            addr = info[4][0]
        except (IndexError, KeyError, TypeError):
            return True
        try:
            ip = ipaddress.ip_address(addr.split("%")[0])  # strip zone id
        except (AttributeError, ValueError):
            return True
        if _ip_is_blocked(ip):
            return True
    return False


def _public_addresses(infos: list[tuple]) -> tuple[str, ...]:
    """Return de-duplicated public resolver candidates, or fail closed."""
    if _resolved_addresses_are_blocked(infos):
        return ()
    addresses: list[str] = []
    for info in infos:
        # Safe after the fail-closed structural validation above.
        address = str(info[4][0]).split("%", 1)[0]
        if address not in addresses:
            addresses.append(address)
    return tuple(addresses)


async def _is_blocked_ip_async(host: str) -> bool:
    """Async version of ``_is_blocked_ip``.

    Runs the blocking ``socket.getaddrinfo`` call in a thread pool via
    ``asyncio.to_thread`` so it never blocks the event loop. All async
    callers (integration drivers, CMDB handlers, scheduled job handlers)
    MUST use this variant.
    """
    if host.lower() in _BLOCKED_HOSTNAMES:
        return True
    try:
        ip = ipaddress.ip_address(host)
        return _ip_is_blocked(ip)
    except ValueError:
        pass
    try:
        infos = await asyncio.to_thread(socket.getaddrinfo, host, None)
    except (socket.gaierror, OSError):
        return True
    return _resolved_addresses_are_blocked(infos)


def _ip_is_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
    """True unless ``ip`` is an unambiguously public Internet address.

    For IPv6 addresses that wrap an IPv4 address (``::ffff:x.x.x.x``),
    re-check the unwrapped IPv4 against the v4 ranges so an attacker
    can't bypass the v4 list by writing an IPv4-mapped v6 literal.
    """
    if any(ip in net for net in _PRIVATE_NETWORKS):
        return True
    # An explicit denylist is not exhaustive: unspecified, multicast,
    # documentation, reserved and other special-purpose ranges must never be
    # valid external SaaS destinations.  ``is_global`` is the standard-library
    # source of truth and keeps the sync, async and certificate-probe callers
    # fail-closed as new special ranges are classified.
    # ``ipaddress`` classifies globally-scoped multicast as ``is_global`` on
    # some Python versions. It is still not a valid unicast SaaS destination,
    # so keep the special-purpose predicates explicit as well.
    if ip.is_multicast or ip.is_unspecified or ip.is_reserved or not ip.is_global:
        return True
    if isinstance(ip, ipaddress.IPv6Address):
        mapped = ip.ipv4_mapped
        if mapped is not None and any(mapped in net for net in _V4_PRIVATE_NETWORKS):
            return True
    return False


def _validated_url_origin(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
) -> tuple[str, int]:
    """Apply URL syntax/allowlist policy and return canonical origin."""
    if not url:
        raise UnsafeURLError(url, "empty")
    parsed = urlparse(url)
    if require_https and parsed.scheme != "https":
        raise UnsafeURLError(url, f"scheme {parsed.scheme!r} not https")
    host, explicit_port = _normalized_url_authority(url, parsed)
    # Port restriction. urllib's ``parsed.port`` is None when the URL
    # has no explicit port — that maps to the default scheme port
    # (443 for https, 80 for http) which is fine. Only reject when
    # an explicit non-default port is present and not in the allow set.
    #
    # Pre-launch hardening 2026-05-06: previously the default
    # (``allowed_ports=None``) only kicked in on the require_https
    # path. With ``require_https=False`` AND ``allowed_ports=None``
    # the port check was silently skipped — exactly the path where
    # internal-port SSRF bites hardest. Default to the public-web
    # ports on both branches so hardening-by-default applies
    # uniformly. Callers that genuinely need a non-standard port can
    # opt in by passing ``allowed_ports`` explicitly.
    if allowed_ports is None:
        # Self-hosted destinations (GitHub Enterprise Server, Jira
        # Data Center, SuccessFactors preview tenants) commonly run
        # on :8443 / :9443 / :8000 — when the caller hasn't pinned a
        # public-suffix allowlist, treat the destination as
        # potentially-self-hosted and accept the common HTTPS ports.
        # SaaS callers pin ``allowed_suffixes`` so they get the
        # tighter ``(443,)`` default.
        if allowed_suffixes is None:
            allowed_ports = (443, 8443, 9443)
        else:
            allowed_ports = (443,) if require_https else (80, 443)
    if explicit_port is not None and explicit_port not in allowed_ports:
        raise UnsafeURLError(
            url,
            f"port {explicit_port} not in allowlist {allowed_ports!r}",
        )
    if allowed_suffixes:
        ok = False
        for suffix in allowed_suffixes:
            sfx = suffix.lower().lstrip(".")
            if host == sfx or host.endswith("." + sfx):
                ok = True
                break
        if not ok:
            raise UnsafeURLError(
                url,
                f"host {host!r} not in allowlist {allowed_suffixes!r}",
            )
    default_port = 443 if parsed.scheme == "https" else 80
    return host, explicit_port or default_port


def resolve_safe_external_url(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
) -> PinnedURLTarget:
    """Resolve once, validate every candidate, and return a TCP snapshot."""
    host, port = _validated_url_origin(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )
    if host.lower() in _BLOCKED_HOSTNAMES:
        raise UnsafeURLError(
            url, f"host {host!r} resolves to private/loopback/metadata IP",
        )
    try:
        literal = ipaddress.ip_address(host)
    except ValueError:
        try:
            infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
        except (socket.gaierror, OSError) as exc:
            raise UnsafeURLError(url, f"host {host!r} could not be resolved") from exc
        addresses = _public_addresses(infos)
    else:
        addresses = () if _ip_is_blocked(literal) else (str(literal),)
    if not addresses:
        raise UnsafeURLError(
            url, f"host {host!r} resolves to private/loopback/metadata IP",
        )
    return PinnedURLTarget(host=host, port=port, addresses=addresses)


async def resolve_safe_external_url_async(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
) -> PinnedURLTarget:
    """Resolve in an isolated, bounded pool with a fail-closed deadline.

    ``ThreadPoolExecutor.submit`` deliberately does not copy ``contextvars``
    into the resolver thread. DNS validation has no need for request/auth
    context, and keeping it out prevents accidental context retention by a
    resolver call that outlives a cancelled request.
    """
    # Reject malformed authorities before consuming scarce resolver capacity.
    _validated_url_origin(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )
    if not _DNS_RESOLUTION_SLOTS.acquire(blocking=False):
        raise UnsafeURLError(url, "DNS validation capacity is exhausted")

    try:
        # An explicit empty context keeps this invariant stable even on
        # free-threaded Python builds where new threads may inherit context.
        dns_context = contextvars.Context()
        future = _DNS_RESOLUTION_EXECUTOR.submit(
            dns_context.run,
            resolve_safe_external_url,
            url,
            allowed_suffixes=allowed_suffixes,
            require_https=require_https,
            allowed_ports=allowed_ports,
        )
    except Exception as exc:  # noqa: BLE001 -- executor failure is fail-closed
        _DNS_RESOLUTION_SLOTS.release()
        raise UnsafeURLError(url, "DNS validation worker is unavailable") from exc
    future.add_done_callback(lambda _future: _DNS_RESOLUTION_SLOTS.release())
    try:
        return await asyncio.wait_for(
            asyncio.wrap_future(future),
            timeout=_DNS_RESOLUTION_TIMEOUT_SECONDS,
        )
    except TimeoutError as exc:
        raise UnsafeURLError(url, "DNS validation timed out") from exc


def assert_safe_external_url(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
) -> None:
    """Validate a URL. HTTP callers should use a pinned transport too."""
    resolve_safe_external_url(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )


async def assert_safe_external_url_async(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
) -> None:
    """Async URL validation. HTTP callers should use a pinned transport."""
    await resolve_safe_external_url_async(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )


class _PinnedNetworkBackend(httpcore.NetworkBackend):
    """Connect an HTTP origin only through its approved DNS snapshot."""

    def __init__(
        self,
        target: PinnedURLTarget,
        backend: httpcore.NetworkBackend,
    ) -> None:
        self._target = target
        self._backend = backend

    def connect_tcp(
        self,
        host: str,
        port: int,
        timeout: float | None = None,
        local_address: str | None = None,
        socket_options: Iterable[Any] | None = None,
    ) -> httpcore.NetworkStream:
        _assert_pinned_origin(self._target, host, port)
        deadline = None if timeout is None else time.monotonic() + timeout
        last_error: Exception | None = None
        for address in self._target.addresses:
            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
            try:
                return self._backend.connect_tcp(
                    address,
                    port,
                    timeout=remaining,
                    local_address=local_address,
                    socket_options=socket_options,
                )
            except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc:
                last_error = exc
        if last_error is not None:
            raise last_error
        raise httpcore.ConnectError("safe URL target has no approved address")


class _PinnedAsyncNetworkBackend(httpcore.AsyncNetworkBackend):
    """Async counterpart of :class:`_PinnedNetworkBackend`."""

    def __init__(
        self,
        target: PinnedURLTarget,
        backend: httpcore.AsyncNetworkBackend,
    ) -> None:
        self._target = target
        self._backend = backend

    async def connect_tcp(
        self,
        host: str,
        port: int,
        timeout: float | None = None,
        local_address: str | None = None,
        socket_options: Iterable[Any] | None = None,
    ) -> httpcore.AsyncNetworkStream:
        _assert_pinned_origin(self._target, host, port)
        deadline = None if timeout is None else time.monotonic() + timeout
        last_error: Exception | None = None
        for address in self._target.addresses:
            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
            try:
                return await self._backend.connect_tcp(
                    address,
                    port,
                    timeout=remaining,
                    local_address=local_address,
                    socket_options=socket_options,
                )
            except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc:
                last_error = exc
        if last_error is not None:
            raise last_error
        raise httpcore.ConnectError("safe URL target has no approved address")


def _assert_pinned_origin(target: PinnedURLTarget, host: str, port: int) -> None:
    """Reject redirects or request reuse that escapes the approved origin."""
    try:
        canonical_host = host.rstrip(".").encode("idna").decode("ascii").lower()
    except UnicodeError as exc:
        raise httpcore.ConnectError("HTTP transport requested invalid hostname") from exc
    if canonical_host != target.host or port != target.port:
        raise httpcore.ConnectError(
            "HTTP transport refused destination outside its approved origin"
        )


class PinnedHTTPTransport(httpx.HTTPTransport):
    """HTTPX transport whose TCP connections use one validated DNS answer."""

    def __init__(
        self,
        target: PinnedURLTarget,
        *,
        verify: bool | ssl.SSLContext = True,
    ) -> None:
        super().__init__(verify=verify, trust_env=False)
        backend = self._pool._network_backend  # type: ignore[attr-defined]
        self._pool._network_backend = _PinnedNetworkBackend(  # type: ignore[attr-defined]
            target, backend,
        )


class PinnedAsyncHTTPTransport(httpx.AsyncHTTPTransport):
    """Async HTTPX transport pinned to one validated public origin."""

    def __init__(
        self,
        target: PinnedURLTarget,
        *,
        verify: bool | ssl.SSLContext = True,
    ) -> None:
        super().__init__(verify=verify, trust_env=False)
        backend = self._pool._network_backend  # type: ignore[attr-defined]
        self._pool._network_backend = _PinnedAsyncNetworkBackend(  # type: ignore[attr-defined]
            target, backend,
        )


def safe_http_transport(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
    verify: bool | ssl.SSLContext = True,
) -> PinnedHTTPTransport:
    """Build a sync transport from the same DNS answer that was validated."""
    target = resolve_safe_external_url(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )
    return PinnedHTTPTransport(target, verify=verify)


async def safe_async_http_transport(
    url: str,
    *,
    allowed_suffixes: tuple[str, ...] | None = None,
    require_https: bool = True,
    allowed_ports: tuple[int, ...] | None = None,
    verify: bool | ssl.SSLContext = True,
) -> PinnedAsyncHTTPTransport:
    """Build an async transport from one fail-closed DNS snapshot."""
    target = await resolve_safe_external_url_async(
        url,
        allowed_suffixes=allowed_suffixes,
        require_https=require_https,
        allowed_ports=allowed_ports,
    )
    return PinnedAsyncHTTPTransport(target, verify=verify)


def assert_safe_smtp_host(host: str | None) -> None:
    """Reject SMTP hosts that resolve to private / loopback / metadata IPs.

    Pre-launch hardening 2026-05-06: previously the SMTP driver
    accepted any ``host`` config string with no SSRF check (it's not
    an HTTPS URL so ``assert_safe_external_url`` doesn't apply, and
    the original config-time validation only checked the value was
    non-empty). An admin pasting ``host=169.254.169.254 port=80
    use_tls=false`` would send the SMTP credential AND every email
    body — which carries employee PII like termination dates, asset
    serials, etc. — to the cloud metadata service. This helper plugs
    that gap by reusing the same private-IP block list.
    """
    if not host:
        raise UnsafeURLError(host or "", "SMTP host empty")
    cleaned = host.strip()
    if not cleaned:
        raise UnsafeURLError(host, "SMTP host blank")
    if _is_blocked_ip(cleaned):
        raise UnsafeURLError(
            cleaned,
            f"SMTP host {cleaned!r} resolves to private/loopback/"
            "metadata IP — refusing to send credentials there",
        )