"""Active TLS/SSL certificate probing.

Opens a TLS connection to ``host:port`` (SNI = host), reads the leaf
certificate the endpoint actually serves, and parses the fields Helix
cares about for expiry tracking. This is the PRIMARY data source for the
certificate inventory — it covers every PUBLIC https endpoint regardless
of where the cert is hosted (server, load balancer, CDN, cloud cert
manager), which a cloud-cert-manager sync alone cannot.

Design notes
------------
* ``verify_mode = CERT_NONE`` / ``check_hostname = False`` — we MUST be
  able to read a cert that is already expired or self-signed; that's the
  exact condition we're trying to detect. Verification is irrelevant to
  "when does this cert expire".
* TLS 1.0/1.1 are never negotiated. Certificate inventory does not justify
  enabling obsolete wire protocols; an endpoint that only supports them is
  recorded as unreachable and must be remediated at the endpoint.
* The blocking socket + handshake + DER parse runs in a worker thread
  via :func:`asyncio.to_thread` so the probe-sweep cron stays async and
  can fan out with bounded concurrency.
* An unreachable host (refused / timeout / DNS failure) returns
  ``ok=False`` with an ``error`` string — the caller records this as
  ``unreachable`` status, NEVER as ``expired``. Paging an owner that an
  internal-only host's cert "expired" when Helix simply couldn't reach
  it would be a false alarm.
"""
from __future__ import annotations

import asyncio
import ipaddress
import socket
import ssl
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime

from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa
from cryptography.x509.oid import ExtensionOID, NameOID

# Default per-host handshake budget. Kept modest so a sweep over hundreds
# of targets can't stall for minutes on a single black-holed host.
DEFAULT_TIMEOUT = 8.0
DEFAULT_PORT = 443


def _probe_tls_context() -> ssl.SSLContext:
    """Build the deliberately non-verifying but protocol-hardened context."""
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    # We want the served cert even if expired / self-signed / hostname
    # mismatch. Disable hostname checks before CERT_NONE: Python rejects
    # CERT_NONE while check_hostname is still enabled.
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


@dataclass
class ProbedCert:
    """Result of one probe attempt. ``ok`` distinguishes a successful
    read from an unreachable / handshake-failed target."""

    host: str
    port: int
    ok: bool
    error: str | None = None
    common_name: str | None = None
    sans: list[str] = field(default_factory=list)
    issuer: str | None = None
    serial_number: str | None = None
    fingerprint_sha256: str | None = None
    not_before: datetime | None = None
    not_after: datetime | None = None
    key_algorithm: str | None = None
    key_bits: int | None = None
    signature_algorithm: str | None = None


def _fetch_der(host: str, port: int, timeout: float) -> bytes:
    """Blocking: TLS-handshake to host:port and return the leaf cert DER.

    Runs in a thread. Raises on any connection / handshake failure; the
    caller translates the exception into ``ok=False``.
    """
    ctx = _probe_tls_context()
    with _connect_public_socket(host, port, timeout) as sock:
        # The raw socket helper bounds the connect phase; set the timeout
        # explicitly again so the TLS handshake (do_handshake inside
        # wrap_socket) is also bounded — a host that completes TCP connect
        # then stalls mid-handshake otherwise holds the worker thread open.
        sock.settimeout(timeout)
        with ctx.wrap_socket(sock, server_hostname=host) as ssock:
            der = ssock.getpeercert(binary_form=True)
    if not der:
        raise ssl.SSLError("no peer certificate returned")
    return der


