WGSecure/app/core/wireguard.py

import base64
import ipaddress
import os
import socket
import time
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from app.core import dns as dns_util
from app.core import wg_handshake
from app.core.config import Config
from app.utils.helper_ipc import call_helper
from app.utils.platform_utils import (
    get_config_dir,
    has_root_privileges,
    get_wg_config_dir,
    is_windows,
    run_command,
    run_privileged,
    run_privileged_readonly,
    secure_delete,
    wg_available,
    wg_exe,
    wg_quick_available,
    wireguard_exe,
)


def generate_keypair() -> tuple[str, str]:
    """Retourne (private_key_b64, public_key_b64) au format WireGuard."""
    private = X25519PrivateKey.generate()
    private_bytes = private.private_bytes_raw()
    public_bytes = private.public_key().public_bytes_raw()
    return (
        base64.b64encode(private_bytes).decode(),
        base64.b64encode(public_bytes).decode(),
    )


def generate_preshared_key() -> str:
    return base64.b64encode(os.urandom(32)).decode()


def uses_preshared_key(wg: dict) -> bool:
    """La configuration met-elle une clé pré-partagée en jeu ?

    Le drapeau `use_preshared_key` fait foi. Une configuration importée avant
    son introduction n'a que la clé : une PSK non vide vaut alors activation,
    sinon un import de .conf perdrait sa PSK au premier enregistrement.
    """
    if wg.get("use_preshared_key"):
        return True
    return bool(wg.get("use_preshared_key") is None
                and (wg.get("preshared_key") or "").strip())


def build_client_config(cfg: Config) -> str:
    wg = cfg.wg
    lines = [
        "[Interface]",
        f"PrivateKey = {wg['client_private_key']}",
        f"Address = {wg['client_address']}",
    ]
    # Une directive DNS = sans backend resolvconf/resolvectl fait échouer
    # `wg-quick up` en plein milieu de son exécution : l'interface est déjà
    # créée, le trap de nettoyage laisse un état partiel. Mieux vaut monter le
    # tunnel sans toucher au DNS système.
    
    # Split-DNS : si configuré, on ne met pas de DNS dans la config WireGuard
    # car les règles NRPT (Windows) ou systemd-resolved (Linux) géreront le routage.
    split_dns_server = (wg.get("split_dns_server") or "").strip()
    split_dns_domains = wg.get("split_dns_domains") or []
    
    if split_dns_server and split_dns_domains and is_windows():
        # Sous Windows avec split-DNS : pas de DNS dans la config, tout sera géré par NRPT
        pass
    else:
        # Cas normal : utiliser le DNS principal
        dns_value = (wg.get("dns") or "").strip()
        if dns_value and dns_util.has_dns_backend():
            lines.append(f"DNS = {dns_value}")
    mtu = int(wg.get("mtu") or 0)
    if mtu:
        lines.append(f"MTU = {mtu}")
    lines += [
        "",
        "[Peer]",
        f"PublicKey = {wg['server_public_key']}",
    ]
    # Sans cette ligne, un serveur configuré avec une clé pré-partagée rejette
    # silencieusement toutes nos tentatives : le service monte, l'interface
    # existe, l'application affiche « Connecté » — et le handshake n'aboutit
    # jamais. La PSK était générée par l'assistant puis perdue, ni écrite dans
    # la config client ni relue à l'import d'un .conf.
    #
    # Symétriquement, écrire une PSK que le serveur n'attend pas casse le
    # handshake tout aussi silencieusement : le drapeau tranche, la présence
    # d'un reliquat dans le champ ne suffit pas.
    psk = (wg.get("preshared_key") or "").strip()
    if psk and uses_preshared_key(wg):
        lines.append(f"PresharedKey = {psk}")
    lines += [
        f"AllowedIPs = {wg['allowed_ips']}",
        f"Endpoint = {wg['server_endpoint']}:{wg['server_port']}",
        f"PersistentKeepalive = {wg['keepalive']}",
    ]
    return "\n".join(lines) + "\n"


def get_client_config_path(cfg: Config) -> str:
    name = cfg.wg.get("interface_name", "wgs0")
    if is_windows():
        return os.path.join(get_config_dir(), f"{name}.conf")
    return os.path.join(get_wg_config_dir(), f"{name}.conf")  # /etc/wireguard/


def write_client_config(cfg: Config) -> tuple[bool, str]:
    """Écrit la config client, en élevant les privilèges si nécessaire.

    /etc/wireguard est en 0700 root:root : l'écriture directe échoue pour un
    utilisateur normal. On passe alors par un fichier temporaire dans le
    répertoire de configuration de l'utilisateur, installé ensuite par
    `install` — une seule commande privilégiée, qui pose contenu, propriétaire
    et permissions d'un coup, et qui traverse toute la chaîne
    sudo -n → askpass → pkexec de run_privileged().
    """
    content = build_client_config(cfg)
    path = get_client_config_path(cfg)

    # Chemin direct : Windows, ou exécution en root.
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write(content)
        if not is_windows():
            os.chmod(path, 0o600)
        return True, path
    except (PermissionError, FileNotFoundError, OSError):
        if is_windows():
            return False, f"Écriture impossible dans {path}"

    tmp = os.path.join(get_config_dir(), ".wgs_tmp.conf")
    try:
        # Créer en 0600 dès l'origine : la clé privée ne doit jamais être
        # lisible par les autres utilisateurs, même brièvement.
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(content)

        code, _, err = run_privileged(
            ["install", "-D", "-m", "600", "-o", "root", "-g", "root", tmp, path],
            timeout=60,
        )
        if code == 0:
            return True, path
        return False, err or f"Écriture impossible dans {path}"
    except OSError as e:
        return False, str(e)
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


def _wg_show(name: str) -> str:
    """Sortie de `wg show <iface>`, ou "" si les privilèges manquent.

    `wg show` exige CAP_NET_ADMIN (root) pour lire l'état d'une interface,
    même en lecture seule — un utilisateur ordinaire reçoit « Unable to
    access interface: Operation not permitted ». D'où l'appel via
    `run_privileged_readonly` : sans lui, le handshake et les compteurs de
    trafic restent vides dès que l'application ne tourne plus en root (le cas
    normal une fois `make setup-sudoers` en place pour wg-quick).

    Sous Windows, `wg.exe show` échoue même avec les délégations existantes
    (ACL du service, LimitedOperatorUI) — délègue donc d'abord à
    wgsecure-helper (service SYSTEM), lecture seule, si disponible.
    """
    resp = call_helper({"cmd": "get_status", "name": name})
    if resp is not None:
        return resp.get("message") or "" if resp.get("ok") else ""
    code, out, err = run_privileged_readonly([wg_exe(), "show", name])
    if code != 0 or not out:
        return ""
    if "not permitted" in err.lower() or "unable to access" in err.lower():
        return ""
    return out


