JoinDomainGUI/domainjoin/diagnostics.py

"""Moteur de diagnostic réseau pour une jointure à un domaine Active Directory.

Chaque vérification est indépendante de la plateforme et retourne un CheckResult.
`run_diagnostics()` est un générateur : il yield chaque résultat au fur et à
mesure, pour permettre à l'appelant (thread worker) de mettre à jour l'UI en
continu plutôt que d'attendre la fin de tous les checks.
"""

from __future__ import annotations

import platform
import shutil
import socket
import subprocess
import time
from dataclasses import dataclass, field
from enum import Enum

try:
    import dns.resolver
    import dns.exception

    _HAS_DNSPYTHON = True
except ImportError:
    _HAS_DNSPYTHON = False

try:
    import ntplib

    _HAS_NTPLIB = True
except ImportError:
    _HAS_NTPLIB = False


class Status(Enum):
    OK = "OK"
    WARN = "AVERTISSEMENT"
    FAIL = "ÉCHEC"


@dataclass
class CheckResult:
    name: str
    status: Status
    detail: str


# Ports requis pour rejoindre/fonctionner dans un domaine AD.
AD_PORTS = {
    53: "DNS",
    88: "Kerberos",
    389: "LDAP",
    445: "SMB",
    464: "Kerberos kpasswd",
    636: "LDAPS",
    3268: "LDAP Global Catalog",
}

SRV_RECORDS = [
    "_ldap._tcp.{domain}",
    "_kerberos._tcp.{domain}",
    # C'est ce dernier enregistrement que Windows interroge (DsGetDcName) pour
    # localiser un controleur de domaine lors d'une jointure. Son absence donne
    # « Le domaine specifie n'existe pas ou n'a pas pu etre contacte », alors
    # meme que le domaine resout et que les DC repondent au ping.
    "_ldap._tcp.dc._msdcs.{domain}",
    "_kerberos._tcp.dc._msdcs.{domain}",
]

REQUIRED_LINUX_TOOLS = ["realm", "adcli", "net"]


def _check_hostname() -> CheckResult:
    host = socket.gethostname()
    fqdn = socket.getfqdn()
    if fqdn and fqdn != host and "." in fqdn:
        return CheckResult("Nom d'hôte / FQDN", Status.OK, f"hostname={host}, FQDN={fqdn}")
    return CheckResult(
        "Nom d'hôte / FQDN",
        Status.WARN,
        f"hostname={host}, FQDN non résolu ({fqdn}) — la résolution DNS locale peut être incomplète.",
    )


def _resolve_domain_ips(domain: str) -> list[str]:
    try:
        infos = socket.getaddrinfo(domain, None)
        return sorted({info[4][0] for info in infos})
    except socket.gaierror:
        return []


def _check_dns(domain: str) -> tuple[CheckResult, list[str]]:
    ips = _resolve_domain_ips(domain)
    if ips:
        return CheckResult("Résolution DNS du domaine", Status.OK, f"{domain} → {', '.join(ips)}"), ips
    return (
        CheckResult(
            "Résolution DNS du domaine",
            Status.FAIL,
            f"Impossible de résoudre {domain}. Vérifiez que le DNS du poste pointe vers un contrôleur de domaine.",
        ),
        [],
    )


def _check_srv_records(domain: str) -> CheckResult:
    if not _HAS_DNSPYTHON:
        return CheckResult(
            "Enregistrements SRV (LDAP/Kerberos)",
            Status.WARN,
            "Le module dnspython n'est pas installé — vérification ignorée.",
        )
    found, missing = [], []
    for template in SRV_RECORDS:
        qname = template.format(domain=domain)
        try:
            dns.resolver.resolve(qname, "SRV", lifetime=3.0)
            found.append(qname)
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout, dns.resolver.NoNameservers):
            missing.append(qname)
    if not missing:
        return CheckResult("Enregistrements SRV (LDAP/Kerberos)", Status.OK, f"{len(found)}/{len(found)} trouvés")
    if found:
        return CheckResult(
            "Enregistrements SRV (LDAP/Kerberos)",
            Status.WARN,
            f"Manquants : {', '.join(missing)}",
        )
    return CheckResult(
        "Enregistrements SRV (LDAP/Kerberos)",
        Status.FAIL,
        "Aucun enregistrement SRV trouvé — le DNS ne semble pas être celui du domaine AD.",
    )


