"""Unit tests for certificate parsing + pure helpers (no DB)."""
from __future__ import annotations

import socket
import ssl
from datetime import UTC, datetime, timedelta

import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID

from app.domains.cmdb.cert_probe import (
    _connect_public_socket,
    _fetch_der,
    _probe_tls_context,
    parse_der,
)
from app.domains.cmdb.cert_service import (
    EXPIRING_WINDOW_DAYS,
    derive_cert_status,
    fqdn_for,
    probed_dedup_key,
)
from app.domains.cmdb.models import (
    CERT_STATUS_EXPIRED,
    CERT_STATUS_EXPIRING,
    CERT_STATUS_VALID,
)


def _make_cert(
    *,
    cn: str = "example.com",
    sans: list[str] | None = None,
    not_before: datetime | None = None,
    not_after: datetime | None = None,
) -> bytes:
    """Build a self-signed cert and return its DER bytes."""
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    now = datetime.now(UTC)
    nb = not_before or (now - timedelta(days=1))
    na = not_after or (now + timedelta(days=90))
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, cn),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Test Org"),
    ])
    builder = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(nb)
        .not_valid_after(na)
    )
    if sans:
        builder = builder.add_extension(
            x509.SubjectAlternativeName([x509.DNSName(s) for s in sans]),
            critical=False,
        )
    cert = builder.sign(key, hashes.SHA256())
    return cert.public_bytes(serialization.Encoding.DER)


def test_parse_der_extracts_core_fields():
    sans = ["example.com", "www.example.com", "api.example.com"]
    na = datetime.now(UTC) + timedelta(days=42)
    der = _make_cert(cn="example.com", sans=sans, not_after=na)
    parsed = parse_der("example.com", 443, der)

    assert parsed.ok is True
    assert parsed.common_name == "example.com"
    assert set(parsed.sans) == set(sans)
    assert parsed.issuer in ("example.com", "Test Org")
    assert parsed.key_algorithm == "RSA"
    assert parsed.key_bits == 2048
    assert parsed.fingerprint_sha256 and len(parsed.fingerprint_sha256) == 64
    assert parsed.not_after is not None
    # not_after must be tz-aware (never a naive UTC instant).
    assert parsed.not_after.tzinfo is not None
    # within a day of what we asked for
    assert abs((parsed.not_after - na).total_seconds()) < 86400


def test_probe_context_rejects_obsolete_tls_protocols():
    context = _probe_tls_context()

    assert context.minimum_version == ssl.TLSVersion.TLSv1_2
    assert context.maximum_version in (
        ssl.TLSVersion.MAXIMUM_SUPPORTED,
        ssl.TLSVersion.TLSv1_3,
    )
    # The probe intentionally accepts expired/self-signed leaf certs, but
    # that must not silently re-enable TLS 1.0/1.1 on the wire.
    assert context.verify_mode == ssl.CERT_NONE
    assert context.check_hostname is False


class _FakeSocket:
    def __init__(self, family: int, socktype: int, proto: int, *, fail: bool = False):
        self.family = family
        self.socktype = socktype
        self.proto = proto
        self.fail = fail
        self.connected_to = None
        self.closed = False
        self.timeout = None

    def settimeout(self, timeout: float) -> None:
        self.timeout = timeout

    def connect(self, sockaddr) -> None:
        self.connected_to = sockaddr
        if self.fail:
            raise OSError("connect failed")

    def close(self) -> None:
        self.closed = True

    def __enter__(self):
        return self

    def __exit__(self, *_args) -> None:
        self.close()


def _addr(family: int, address: str, port: int = 443):
    sockaddr = (address, port) if family == socket.AF_INET else (address, port, 0, 0)
    return (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", sockaddr)


def test_probe_rejects_mixed_public_private_dns_before_connect(monkeypatch):
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            _addr(socket.AF_INET, "1.1.1.1"),
            _addr(socket.AF_INET, "10.0.0.8"),
        ],
    )
    monkeypatch.setattr(
        socket,
        "socket",
        lambda *_args: pytest.fail("must reject all candidates before connecting"),
    )

    with pytest.raises(ssl.SSLError, match="non-public"):
        _connect_public_socket("mixed.example", 443, 1.0)