def is_connected(cfg: Config) -> bool:
    """True si le tunnel est monté.

    Déterminé via sysfs, qui ne demande aucun privilège : `wg show` en exige,
    et son échec silencieux faisait afficher « Déconnecté » en permanence —
    d'où des tentatives de connexion sur un tunnel déjà monté, une
    auto-reconnexion en boucle, et surtout aucun démontage à la fermeture,
    laissant le DNS du tunnel en place.
    """
    name = cfg.wg.get("interface_name", "wgs0")
    if is_windows():
        code, out, _ = run_command(
            ["powershell", "-NoProfile", "-NonInteractive", "-Command",
             f"(Get-Service -Name 'WireGuardTunnel${name}' -ErrorAction SilentlyContinue).Status -eq 'Running'"],
            timeout=10
        )
        return code == 0 and "True" in out
    return dns_util.interface_is_up(name)


def foreign_wg_interfaces(cfg: Config) -> list[str]:
    """Tunnels WireGuard actifs portant un autre nom que celui configuré.

    Un tunnel monté hors de l'application (ancien profil, `wg-quick` lancé à
    la main) impose lui aussi son DNS : ni l'interface ni ce DNS ne seront
    retirés par WGSecure, qui ne connaît que son propre nom d'interface.
    """
    name = cfg.wg.get("interface_name", "wgs0")
    return [i for i in dns_util.list_wireguard_interfaces() if i != name]


def interface_exists(cfg: Config) -> bool:
    """True si le périphérique réseau existe, même s'il n'est plus fonctionnel."""
    return dns_util.interface_exists(cfg.wg.get("interface_name", "wgs0"))


def _try_helper_tunnel_cmd(cmd: list[str]) -> tuple[int, str, str] | None:
    """Délègue install/uninstalltunnelservice à wgsecure-helper (service
    SYSTEM) s'il est installé — aucune invite d'authentification, y compris
    sans ACL accordée. `None` si le service est absent/indisponible :
    l'appelant bascule alors sur le chemin existant (ACL puis élévation).

    L'ACL du service WireGuard (bouton « Configurer les permissions ») ne
    délègue que démarrer/arrêter un tunnel déjà installé, jamais le créer/
    supprimer — `/installtunnelservice` et `/uninstalltunnelservice` (qui
    font justement l'un ou l'autre à chaque connexion/déconnexion, cf.
    `connect()`/`force_cleanup()`) continuaient donc à réclamer une
    élévation même une fois l'ACL en place. D'où ce service séparé.
    """
    if len(cmd) < 3:
        return None
    verb = cmd[1]
    if verb == "/installtunnelservice":
        config_path = cmd[2]
        name = os.path.splitext(os.path.basename(config_path))[0]
        request = {"cmd": "install_tunnel", "config_path": config_path, "name": name}
    elif verb == "/uninstalltunnelservice":
        request = {"cmd": "uninstall_tunnel", "name": cmd[2]}
    else:
        return None
    resp = call_helper(request)
    if resp is None:
        return None
    return (0 if resp.get("ok") else 1), "", str(resp.get("message") or "")


def _tunnel_service_name(name: str) -> str:
    """Nom du service Windows créé par wireguard.exe pour ce tunnel."""
    return f"WireGuardTunnel${name}"


def tunnel_service_installed(name: str) -> bool:
    """True si le service du tunnel existe, qu'il tourne ou non.

    Distinct de `is_connected()`, qui n'est vrai que s'il tourne : c'est
    cette différence qui permet d'installer le tunnel une seule fois puis de
    se contenter de le démarrer et l'arrêter — au lieu de le recréer et le
    détruire à chaque connexion, ce qui imposait une élévation à chaque fois.
    """
    if not is_windows():
        return False
    code, out, _ = run_command(
        ["powershell", "-NoProfile", "-NonInteractive", "-Command",
         f"if (Get-Service -Name '{_tunnel_service_name(name)}' "
         f"-ErrorAction SilentlyContinue) {{ 'OUI' }}"],
        timeout=10,
    )
    return code == 0 and "OUI" in out


def _control_tunnel_service(name: str, action: str, timeout: int = 45) -> tuple[bool, str]:
    """Démarre ou arrête le service d'un tunnel déjà installé.

    Trois voies, de la moins intrusive à la plus lourde :

    1. `sc.exe` sans élévation — fonctionne si les droits de démarrage ont
       été accordés sur *ce service* (le bouton « Configurer les
       permissions » les pose sur le service WireGuard, qui est un autre
       objet : accorder l'un ne donne rien sur l'autre) ;
    2. le service privilégié wgsecure-helper, s'il est installé ;
    3. l'élévation UAC, en dernier recours.
    """
    service = _tunnel_service_name(name)
    verb = "start" if action == "start" else "stop"

    code, out, err = run_command(["sc.exe", verb, service], timeout=timeout)
    if code == 0:
        return True, ""

    resp = call_helper({"cmd": f"{verb}_tunnel_service", "name": name})
    if resp is not None:
        return bool(resp.get("ok")), str(resp.get("message") or "")

    code, out, err = run_privileged(["sc.exe", verb, service], timeout=timeout)
    if code == 0:
        return True, ""
    return False, err or out or f"Échec de « sc {verb} {service} » (code {code})"


def _run_wireguard_service_cmd(cmd: list[str], timeout: int) -> tuple[int, str, str]:
    """Lance `wireguard.exe /install…` ou `/uninstall…tunnelservice`.

    Un utilisateur dont les ACLs du service ont été accordées (bouton
    « Configurer les permissions » du panneau Admin) peut créer/supprimer ses
    tunnels sans élévation : l'appel non privilégié est donc tenté en premier,
    pour ne jamais déclencher d'UAC inutile dans ce cas.

    Sans ACL accordée, cet appel échoue sur « Accès refusé » — le manifeste
    UAC de wireguard.exe ne s'applique que lancé depuis l'explorateur ou via
    ShellExecute, jamais via CreateProcess (`run_command`) — et échouait donc
    silencieusement plutôt que de demander l'élévation. On retente alors via
    `run_privileged`, seul à passer par ShellExecuteEx "runas" pour obtenir une
    vraie invite UAC.
    """
    resp = _try_helper_tunnel_cmd(cmd)
    if resp is not None:
        return resp
    code, out, err = run_command(cmd, timeout)
    if code == 0:
        return code, out, err
    return run_privileged(cmd, timeout)


def force_cleanup(cfg: Config) -> list[str]:
    """Démonte tout résidu du tunnel : entrée DNS orpheline + interface morte.

    Idempotent et silencieux si rien ne traîne : aucune élévation de privilèges
    n'est demandée tant qu'un résidu n'a pas été constaté.

    Indispensable parce que `wg-quick down` refuse de s'exécuter
    (« is not a WireGuard interface ») dès que l'interface a disparu — et
    n'exécute donc jamais son `unset_dns`, ce qui laisse la résolution DNS
    pointée sur un serveur devenu injoignable.
    """
    name = cfg.wg.get("interface_name", "wgs0")
    actions: list[str] = []

    if is_windows():
        # Le service du tunnel restaure routes et DNS en s'arrêtant.
        if dns_util.interface_exists(name):
            cmd = [wireguard_exe(), "/uninstalltunnelservice", name]
            _run_wireguard_service_cmd(cmd, timeout=60)
        # Nettoyer les règles split-DNS
        dns_util.cleanup_split_dns(cfg)
        actions += dns_util.cleanup(name)
        return actions

    # 1. Supprimer d'abord l'interface : sous systemd-resolved la configuration
    #    DNS est portée par le lien, la retirer suffit à la faire disparaître.
    #    Une seule élévation de privilèges suffit donc dans le cas courant.
    if dns_util.interface_exists(name):
        code, _, _ = run_privileged(["ip", "link", "delete", "dev", name], timeout=20)
        if code == 0:
            actions.append(f"ip link delete {name}")

    # 2. Retirer l'entrée DNS survivante — celle d'openresolv, notamment, qui
    #    ne disparaît pas avec le lien. C'est la cause du « plus d'internet ».
    #    Sans effet, et sans dialogue d'authentification, si rien ne traîne.
    actions += dns_util.cleanup(name)
    
    # Nettoyer les règles split-DNS
    dns_util.cleanup_split_dns(cfg)

    return actions