def _ping(host: str, timeout_s: int = 2) -> bool:
    is_windows = platform.system() == "Windows"
    if is_windows:
        cmd = ["ping", "-n", "1", "-w", str(timeout_s * 1000), host]
    else:
        cmd = ["ping", "-c", "1", "-W", str(timeout_s), host]
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=timeout_s + 2,
            creationflags=subprocess.CREATE_NO_WINDOW if is_windows else 0,
        )
        return result.returncode == 0
    except (subprocess.TimeoutExpired, OSError):
        return False


def _check_ping(domain: str, ips: list[str]) -> CheckResult:
    targets = ips or [domain]
    reachable = [ip for ip in targets if _ping(ip)]
    if reachable:
        return CheckResult("Ping du/des contrôleur(s) de domaine", Status.OK, f"Répond : {', '.join(reachable)}")
    return CheckResult(
        "Ping du/des contrôleur(s) de domaine",
        Status.WARN,
        "Aucune réponse au ping (peut être normal si ICMP est filtré par un pare-feu).",
    )


def _check_ports(ips: list[str]) -> CheckResult:
    if not ips:
        return CheckResult("Ports AD (53/88/389/445/464/636/3268)", Status.FAIL, "Pas d'IP de contrôleur à tester.")
    ip = ips[0]
    open_ports, closed_ports = [], []
    for port, name in AD_PORTS.items():
        try:
            with socket.create_connection((ip, port), timeout=2):
                open_ports.append(f"{port}/{name}")
        except OSError:
            closed_ports.append(f"{port}/{name}")
    if not closed_ports:
        return CheckResult("Ports AD (53/88/389/445/464/636/3268)", Status.OK, f"Tous ouverts sur {ip}")
    if open_ports:
        return CheckResult(
            "Ports AD (53/88/389/445/464/636/3268)",
            Status.WARN,
            f"Fermés/filtrés sur {ip} : {', '.join(closed_ports)}",
        )
    return CheckResult(
        "Ports AD (53/88/389/445/464/636/3268)",
        Status.FAIL,
        f"Aucun port AD accessible sur {ip} — connectivité réseau ou pare-feu à vérifier.",
    )


def _check_ntp(domain: str, ips: list[str]) -> CheckResult:
    if not _HAS_NTPLIB:
        return CheckResult(
            "Synchronisation horaire (NTP)",
            Status.WARN,
            "Le module ntplib n'est pas installé — vérification ignorée.",
        )
    targets = ips or [domain]
    client = ntplib.NTPClient()
    for target in targets:
        try:
            response = client.request(target, port=123, version=3, timeout=2)
            drift = abs(response.offset)
            if drift > 300:
                return CheckResult(
                    "Synchronisation horaire (NTP)",
                    Status.FAIL,
                    f"Dérive de {drift:.1f}s avec {target} — Kerberos échouera (limite : 300s).",
                )
            if drift > 30:
                return CheckResult(
                    "Synchronisation horaire (NTP)", Status.WARN, f"Dérive de {drift:.1f}s avec {target}."
                )
            return CheckResult("Synchronisation horaire (NTP)", Status.OK, f"Dérive de {drift:.1f}s avec {target}.")
        except Exception:
            continue
    return CheckResult(
        "Synchronisation horaire (NTP)",
        Status.WARN,
        "Aucun contrôleur n'a répondu au NTP (port 123) — vérification impossible.",
    )


def _check_linux_tools() -> CheckResult:
    missing = [tool for tool in REQUIRED_LINUX_TOOLS if shutil.which(tool) is None]
    if not missing:
        return CheckResult("Outils requis (realm/adcli/net)", Status.OK, "Tous les outils sont installés.")
    return CheckResult(
        "Outils requis (realm/adcli/net)",
        Status.FAIL,
        "Manquants : " + ", ".join(missing) + ". Installez-les avec : "
        "sudo apt install realmd sssd adcli samba-common-bin packagekit",
    )


def run_diagnostics(domain: str):
    """Générateur qui exécute tous les checks et yield chaque CheckResult au fur et à mesure."""
    yield _check_hostname()

    dns_result, ips = _check_dns(domain)
    yield dns_result

    yield _check_srv_records(domain)
    yield _check_ping(domain, ips)
    yield _check_ports(ips)
    yield _check_ntp(domain, ips)

    if platform.system() != "Windows":
        yield _check_linux_tools()