@pytest.mark.parametrize(
    "answers",
    [
        [],
        [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ())],
        [_addr(socket.AF_INET, "not-an-ip")],
    ],
)
def test_probe_fails_closed_on_empty_or_invalid_dns_answers(monkeypatch, answers):
    monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: answers)
    monkeypatch.setattr(
        socket,
        "socket",
        lambda *_args: pytest.fail("unvalidated candidates must not be connected"),
    )

    with pytest.raises(ssl.SSLError, match="blocked"):
        _connect_public_socket("invalid.example", 443, 1.0)


def test_probe_resolves_once_connects_pinned_sockaddr_and_preserves_sni(monkeypatch):
    resolutions = 0
    raw_socket = _FakeSocket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)

    def resolve(*_args, **_kwargs):
        nonlocal resolutions
        resolutions += 1
        if resolutions > 1:
            return [_addr(socket.AF_INET, "169.254.169.254")]
        return [_addr(socket.AF_INET, "1.1.1.1")]

    class FakeTlsSocket:
        def __enter__(self):
            return self

        def __exit__(self, *_args) -> None:
            pass

        def getpeercert(self, *, binary_form: bool):
            assert binary_form is True
            return b"leaf-der"

    class FakeContext:
        server_hostname = None

        def wrap_socket(self, sock, *, server_hostname: str):
            assert sock is raw_socket
            self.server_hostname = server_hostname
            return FakeTlsSocket()

    context = FakeContext()
    monkeypatch.setattr(socket, "getaddrinfo", resolve)
    monkeypatch.setattr(socket, "socket", lambda *_args: raw_socket)
    monkeypatch.setattr(
        socket,
        "create_connection",
        lambda *_args, **_kwargs: pytest.fail("must not re-resolve host"),
    )
    monkeypatch.setattr(
        "app.domains.cmdb.cert_probe._probe_tls_context",
        lambda: context,
    )

    assert _fetch_der("cert.example", 443, 1.0) == b"leaf-der"
    assert resolutions == 1
    assert raw_socket.connected_to == ("1.1.1.1", 443)
    assert context.server_hostname == "cert.example"


@pytest.mark.parametrize(
    ("family", "address", "expected_sockaddr"),
    [
        (socket.AF_INET, "1.1.1.1", ("1.1.1.1", 443)),
        (socket.AF_INET6, "2606:4700:4700::1111", ("2606:4700:4700::1111", 443, 0, 0)),
    ],
)
def test_probe_connects_validated_ipv4_and_ipv6(
    monkeypatch, family, address, expected_sockaddr
):
    raw_socket = _FakeSocket(family, socket.SOCK_STREAM, socket.IPPROTO_TCP)
    monkeypatch.setattr(
        socket, "getaddrinfo", lambda *_args, **_kwargs: [_addr(family, address)]
    )
    monkeypatch.setattr(socket, "socket", lambda *_args: raw_socket)

    with _connect_public_socket("cert.example", 443, 1.0) as connected:
        assert connected is raw_socket
        assert connected.connected_to == expected_sockaddr


def test_probe_falls_back_between_validated_addresses(monkeypatch):
    first = _FakeSocket(
        socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, fail=True
    )
    second = _FakeSocket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
    sockets = iter([first, second])
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            _addr(socket.AF_INET6, "2606:4700:4700::1111"),
            _addr(socket.AF_INET, "1.0.0.1"),
        ],
    )
    monkeypatch.setattr(socket, "socket", lambda *_args: next(sockets))

    with _connect_public_socket("cert.example", 443, 1.0) as connected:
        assert connected is second
    assert first.closed is True
    assert second.connected_to == ("1.0.0.1", 443)


def test_probe_rejects_ipv6_link_local_zone_id(monkeypatch):
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [_addr(socket.AF_INET6, "fe80::1%en0")],
    )

    with pytest.raises(ssl.SSLError, match="non-public"):
        _connect_public_socket("cert.example", 443, 1.0)


@pytest.mark.parametrize(
    "address",
    [
        "::",  # IPv6 unspecified
        "ff02::1",  # IPv6 multicast
        "240.0.0.1",  # IPv4 reserved / future-use
    ],
)
def test_probe_rejects_every_non_global_candidate(monkeypatch, address):
    family = socket.AF_INET6 if ":" in address else socket.AF_INET
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [_addr(family, address)],
    )
    monkeypatch.setattr(
        socket,
        "socket",
        lambda *_args: pytest.fail("non-global candidates must not be connected"),
    )

    with pytest.raises(ssl.SSLError, match="non-public"):
        _connect_public_socket("cert.example", 443, 1.0)