def _wait_connected(cfg: Config, timeout: int = 15) -> bool:
    """Attend que le service du tunnel passe à RUNNING, ou expire."""
    deadline = time.monotonic() + timeout
    while True:
        if is_connected(cfg):
            return True
        if time.monotonic() >= deadline:
            return False
        time.sleep(0.5)


def _post_connect_message(cfg: Config) -> str:
    """Message de succès de connexion, complété d'un avertissement si le
    split-DNS configuré n'a pas pu être appliqué.

    Son échec était auparavant ignoré silencieusement (`setup_split_dns`
    appelé sans regarder son retour) : le tunnel se déclarait « activé »
    même quand les noms du réseau distant ne se résolvaient jamais.
    """
    ok, msg = dns_util.setup_split_dns(cfg)
    if not ok:
        return f"Tunnel WireGuard activé\n⚠ {msg}"
    return "Tunnel WireGuard activé"


def connect(cfg: Config) -> tuple[bool, str]:
    if not cfg.configured:
        return False, "WireGuard non configuré. Ouvrez le panneau Admin."

    ok, result = write_client_config(cfg)
    if not ok:
        return False, f"Impossible d'écrire la config : {result}"

    name = cfg.wg.get("interface_name", "wgs0")

    if is_windows():
        # Verbes réels de wireguard.exe : /installtunnelservice <chemin .conf>
        # et /uninstalltunnelservice <nom>. Les formes /installtunnel et
        # /uninstalltunnel n'existent pas — wireguard.exe affichait sa boîte
        # d'aide et sortait sans jamais monter le tunnel.
        #
        # Le nom du tunnel est le nom du fichier sans extension : il doit
        # correspondre à interface_name pour que is_connected() retrouve le
        # service WireGuardTunnel$<nom>.
        if is_connected(cfg):
            return True, "Tunnel WireGuard déjà actif"

        # Tunnel déjà installé : il suffit de le démarrer. C'est le cas
        # courant, et le seul qui ne demande aucune élévation — à condition
        # que les droits de démarrage aient été accordés sur ce service.
        # Réinstaller à chaque connexion, comme auparavant, imposait une
        # authentification administrateur à chaque fois : créer ou supprimer
        # un service n'est jamais délégable, le démarrer peut l'être.
        if tunnel_service_installed(name):
            ok, err = _control_tunnel_service(name, "start", timeout=45)
            if ok and _wait_connected(cfg, timeout=15):
                return True, _post_connect_message(cfg)
            if not ok:
                return False, f"Impossible de démarrer le tunnel : {err}"
            return False, "Le tunnel a démarré mais n'est pas monté"

        cmd = [wireguard_exe(), "/installtunnelservice", result]
        try:
            code, _, err = _run_wireguard_service_cmd(cmd, timeout=90)
            # /installtunnelservice enregistre le service et rend la main : le
            # démarrage effectif du tunnel est asynchrone. Tester l'état
            # immédiatement conclurait « échec » sur un tunnel qui monte.
            if code == 0 and _wait_connected(cfg, timeout=15):
                return True, _post_connect_message(cfg)
            force_cleanup(cfg)
            return False, err or f"Échec de l'installation du tunnel (code {code})"
        finally:
            # Ce .conf porte la clé privée en clair, dans un dossier lisible
            # par l'utilisateur (%APPDATA%\WGSecure) : il ne doit pas y
            # subsister une fois consommé. wireguard.exe en a déjà pris sa
            # propre copie, chiffrée par DPAPI, dans son magasin
            # (ProgramData\WireGuard\Configurations) — et
            # /uninstalltunnelservice ne prend que le nom de l'interface,
            # jamais un chemin : plus rien ne le relit ensuite. Il est
            # réécrit à chaque connexion.
            #
            # Sous Linux au contraire, /etc/wireguard/<iface>.conf est
            # indispensable à `wg-quick down` : on n'y touche pas (branche
            # Windows uniquement).
            secure_delete(result)

    # Déjà monté et fonctionnel : rien à faire.
    if is_connected(cfg):
        return True, "Tunnel WireGuard déjà actif"

    # Interface fantôme d'une session précédente : `wg-quick up` mourrait sur
    # « `wgs0' already exists ». On nettoie avant de remonter.
    if dns_util.interface_exists(name):
        force_cleanup(cfg)

    # Passer le nom d'interface (pas le chemin) : AppArmor autorise /etc/wireguard/ seulement
    code, _, err = run_privileged(["wg-quick", "up", name], timeout=90)
    if code == 0:
        return True, _post_connect_message(cfg)

    # Échec : wg-quick a pu s'arrêter après set_dns (ou être tué par le
    # dépassement de délai avant son trap de nettoyage). On démonte
    # explicitement pour ne pas laisser le DNS système détourné.
    force_cleanup(cfg)
    return False, err or "Erreur lors de la connexion"


def disconnect(cfg: Config) -> tuple[bool, str]:
    name = cfg.wg.get("interface_name", "wgs0")

    if is_windows():
        # Arrêter, pas désinstaller : le service du tunnel reste en place
        # pour la prochaine connexion. Le supprimer à chaque déconnexion
        # obligeait à le recréer ensuite, et donc à redemander une
        # authentification administrateur — alors que l'arrêter peut être
        # délégué. Le service restitue routes et DNS en s'arrêtant.
        ok, err = _control_tunnel_service(name, "stop", timeout=45)
        dns_util.cleanup_split_dns(cfg)
        dns_util.cleanup(name)
        if ok or not is_connected(cfg):
            return True, "Tunnel WireGuard désactivé"
        return False, f"Erreur lors de la déconnexion : {err}"

    err = ""
    if is_connected(cfg):
        code, _, err = run_privileged(["wg-quick", "down", name], timeout=60)
    else:
        # wg-quick refuserait de tourner ici ; on passe directement au
        # démontage manuel, sinon l'entrée DNS du tunnel resterait active.
        code = -1

    # Nettoyage inconditionnel : `wg-quick down` peut avoir réussi partiellement
    # (échec d'un PostDown, interface déjà supprimée) et laissé l'entrée DNS.
    force_cleanup(cfg)

    if code == 0 or not dns_util.interface_exists(name):
        return True, "Tunnel WireGuard désactivé"
    return False, err or "Erreur lors de la déconnexion"


