# Trouve-KA — prévention SSRF
# Author: Simon-Pierre Boucher
# Contact: contact@spboucher.ai

"""Garde SSRF du crawler (CLAUDE.md §5.7).

Bloque localhost, plages privées IPv4/IPv6, link-local, métadonnées cloud,
schémas non-HTTP. La résolution DNS est revalidée à chaque saut de redirection
par l'appelant (fetcher).
"""

import ipaddress
import socket
from urllib.parse import urlsplit

BLOCKED_HOSTS = {
    "localhost",
    "metadata.google.internal",
    "metadata.gke.internal",
}

# Endpoint de métadonnées AWS/GCP/Azure/OpenStack
METADATA_IPS = {"169.254.169.254", "fd00:ec2::254"}


def is_safe_ip(ip_str: str) -> bool:
    try:
        ip = ipaddress.ip_address(ip_str)
    except ValueError:
        return False
    if ip_str in METADATA_IPS:
        return False
    return not (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
    )


def resolve_host(host: str) -> list[str]:
    """Résout un hôte en IPs (IPv4+IPv6). Lève socket.gaierror si introuvable."""
    infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
    return list({info[4][0] for info in infos})


def is_safe_url(url: str, *, resolve: bool = True) -> bool:
    """Vérifie qu'une URL est sûre à fetcher (schéma, hôte, IPs résolues)."""
    try:
        parts = urlsplit(url)
    except ValueError:
        return False
    if parts.scheme.lower() not in ("http", "https"):
        return False
    host = parts.hostname
    if not host:
        return False
    host = host.lower().strip(".")
    if host in BLOCKED_HOSTS or host.endswith(".localhost") or host.endswith(".internal"):
        return False
    # Hôte littéral IP
    try:
        ipaddress.ip_address(host)
        return is_safe_ip(host)
    except ValueError:
        pass
    if not resolve:
        return True
    try:
        ips = resolve_host(host)
    except (socket.gaierror, OSError):
        return False
    return bool(ips) and all(is_safe_ip(ip) for ip in ips)