def test_probe_falls_back_when_socket_construction_fails(monkeypatch):
    second = _FakeSocket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
    attempts = 0

    def make_socket(*_args):
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            raise OSError("address family unavailable")
        return second

    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            _addr(socket.AF_INET6, "2606:4700:4700::1111"),
            _addr(socket.AF_INET, "1.0.0.1"),
        ],
    )
    monkeypatch.setattr(socket, "socket", make_socket)

    with _connect_public_socket("cert.example", 443, 1.0) as connected:
        assert connected is second
    assert attempts == 2
    assert second.connected_to == ("1.0.0.1", 443)


def test_probe_raises_last_error_when_all_validated_addresses_fail(monkeypatch):
    sockets = iter(
        [
            _FakeSocket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, fail=True),
            _FakeSocket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, fail=True),
        ]
    )
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            _addr(socket.AF_INET, "1.1.1.1"),
            _addr(socket.AF_INET, "1.0.0.1"),
        ],
    )
    monkeypatch.setattr(socket, "socket", lambda *_args: next(sockets))

    with pytest.raises(OSError, match="connect failed"):
        _connect_public_socket("cert.example", 443, 1.0)


def test_parse_der_clips_oversized_untrusted_fields():
    """A hostile/misconfigured endpoint can serve a CN / issuer longer
    than the DB column. parse_der must clamp untrusted string fields to
    their column width so the row stays insertable (the sweep upsert runs
    on a shared session — an over-length flush would otherwise poison the
    whole batch). Verifies common_name(256) / issuer(256)."""
    long_cn = "a" * 400 + ".example.com"
    long_org = "O" * 400
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    now = datetime.now(UTC)
    # _validate=False lets us forge an out-of-spec cert (CN/O > RFC bounds)
    # exactly like a hostile endpoint serving raw DER would — the normal
    # builder enforces 64 bytes, but load_der_x509_certificate does not.
    name = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, long_cn, _validate=False),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, long_org, _validate=False),
    ])
    cert = (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(days=1))
        .not_valid_after(now + timedelta(days=90))
        .sign(key, hashes.SHA256())
    )
    parsed = parse_der("evil.example.com", 443, cert.public_bytes(
        serialization.Encoding.DER,
    ))
    assert parsed.ok is True
    assert parsed.common_name is not None and len(parsed.common_name) <= 256
    assert parsed.issuer is not None and len(parsed.issuer) <= 256
    assert parsed.serial_number is not None and len(parsed.serial_number) <= 128


def test_derive_status_valid_expiring_expired():
    now = datetime(2026, 6, 9, tzinfo=UTC)
    assert derive_cert_status(now + timedelta(days=120), now) == CERT_STATUS_VALID
    assert derive_cert_status(
        now + timedelta(days=EXPIRING_WINDOW_DAYS - 1), now
    ) == CERT_STATUS_EXPIRING
    assert derive_cert_status(now - timedelta(days=1), now) == CERT_STATUS_EXPIRED
    # No expiry → benign (nothing to warn on).
    assert derive_cert_status(None, now) == CERT_STATUS_VALID


def test_derive_status_handles_naive_not_after():
    """A naive datetime (fresh MySQL load strips tzinfo) must not crash."""
    now = datetime(2026, 6, 9, tzinfo=UTC)
    naive = datetime(2026, 12, 1)  # noqa: DTZ001 — intentionally naive
    assert derive_cert_status(naive, now) == CERT_STATUS_VALID


@pytest.mark.parametrize(
    "name,zone,expected",
    [
        ("www", "example.com", "www.example.com"),
        ("@", "example.com", "example.com"),
        ("", "example.com", "example.com"),
        ("example.com", "example.com", "example.com"),
        ("api.example.com", "example.com", "api.example.com"),
        ("www.", "example.com", "www.example.com"),
        ("*", "example.com", None),
        ("*.example.com", "example.com", None),
        ("www", "", None),
    ],
)
def test_fqdn_for(name, zone, expected):
    assert fqdn_for(name, zone) == expected


def test_probed_dedup_key():
    assert probed_dedup_key("a.example.com", 443) == "probed:a.example.com:443"