def test_connection(cfg: Config, timeout: int = 5) -> tuple[bool, str]:
    """Teste la joignabilité du serveur WireGuard.

    Note : `socket.connect()` sur un socket UDP n'émet aucun paquet et réussit
    donc toujours — l'ancien test renvoyait « joignable » même serveur éteint.
    On résout le nom, on envoie un datagramme et on interprète l'absence de
    réponse ICMP « port unreachable » comme un port ouvert/filtré.
    """
    endpoint = cfg.wg.get("server_endpoint", "")
    port = int(cfg.wg.get("server_port", 51820))
    if not endpoint:
        return False, "Aucun serveur configuré"

    # 1. Résolution du nom — échoue aussi si le DNS système est cassé.
    try:
        infos = socket.getaddrinfo(endpoint, port, type=socket.SOCK_DGRAM)
    except socket.gaierror as e:
        return False, f"Nom « {endpoint} » non résolu : {e.strerror or e}"
    family, socktype, proto, _, addr = infos[0]

    # 2. Datagramme sonde : un ICMP port unreachable remonte en
    #    ConnectionRefusedError, ce qui prouve que rien n'écoute.
    sock = socket.socket(family, socktype, proto)
    sock.settimeout(timeout)
    try:
        start = time.monotonic()
        sock.sendto(b"\x00" * 32, addr)
        try:
            sock.recvfrom(1024)
            latency = int((time.monotonic() - start) * 1000)
            return True, f"Serveur joignable ({latency} ms)"
        except socket.timeout:
            pass
        except ConnectionRefusedError:
            return False, f"Port UDP {port} fermé sur {addr[0]}"
    except OSError as e:
        return False, str(e)
    finally:
        sock.close()

    # 3. Pas de rejet ICMP : le port est ouvert ou filtré. On complète avec un
    #    ping pour distinguer « hôte injoignable » de « serveur silencieux ».
    ms = ping_server(addr[0], timeout=2)
    if ms is None:
        return True, (f"Port UDP {port} ouvert ou filtré sur {addr[0]} "
                      "(hôte ne répond pas au ping)")
    return True, f"Serveur joignable ({ms} ms, port UDP {port} non rejeté)"


# ── Diagnostic de bout en bout ────────────────────────────────────────────

# États d'une étape : ok = validée, warn = suspecte mais non bloquante,
# fail = point de rupture, skip = non évaluable (dépend d'une étape échouée).
_OK, _WARN, _FAIL, _SKIP = "ok", "warn", "fail", "skip"


def _step(state: str, label: str, detail: str = "") -> dict:
    return {"state": state, "label": label, "detail": detail}


def last_handshake_age(name: str) -> int | None:
    """Secondes écoulées depuis le dernier handshake.

    Retourne -1 si aucun handshake n'a jamais eu lieu, None si l'information
    est inaccessible (privilèges insuffisants, `wg` absent).
    """
    resp = call_helper({"cmd": "get_status", "name": name, "latest_handshakes": True})
    if resp is not None:
        out = resp.get("message") or "" if resp.get("ok") else ""
    else:
        code, out, _ = run_privileged_readonly([wg_exe(), "show", name, "latest-handshakes"], 10)
        if code != 0:
            out = ""
    if not out.strip():
        return None
    newest = 0
    for line in out.splitlines():
        parts = line.split()
        if len(parts) >= 2:
            try:
                newest = max(newest, int(parts[-1]))
            except ValueError:
                continue
    if newest == 0:
        return -1
    return max(0, int(time.time()) - newest)


