"""SSRF guard for outbound requests to user-configured URLs.

Connectors and MCP servers point Legate at URLs the user provides, which is the
feature, but on a shared deployment a malicious URL could target internal
services (cloud metadata at 169.254.169.254, localhost admin ports, private
network hosts). :func:`validate_public_url` resolves the host and rejects any URL
that lands on a private, loopback, link-local, or otherwise non-public address.

Trusted single-tenant deployments that legitimately need internal endpoints can
set ``ALLOW_PRIVATE_CONNECTOR_URLS=true`` to bypass the check.
"""

from __future__ import annotations

import ipaddress
import socket
from urllib.parse import urlparse

from legate.config import Settings, get_settings

_BLOCKED_ATTRS = (
    "is_private",
    "is_loopback",
    "is_link_local",
    "is_reserved",
    "is_multicast",
    "is_unspecified",
)


class SSRFError(ValueError):
    """Raised when a URL is not allowed for outbound requests."""


def _is_blocked(ip: str) -> bool:
    addr = ipaddress.ip_address(ip)
    return any(getattr(addr, attr) for attr in _BLOCKED_ATTRS)


def _resolve(host: str) -> list[str]:
    try:
        infos = socket.getaddrinfo(host, None)
    except socket.gaierror as exc:
        raise SSRFError(f"could not resolve host: {host}") from exc
    return [str(info[4][0]) for info in infos]


def validate_public_url(url: str, *, settings: Settings | None = None) -> str:
    """Return the URL if it is safe to fetch, otherwise raise :class:`SSRFError`."""
    settings = settings or get_settings()
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise SSRFError("url must use http or https")
    host = parsed.hostname
    if not host:
        raise SSRFError("url has no host")

    if settings.allow_private_connector_urls:
        return url

    # A literal IP needs no DNS; a hostname is resolved and every address checked
    # so a name that maps to a private range cannot slip through.
    try:
        ipaddress.ip_address(host)
        candidates = [host]
    except ValueError:
        candidates = _resolve(host)

    for ip in candidates:
        if _is_blocked(ip):
            raise SSRFError(f"url resolves to a blocked (non-public) address: {ip}")
    return url