def _connect_public_socket(host: str, port: int, timeout: float) -> socket.socket:
    """Resolve once, validate every candidate, then connect by sockaddr.

    Connecting to ``host`` after an SSRF check would ask DNS a second time
    and permit DNS rebinding between validation and use.  This helper pins
    the connection to the exact ``getaddrinfo`` results that were checked.
    If *any* candidate is private/loopback/link-local/metadata, reject the
    whole hostname: silently selecting a public sibling would make safety
    depend on resolver ordering and permit later fallback into the private
    address.
    """
    from app.integrations._url_safety import _BLOCKED_HOSTNAMES, _ip_is_blocked

    if host.rstrip(".").lower() in _BLOCKED_HOSTNAMES:
        raise ssl.SSLError("blocked: private/loopback/metadata hostname")

    try:
        candidates = socket.getaddrinfo(
            host,
            port,
            family=socket.AF_UNSPEC,
            type=socket.SOCK_STREAM,
        )
    except (socket.gaierror, OSError) as exc:
        raise ssl.SSLError("blocked: hostname resolution failed") from exc
    if not candidates:
        raise ssl.SSLError("blocked: hostname returned no addresses")

    validated: list[tuple[int, int, int, tuple]] = []
    seen: set[tuple[int, int, int, tuple]] = set()
    for family, socktype, proto, _canonname, sockaddr in candidates:
        try:
            address = str(sockaddr[0]).split("%", 1)[0]
            ip = ipaddress.ip_address(address)
        except (IndexError, TypeError, ValueError) as exc:
            raise ssl.SSLError("blocked: resolver returned an invalid address") from exc
        # ``ipaddress`` considers multicast globally scoped on some Python
        # versions, but a TLS inventory probe must never target multicast.
        if _ip_is_blocked(ip) or not ip.is_global or ip.is_multicast:
            raise ssl.SSLError("blocked: resolves to a non-public IP")
        candidate = (family, socktype, proto, sockaddr)
        if candidate not in seen:
            seen.add(candidate)
            validated.append(candidate)

    deadline = time.monotonic() + timeout
    last_error: OSError | None = None
    for v_family, v_socktype, v_proto, v_sockaddr in validated:
        sock: socket.socket | None = None
        try:
            sock = socket.socket(v_family, v_socktype, v_proto)
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("certificate probe connect timed out")
            sock.settimeout(remaining)
            # sockaddr came from the single validated getaddrinfo call;
            # passing it directly cannot trigger another DNS lookup.
            sock.connect(v_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 OSError("no usable address returned for certificate probe")


def _name_attr(name: x509.Name, oid) -> str | None:
    try:
        attrs = name.get_attributes_for_oid(oid)
    except Exception:  # noqa: BLE001
        return None
    if not attrs:
        return None
    val = attrs[0].value
    return val if isinstance(val, str) else None


def _key_meta(cert: x509.Certificate) -> tuple[str | None, int | None]:
    pub = cert.public_key()
    if isinstance(pub, rsa.RSAPublicKey):
        return "RSA", pub.key_size
    if isinstance(pub, ec.EllipticCurvePublicKey):
        return f"EC ({pub.curve.name})", pub.key_size
    if isinstance(pub, ed25519.Ed25519PublicKey):
        return "Ed25519", None
    return type(pub).__name__, getattr(pub, "key_size", None)


def _as_utc(dt: datetime | None) -> datetime | None:
    if dt is None:
        return None
    if dt.tzinfo is None:
        return dt.replace(tzinfo=UTC)
    return dt.astimezone(UTC)


def _clip(value: str | None, limit: int) -> str | None:
    """Clamp an untrusted parsed cert field to its DB column width.

    The probe deliberately reads UNTRUSTED leaf certs (CERT_NONE, no
    hostname check). A hostile or misconfigured endpoint can serve a CN /
    issuer / serial longer than the ``cmdb_certificates`` column, which
    would otherwise raise on flush and — combined with the sweep's shared
    session — poison the whole batch. Clamping keeps the row insertable
    (best-effort record) instead of dropping the cert silently.
    """
    if value is None:
        return None
    return value[:limit]


def parse_der(host: str, port: int, der: bytes) -> ProbedCert:
    """Parse a DER-encoded leaf certificate into a :class:`ProbedCert`."""
    cert = x509.load_der_x509_certificate(der)

    common_name = _name_attr(cert.subject, NameOID.COMMON_NAME)
    issuer = (
        _name_attr(cert.issuer, NameOID.COMMON_NAME)
        or _name_attr(cert.issuer, NameOID.ORGANIZATION_NAME)
    )
    sans: list[str] = []
    try:
        ext = cert.extensions.get_extension_for_oid(
            ExtensionOID.SUBJECT_ALTERNATIVE_NAME,
        )
        sans = list(ext.value.get_values_for_type(x509.DNSName))
    except x509.ExtensionNotFound:
        pass
    except Exception:  # noqa: BLE001
        pass

    # cryptography >=42 exposes the tz-aware ``*_utc`` accessors; prefer
    # them so we never store a naive UTC instant.
    not_before = _as_utc(getattr(cert, "not_valid_before_utc", None)
                          or cert.not_valid_before)
    not_after = _as_utc(getattr(cert, "not_valid_after_utc", None)
                        or cert.not_valid_after)

    key_alg, key_bits = _key_meta(cert)
    sig_alg = None
    try:
        sig_alg = cert.signature_algorithm_oid._name  # noqa: SLF001
    except Exception:  # noqa: BLE001
        pass

    return ProbedCert(
        host=host,
        port=port,
        ok=True,
        common_name=_clip(common_name, 256),
        sans=sans,
        issuer=_clip(issuer, 256),
        serial_number=_clip(format(cert.serial_number, "x"), 128),
        fingerprint_sha256=cert.fingerprint(hashes.SHA256()).hex(),
        not_before=not_before,
        not_after=not_after,
        key_algorithm=_clip(key_alg, 32),
        key_bits=key_bits,
        signature_algorithm=_clip(sig_alg, 64),
    )


async def probe_certificate(
    host: str,
    port: int = DEFAULT_PORT,
    timeout: float = DEFAULT_TIMEOUT,
) -> ProbedCert:
    """Probe ``host:port`` and return the parsed leaf certificate.

    Never raises: connection / handshake / parse failures come back as
    ``ProbedCert(ok=False, error=...)``.
    """
    try:
        der = await asyncio.to_thread(_fetch_der, host, port, timeout)
    except (TimeoutError, OSError, ssl.SSLError) as exc:
        return ProbedCert(host=host, port=port, ok=False, error=str(exc) or type(exc).__name__)
    except Exception as exc:  # noqa: BLE001
        return ProbedCert(host=host, port=port, ok=False, error=str(exc) or type(exc).__name__)
    try:
        return await asyncio.to_thread(parse_der, host, port, der)
    except Exception as exc:  # noqa: BLE001
        return ProbedCert(
            host=host, port=port, ok=False,
            error=f"parse failed: {exc}",
        )