def _networks(value: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
    """Découpe une liste « a/24, b/16 » en réseaux, en ignorant les entrées
    invalides."""
    nets = []
    for part in (value or "").split(","):
        part = part.strip()
        if not part:
            continue
        try:
            nets.append(ipaddress.ip_network(part, strict=False))
        except ValueError:
            continue
    return nets


def _tunnel_gateway(client_address: str, allowed_ips: str = "") -> str:
    """Première IP utilisable du réseau du tunnel (typiquement le serveur).

    Une adresse client en /32 — la forme la plus courante — ne décrit aucun
    réseau : on se replie alors sur le premier réseau des « IPs autorisées »,
    qui contient le pair distant.
    """
    try:
        iface = ipaddress.ip_interface(client_address.split(",")[0].strip())
    except ValueError:
        return ""

    candidates = []
    if iface.network.prefixlen < iface.network.max_prefixlen - 1:
        candidates.append(iface.network)
    candidates += [n for n in _networks(allowed_ips)
                   if n.prefixlen < n.max_prefixlen - 1
                   and n.version == iface.ip.version]

    for net in candidates:
        for host in net.hosts():
            if host != iface.ip:
                return str(host)
    return ""


def _local_route_source(ip: str, port: int) -> str:
    """IP source que le système emploierait pour joindre `ip`.

    Aucun paquet n'est émis : `connect()` sur un socket UDP ne fait que fixer
    la route. Sert à détecter le cas où l'endpoint du serveur est lui-même
    routé dans le tunnel — le tunnel a alors besoin de lui-même pour se
    maintenir et le handshake s'arrête net après la connexion.
    """
    try:
        family = socket.AF_INET6 if ":" in ip else socket.AF_INET
        with socket.socket(family, socket.SOCK_DGRAM) as sock:
            sock.connect((ip, port or 51820))
            return sock.getsockname()[0]
    except OSError:
        return ""


def _next_hop(ip: str) -> str:
    """Passerelle réellement empruntée pour joindre `ip`, "" si lien direct.

    Une plage « IPs autorisées » qui contient cette passerelle est fatale :
    `wg-quick` pose sa route avec une métrique de 0, elle bat la route de lien
    de l'interface physique, la passerelle devient joignable uniquement par le
    tunnel — et les paquets de handshake, qui doivent sortir *par* elle, ne
    quittent plus jamais la machine.
    """
    if is_windows():
        code, out, _ = run_command(
            ["powershell", "-NoProfile", "-Command",
             f"(Find-NetRoute -RemoteIPAddress {ip}).NextHop"], 10)
        if code != 0:
            return ""
        for line in out.splitlines():
            hop = line.strip()
            if hop and hop not in ("0.0.0.0", "::"):
                return hop
        return ""

    code, out, _ = run_command(["ip", "route", "get", ip], 10)
    if code != 0:
        return ""
    parts = out.split()
    if "via" in parts:
        return parts[parts.index("via") + 1]
    return ""


def _diag_config(cfg: Config) -> list[dict]:
    wg = cfg.wg
    steps: list[dict] = []

    missing = [
        label for label, key in (
            ("adresse du serveur", "server_endpoint"),
            ("clé publique du serveur", "server_public_key"),
            ("clé privée du client", "client_private_key"),
            ("adresse IP du client", "client_address"),
        ) if not (wg.get(key) or "").strip()
    ]
    if missing:
        steps.append(_step(_FAIL, "Configuration",
                           "Champs manquants : " + ", ".join(missing)))
        return steps
    steps.append(_step(_OK, "Configuration", "Tous les champs requis sont remplis"))

    # Clé publique client cohérente avec la clé privée : une clé publique
    # obsolète est déclarée au serveur et le handshake est refusé.
    try:
        priv = X25519PrivateKey.from_private_bytes(
            base64.b64decode(wg["client_private_key"])
        )
        derived = base64.b64encode(
            priv.public_key().public_bytes_raw()
        ).decode()
    except Exception:
        steps.append(_step(_FAIL, "Clé privée du client",
                           "Clé illisible : ce n'est pas une clé X25519 base64"))
        return steps
    declared = (wg.get("client_public_key") or "").strip()
    if declared and declared != derived:
        steps.append(_step(
            _FAIL, "Paire de clés du client",
            f"La clé publique enregistrée ne correspond pas à la clé privée. "
            f"Le serveur doit connaître : {derived}"))
    else:
        steps.append(_step(_OK, "Paire de clés du client",
                           f"Clé publique déclarée au serveur : {derived[:20]}…"))

    # L'ancien message avertissait sur toute PSK vide : sur un serveur qui
    # n'en utilise pas — le cas par défaut de WireGuard — le diagnostic
    # pointait un coupable inexistant à chaque exécution.
    psk = (wg.get("preshared_key") or "").strip()
    if not uses_preshared_key(wg):
        if psk:
            steps.append(_step(
                _WARN, "Clé pré-partagée",
                "Désactivée, mais une clé subsiste dans le champ : elle n'est "
                "pas écrite dans la configuration. Cochez l'option si le "
                "serveur l'attend."))
        else:
            steps.append(_step(
                _OK, "Clé pré-partagée",
                "Non utilisée — le serveur ne doit pas en déclarer une pour "
                "ce pair"))
    elif not psk:
        steps.append(_step(
            _FAIL, "Clé pré-partagée",
            "Option activée mais champ vide : le pair déclare une PSK au "
            "serveur sans en fournir. Renseignez-la, ou décochez l'option si "
            "le serveur n'en utilise pas."))
    else:
        try:
            valid = len(base64.b64decode(psk)) == 32
        except Exception:
            valid = False
        steps.append(_step(_OK if valid else _FAIL, "Clé pré-partagée",
                           "Activée (doit être identique côté serveur)" if valid
                           else "Format invalide : 32 octets en base64 attendus"))

    # En tunnel scindé (AllowedIPs ≠ 0.0.0.0/0), NetworkManager/systemd-resolved
    # associe le DNS configuré à l'interface WireGuard avec un domaine de
    # routage global (`~.`) : toute résolution de nom est alors tentée via
    # cette interface, qui n'a de route que vers les réseaux listés dans les
    # IPs autorisées. Un DNS hors de cette liste (ex. 1.1.1.1 sur un tunnel qui
    # ne route que des sous-réseaux internes) devient injoignable par ce
    # chemin — la résolution échoue en bloc alors que le reste d'Internet,
    # non concerné par ce DNS, continue de fonctionner (d'où un `ping` sur une
    # IP qui réussit à côté d'un `ping` sur un nom qui échoue).
    dns_value = (wg.get("dns") or "").strip()
    if dns_value and dns_util.has_dns_backend():
        nets = _networks(wg.get("allowed_ips", ""))
        full_tunnel = any(n.prefixlen == 0 for n in nets)
        if nets and not full_tunnel:
            uncovered = []
            for part in dns_value.split(","):
                part = part.strip()
                if not part:
                    continue
                try:
                    addr = ipaddress.ip_address(part)
                except ValueError:
                    continue
                if not any(addr in n for n in nets if n.version == addr.version):
                    uncovered.append(part)
            if uncovered:
                steps.append(_step(
                    _WARN, "Serveur DNS routé par le tunnel",
                    "DNS " + ", ".join(uncovered) + " absent des IPs "
                    "autorisées : une fois le tunnel monté, le système tente "
                    "de le joindre via l'interface WireGuard, qui n'a pourtant "
                    "aucune route vers cette adresse — toute résolution de "
                    "nom échoue, même si le reste d'Internet fonctionne. "
                    "Utilisez un DNS compris dans les IPs autorisées (ex. un "
                    "résolveur interne au réseau distant), ou ajoutez son "
                    "réseau à la liste."))
    return steps


def _diag_reachability(cfg: Config, timeout: int) -> tuple[list[dict], str]:
    """Étapes réseau jusqu'au serveur. Retourne (étapes, ip_résolue)."""
    wg = cfg.wg
    endpoint = (wg.get("server_endpoint") or "").strip()
    port = int(wg.get("server_port", 51820))
    steps: list[dict] = []

    try:
        infos = socket.getaddrinfo(endpoint, port, type=socket.SOCK_DGRAM)
    except socket.gaierror as e:
        steps.append(_step(_FAIL, "Résolution du nom du serveur",
                           f"« {endpoint} » non résolu : {e.strerror or e}"))
        return steps, ""
    family, socktype, proto, _, addr = infos[0]
    server_ip = addr[0]
    steps.append(_step(_OK, "Résolution du nom du serveur",
                       f"{endpoint} → {server_ip}"))

    # Route empruntée : si elle sort par l'adresse du tunnel, le trafic du
    # tunnel repasse par le tunnel et rien ne circule plus.
    src = _local_route_source(server_ip, port)
    client_ip = (wg.get("client_address") or "").split("/")[0].strip()
    if not src:
        steps.append(_step(_FAIL, "Route vers le serveur",
                           f"Aucune route vers {server_ip}"))
    elif src == client_ip:
        steps.append(_step(
            _FAIL, "Route vers le serveur",
            f"Le serveur ({server_ip}) est routé dans le tunnel lui-même "
            f"(source {src}). Excluez l'IP du serveur des « IPs autorisées »."))
    else:
        steps.append(_step(_OK, "Route vers le serveur",
                           f"Sortie par {src}"))

    allowed = _networks(wg.get("allowed_ips", ""))

    # L'endpoint capturé par AllowedIPs : le tunnel se coupe l'herbe sous le
    # pied dès qu'il monte, et le handshake suivant n'arrive jamais.
    try:
        server_addr = ipaddress.ip_address(server_ip)
    except ValueError:
        server_addr = None
    if server_addr and any(server_addr in n for n in allowed):
        steps.append(_step(
            _FAIL, "Cohérence des « IPs autorisées »",
            f"{server_ip} (le serveur) est inclus dans les IPs autorisées : "
            f"une fois monté, le tunnel s'auto-route et le handshake cesse. "
            f"Retirez cette plage ou excluez {server_ip}/32."))
    elif src and allowed:
        try:
            local_addr = ipaddress.ip_address(src)
        except ValueError:
            local_addr = None
        clash = [str(n) for n in allowed if local_addr and local_addr in n]

        # Cas mortel du lot : la passerelle par défaut tombe dans une plage
        # autorisée. `wg-quick` route la plage vers le tunnel avec une
        # métrique de 0, ce qui bat la route de lien de la carte physique ; la
        # passerelle n'est plus joignable que *par* le tunnel, alors que le
        # tunnel a besoin d'elle pour sortir. Le tunnel monte, l'interface
        # existe, et plus un seul paquet de handshake ne quitte la machine.
        hop = _next_hop(server_ip)
        try:
            hop_addr = ipaddress.ip_address(hop) if hop else None
        except ValueError:
            hop_addr = None
        hop_clash = [str(n) for n in allowed if hop_addr and hop_addr in n]
        if hop_clash:
            steps.append(_step(
                _FAIL, "Cohérence des « IPs autorisées »",
                f"La passerelle qui mène au serveur ({hop}) est incluse dans "
                f"{', '.join(hop_clash)} : dès que le tunnel monte, cette "
                f"plage est routée vers lui, la passerelle devient injoignable "
                f"et le handshake n'arrive jamais. Retirez "
                f"{', '.join(hop_clash)} des IPs autorisées."))
        elif clash:
            steps.append(_step(
                _WARN, "Cohérence des « IPs autorisées »",
                f"Votre réseau local ({src}) est couvert par {', '.join(clash)} : "
                f"une fois le tunnel monté, imprimante, NAS et box locaux "
                f"passeront par le VPN"))
        else:
            steps.append(_step(_OK, "Cohérence des « IPs autorisées »",
                               f"{len(allowed)} plage(s), sans conflit avec le "
                               f"réseau local"))

    ms = ping_server(server_ip, timeout=2)
    if ms is None:
        steps.append(_step(_WARN, "Ping du serveur (ICMP)",
                           "Pas de réponse — normal si l'ICMP est filtré"))
    else:
        steps.append(_step(_OK, "Ping du serveur (ICMP)", f"{ms} ms"))

    sock = socket.socket(family, socktype, proto)
    sock.settimeout(timeout)
    try:
        sock.sendto(b"\x00" * 32, addr)
        try:
            sock.recvfrom(1024)
            steps.append(_step(_OK, f"Port UDP {port}", "Réponse reçue"))
        except socket.timeout:
            steps.append(_step(
                _OK if ms is not None else _WARN, f"Port UDP {port}",
                "Ouvert ou filtré : aucun rejet ICMP (un serveur WireGuard "
                "ignore silencieusement les paquets non authentifiés, c'est "
                "le résultat attendu)"))
        except ConnectionRefusedError:
            steps.append(_step(_FAIL, f"Port UDP {port}",
                               f"Fermé sur {server_ip} : rien n'écoute, ou un "
                               f"pare-feu rejette le trafic"))
    except OSError as e:
        steps.append(_step(_FAIL, f"Port UDP {port}", str(e)))
    finally:
        sock.close()

    return steps, server_ip


def _diag_handshake(cfg: Config, timeout: float = 5.0) -> list[dict]:
    """Négocie un vrai handshake, sans monter le tunnel.

    Sans cette étape, le diagnostic s'arrêtait sur « le port ne rejette rien »
    puis, tunnel non monté, concluait « chaîne validée — reste à le monter ».
    Un serveur WireGuard étant silencieux avec les paquets non authentifiés,
    ce verdict tombait aussi bien sur une configuration saine que sur un port
    non redirigé ou une clé inconnue du serveur. C'est le seul test qui
    distingue les deux.
    """
    wg = cfg.wg
    issue, detail = wg_handshake.probe_handshake(
        server_endpoint=(wg.get("server_endpoint") or "").strip(),
        server_port=int(wg.get("server_port", 51820)),
        server_public_key=(wg.get("server_public_key") or "").strip(),
        client_private_key=(wg.get("client_private_key") or "").strip(),
        preshared_key=((wg.get("preshared_key") or "").strip()
                       if uses_preshared_key(wg) else ""),
        timeout=timeout,
    )
    state = {
        wg_handshake.OK_NO_PSK:    _OK,
        wg_handshake.OK_PSK:       _OK,
        wg_handshake.COOKIE:       _WARN,
        wg_handshake.SRC_PORT_FILTERED: _FAIL,
        wg_handshake.PSK_REQUIRED: _FAIL,
        wg_handshake.NO_REPLY:     _FAIL,
        wg_handshake.BAD_CONFIG:   _FAIL,
        wg_handshake.NET_ERROR:    _FAIL,
    }.get(issue, _WARN)
    return [_step(state, "Négociation du handshake", detail)]


def _diag_tunnel(cfg: Config) -> list[dict]:
    """Étapes portant sur le tunnel lui-même : montage, handshake, trafic."""
    wg = cfg.wg
    name = wg.get("interface_name", "wgs0")
    steps: list[dict] = []

    if not is_connected(cfg):
        steps.append(_step(_SKIP, "Tunnel monté",
                           "Tunnel inactif — connectez-vous pour tester la "
                           "suite de la chaîne"))
        return steps
    steps.append(_step(_OK, "Tunnel monté",
                       f"Service actif sur l'interface {name}"))

    age = last_handshake_age(name)
    if age is None:
        steps.append(_step(
            _WARN, "Handshake",
            "État inaccessible : `wg show` exige des droits "
            + ("administrateur" if is_windows() else "root")))
        return steps
    if age < 0:
        if uses_preshared_key(wg):
            psk_cause = ("clé pré-partagée différente de celle du serveur (ou "
                         "absente côté serveur)")
        else:
            psk_cause = ("clé pré-partagée attendue par le serveur alors que "
                         "l'option est décochée ici")
        steps.append(_step(
            _FAIL, "Handshake",
            "Jamais négocié. Le tunnel est monté mais le serveur ne répond "
            "pas à nos paquets. Causes usuelles : clé publique du client "
            f"absente côté serveur, clé publique du serveur erronée, "
            f"{psk_cause}, port UDP du serveur différent de celui configuré, "
            "port UDP bloqué en chemin."))
        return steps
    if age > 180:
        steps.append(_step(_WARN, "Handshake",
                           f"Dernier échange il y a {age} s — le pair ne "
                           f"répond plus (keepalive recommandé : 25 s)"))
    else:
        steps.append(_step(_OK, "Handshake", f"Négocié il y a {age} s"))

    info = get_status_info(cfg)
    rx, tx = info.get("rx_bytes"), info.get("tx_bytes")
    if rx in (0, "0", "", None) and tx in (0, "0", "", None):
        steps.append(_step(_WARN, "Trafic chiffré", "Aucun octet échangé"))
    else:
        steps.append(_step(_OK, "Trafic chiffré", f"↓ {rx}   ↑ {tx}"))

    gateway = _tunnel_gateway(wg.get("client_address", ""),
                              wg.get("allowed_ips", ""))
    if not gateway:
        steps.append(_step(_SKIP, "Pair joignable dans le tunnel",
                           "Adresse du tunnel non exploitable"))
    else:
        ms = ping_server(gateway, timeout=2)
        if ms is None:
            steps.append(_step(
                _WARN, "Pair joignable dans le tunnel",
                f"{gateway} ne répond pas — ICMP filtré côté serveur, ou "
                f"« IPs autorisées » ne couvre pas ce réseau"))
        else:
            steps.append(_step(_OK, "Pair joignable dans le tunnel",
                               f"{gateway} répond en {ms} ms"))

    if not dns_util.resolution_works():
        steps.append(_step(_FAIL, "Résolution DNS dans le tunnel",
                           "Plus aucun nom ne se résout — utilisez "
                           "« Réparer le DNS »"))
        return steps
    steps.append(_step(_OK, "Résolution DNS dans le tunnel", "Fonctionnelle"))

    # Un résolveur qui répond ne dit rien sur le split-DNS lui-même : le nom
    # peut très bien se résoudre... vers une réponse publique inattendue au
    # lieu du réseau distant. On vérifie donc explicitement le(s) domaine(s)
    # configurés, pas seulement qu'une résolution quelconque fonctionne.
    domains = wg.get("split_dns_domains") or []
    for domain in domains:
        ok, ips = dns_util.resolve_host(domain, timeout=3.0)
        if ok:
            steps.append(_step(_OK, f"Résolution distante ({domain})",
                               "→ " + ", ".join(ips)))
        else:
            steps.append(_step(
                _WARN, f"Résolution distante ({domain})",
                "Ne se résout pas — vérifiez le serveur DNS distant "
                f"({wg.get('split_dns_server', '?')}) et les règles NRPT/"
                "systemd-resolved du split-DNS"))

    return steps


def diagnose_connection(cfg: Config, timeout: int = 5) -> list[dict]:
    """Vérifie toute la chaîne de connexion et désigne le point de rupture.

    Le test précédent s'arrêtait à la joignabilité UDP du serveur : il
    concluait « serveur joignable » sur un tunnel qui n'échangeait rien, sans
    jamais dire où ça coinçait. Les étapes vont ici de la configuration locale
    jusqu'à la résolution DNS à l'intérieur du tunnel.

    Retourne une liste de {"state", "label", "detail"}, dans l'ordre de la
    chaîne. Une étape `fail` arrête la progression : les suivantes seraient
    ininterprétables.
    """
    steps: list[dict] = []

    if not wg_quick_available():
        steps.append(_step(
            _FAIL, "WireGuard installé",
            "wireguard.exe introuvable — installez WireGuard for Windows"
            if is_windows() else
            "wg-quick introuvable — installez le paquet wireguard-tools"))
        return steps
    if not wg_available():
        steps.append(_step(_WARN, "Outils WireGuard",
                           "`wg` introuvable : l'état du handshake ne pourra "
                           "pas être lu"))
    else:
        steps.append(_step(_OK, "WireGuard installé", "Binaires présents"))

    if not is_windows() and not has_root_privileges():
        steps.append(_step(_WARN, "Privilèges",
                           "Exécution sans root : certains détails du tunnel "
                           "resteront masqués"))

    steps += _diag_config(cfg)
    if any(s["state"] == _FAIL for s in steps):
        return steps

    net_steps, _ = _diag_reachability(cfg, timeout)
    steps += net_steps
    if any(s["state"] == _FAIL for s in net_steps):
        return steps

    hs_steps = _diag_handshake(cfg, timeout)
    steps += hs_steps
    if any(s["state"] == _FAIL for s in hs_steps):
        return steps

    steps += _diag_tunnel(cfg)
    return steps


def parse_conf_file(path: str) -> dict | None:
    """Parse un fichier .conf WireGuard et retourne les valeurs extraites."""
    result: dict = {}
    section = None
    try:
        with open(path, "r", encoding="utf-8") as f:
            for raw in f:
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                if line.startswith("["):
                    section = line.strip("[]").lower()
                    continue
                if "=" not in line:
                    continue
                key, val = (x.strip() for x in line.split("=", 1))
                if section == "interface":
                    if key == "PrivateKey":
                        result["client_private_key"] = val
                    elif key == "Address":
                        result["client_address"] = val.split(",")[0].strip()
                    elif key == "DNS":
                        result["dns"] = val.split(",")[0].strip()
                    elif key == "MTU":
                        try:
                            result["mtu"] = int(val)
                        except ValueError:
                            pass
                elif section == "peer":
                    if key == "PublicKey":
                        result["server_public_key"] = val
                    elif key == "PresharedKey":
                        # Ignorée jusqu'ici : le .conf importé perdait sa PSK
                        # et le tunnel ne négociait plus rien. La présence de
                        # la directive vaut activation de l'option.
                        result["preshared_key"] = val
                        result["use_preshared_key"] = True
                    elif key == "Endpoint" and ":" in val:
                        host, port = val.rsplit(":", 1)
                        result["server_endpoint"] = host.strip("[]")
                        try:
                            result["server_port"] = int(port)
                        except ValueError:
                            pass
                    elif key == "AllowedIPs":
                        result["allowed_ips"] = val
                    elif key == "PersistentKeepalive":
                        try:
                            result["keepalive"] = int(val)
                        except ValueError:
                            pass
        # Dériver la clé publique depuis la clé privée importée
        if "client_private_key" in result and "client_public_key" not in result:
            try:
                priv_bytes = base64.b64decode(result["client_private_key"])
                priv = X25519PrivateKey.from_private_bytes(priv_bytes)
                pub_bytes = priv.public_key().public_bytes_raw()
                result["client_public_key"] = base64.b64encode(pub_bytes).decode()
            except Exception:
                pass
        return result or None
    except Exception:
        return None


def export_conf_file(cfg: Config, dest_path: str) -> tuple[bool, str]:
    """Exporte la configuration courante vers un fichier .conf."""
    try:
        content = build_client_config(cfg)
        with open(dest_path, "w") as f:
            f.write(content)
        if not is_windows():
            os.chmod(dest_path, 0o600)
        return True, dest_path
    except Exception as e:
        return False, str(e)


def get_interface_bytes(iface: str) -> tuple[int, int] | None:
    """Retourne (rx_bytes, tx_bytes) depuis /proc/net/dev (Linux).

    Le nom d'interface est comparé exactement : un `in` sur la ligne entière
    confondait « wg0 » avec « wg0bis » et pouvait aussi matcher un compteur.
    Les colonnes sont lues après le « : », car sur un compteur très large
    /proc/net/dev colle le nom et la valeur (« wgs0:123456789 »).
    """
    try:
        with open("/proc/net/dev") as f:
            for line in f:
                if ":" not in line:
                    continue
                name, _, counters = line.partition(":")
                if name.strip() != iface:
                    continue
                parts = counters.split()
                if len(parts) < 9:
                    return None
                return int(parts[0]), int(parts[8])
    except (OSError, ValueError):
        pass
    return None


def ping_server(host: str, timeout: int = 2) -> int | None:
    """Ping ICMP du serveur. Retourne la latence en ms, ou None si inaccessible."""
    if not host:
        return None
    if is_windows():
        code, out, _ = run_command(
            ["ping", "-n", "1", "-w", str(timeout * 1000), host], timeout + 2
        )
    else:
        code, out, _ = run_command(
            ["ping", "-c", "1", "-W", str(timeout), host], timeout + 2
        )
    if code != 0:
        return None
    # Parser "time=X.X ms" ou "temps=X.X ms"
    import re
    # `ping` est traduit : « temps=12ms » sur un Windows français, « tempo »
    # sur un italien. Ne matcher que « time » faisait conclure « hôte
    # injoignable » alors que la commande avait réussi (code 0).
    m = re.search(r"(?:time|temps|tempo|Zeit|tiempo)[<=]\s*([\d.,]+)\s*m?s",
                  out, re.IGNORECASE)
    if m:
        try:
            return int(float(m.group(1).replace(",", ".")))
        except ValueError:
            pass
    # Réponse reçue mais latence non analysable : 0 ms plutôt que « injoignable ».
    return 0


#: Au-delà de cet âge (s), un handshake est considéré trop ancien pour
#: garantir que le tunnel répond encore (keepalive par défaut : 25 s).
_HANDSHAKE_STALE_S = 150


def ping_or_handshake(cfg: Config, timeout: int = 2) -> tuple[int | None, str]:
    """Latence ICMP vers l'endpoint, avec repli sur l'âge du handshake.

    De nombreux serveurs (ou un pare-feu en chemin) bloquent l'ICMP echo
    tout en laissant le tunnel WireGuard fonctionner parfaitement : sans ce
    repli, le badge de statut affichait « hors ligne » en continu sur ces
    configurations, alors que le handshake était récent et le trafic RX/TX
    bien réel. Retourne (valeur, source) où `source` vaut "ping" (latence
    ICMP), "handshake" (âge du dernier handshake, tunnel sain), "stale"
    (handshake trop ancien) ou "down" (aucune des deux sondes n'aboutit).
    """
    host = cfg.wg.get("server_endpoint", "")
    ms = ping_server(host, timeout) if host else None
    if ms is not None:
        return ms, "ping"
    age = last_handshake_age(cfg.wg.get("interface_name", "wgs0"))
    if age is not None and 0 <= age <= _HANDSHAKE_STALE_S:
        return age, "handshake"
    if age is not None and age > _HANDSHAKE_STALE_S:
        return age, "stale"
    return None, "down"


# ── Test DNS leak ─────────────────────────────────────────────────────────

def dns_leak_test(cfg: Config) -> dict:
    """
    Teste si le DNS fuit en dehors du tunnel.
    Retourne {"status": ok|leak|unknown|down, "resolvers": [...], "expected": str}
    """
    expected_dns = (cfg.wg.get("dns", "") or "").split(",")[0].strip()

    # Sous systemd-resolved, /etc/resolv.conf ne contient que le stub
    # 127.0.0.53 : lire ce fichier signalait donc toujours une « fuite ».
    # dns_util.active_resolvers() interroge resolvectl quand il est présent.
    resolvers = dns_util.active_resolvers()
    resolvers = [r for r in resolvers if r not in ("127.0.0.53", "127.0.0.1")] or resolvers

    tunnel_up = is_connected(cfg)

    # Sonde : quel résolveur récursif sort réellement sur le réseau.
    probe_ip = ""
    code, out, _ = run_command(["dig", "+short", "+time=3", "whoami.akamai.net"], 8)
    if code == 0:
        lines = [line.strip() for line in out.splitlines() if line.strip()]
        if lines:
            probe_ip = lines[-1]

    if not dns_util.resolution_works():
        status = "down"
    elif not resolvers:
        status = "unknown"
    elif not tunnel_up:
        # Tunnel inactif : parler de fuite n'a pas de sens.
        status = "unknown"
    elif not expected_dns:
        status = "unknown"
    elif any(expected_dns == r or expected_dns in r for r in resolvers):
        status = "ok"
    else:
        status = "leak"

    return {
        "status":    status,
        "resolvers": resolvers,
        "expected":  expected_dns,
        "probe_ip":  probe_ip,
        "tunnel_up": tunnel_up,
    }


# ── Génération config serveur ─────────────────────────────────────────────

def generate_server_config(
    server_port: int = 51820,
    server_address: str = "10.8.0.1/24",
    client_address: str = "10.8.0.2/24",
    client_allowed_ips: str = "10.8.0.2/32",
    use_psk: bool = True,
) -> dict:
    """
    Génère une paire complète (server_conf, client_conf) avec de nouvelles clés.
    Retourne un dict avec server_priv/pub, client_priv/pub, server_conf, client_conf.

    `use_psk=False` produit les deux configurations sans clé pré-partagée :
    l'assistant en imposait une des deux côtés, sans moyen de s'en passer.
    """
    srv_priv, srv_pub = generate_keypair()
    cli_priv, cli_pub = generate_keypair()
    psk = generate_preshared_key() if use_psk else ""
    psk_line = [f"PresharedKey = {psk}"] if use_psk else []

    server_conf = "\n".join([
        "[Interface]",
        f"PrivateKey = {srv_priv}",
        f"Address = {server_address}",
        f"ListenPort = {server_port}",
        "PostUp = iptables -A FORWARD -i %i -j ACCEPT; iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE",
        "PostDown = iptables -D FORWARD -i %i -j ACCEPT; iptables -t nat -D POSTROUTING -o eth0 -j MASQUERADE",
        "",
        "# === Peer client ===",
        "[Peer]",
        f"PublicKey = {cli_pub}",
        *psk_line,
        f"AllowedIPs = {client_allowed_ips}",
    ])

    client_conf = "\n".join([
        "[Interface]",
        f"PrivateKey = {cli_priv}",
        f"Address = {client_address}",
        "DNS = 1.1.1.1",
        "",
        "[Peer]",
        f"PublicKey = {srv_pub}",
        *psk_line,
        "AllowedIPs = 10.8.0.0/24",
        "Endpoint = <SERVER_IP>:" + str(server_port),
        "PersistentKeepalive = 25",
    ])

    return {
        "server_priv": srv_priv, "server_pub": srv_pub,
        "client_priv": cli_priv, "client_pub": cli_pub,
        "psk":         psk,
        "server_conf": server_conf,
        "client_conf": client_conf,
    }


def _fmt_bytes(n: int) -> str:
    for unit, div in (("GiB", 1 << 30), ("MiB", 1 << 20), ("KiB", 1 << 10)):
        if n >= div:
            return f"{n / div:.2f} {unit}"
    return f"{n} B"


def _fill_peer_stats(info: dict, out: str) -> None:
    """Complète `info` avec la sortie de `wg show <iface>`.

    Sans privilèges (Linux) ou sans droits administrateur (Windows), `wg show`
    ne renvoie rien : l'état de la connexion, lui, reste correct.
    """
    if not out:
        info["last_handshake"] = "—"
        return

    for line in out.splitlines():
        line = line.strip()
        if line.startswith("peer:"):
            info["peer"] = line.split(":", 1)[1].strip()[:16] + "…"
        elif line.startswith("transfer:"):
            parts = line.split(":", 1)[1].strip().split(",")
            try:
                # « 1.23 MiB received » → « 1.23 MiB » (l'unité était perdue)
                info["rx_bytes"] = " ".join(parts[0].split()[:2])
                info["tx_bytes"] = " ".join(parts[1].split()[:2])
            except Exception:
                pass
        elif line.startswith("latest handshake:"):
            info["last_handshake"] = line.split(":", 1)[1].strip()


def get_status_info(cfg: Config) -> dict:
    name = cfg.wg.get("interface_name", "wgs0")
    info = {
        "connected": False,
        "interface": name,
        "peer": "",
        "rx_bytes": 0,
        "tx_bytes": 0,
        "last_handshake": "",
    }
    info["connected"] = is_connected(cfg)
    if is_windows():
        if info["connected"]:
            _fill_peer_stats(info, _wg_show(name))
        return info
    if not info["connected"]:
        return info

    # Compteurs : /proc/net/dev est lisible sans privilèges, contrairement à
    # `wg show`. Sert de repli quand l'application tourne en simple utilisateur.
    counters = get_interface_bytes(name)
    if counters:
        info["rx_bytes"] = _fmt_bytes(counters[0])
        info["tx_bytes"] = _fmt_bytes(counters[1])

    _fill_peer_stats(info, _wg_show(name))
    return info