WGSecure/app/core/dns.py

"""Gestion et réparation de l'état DNS système.

Pourquoi ce module existe
-------------------------
``wg-quick up`` installe le DNS du tunnel via ``resolvconf -a <iface> -m 0 -x``
(mode exclusif : le résolveur du tunnel devient le seul utilisé). Le retrait de
cette entrée n'a lieu que dans ``wg-quick down``, qui commence par :

    [[ " $(wg show interfaces) " == *" $INTERFACE "* ]] || die "... is not a WireGuard interface"

Autrement dit : si l'interface WireGuard a disparu sans passer par
``wg-quick down`` (crash de l'application, kill, veille/reprise, redémarrage,
délai d'attente sur l'élévation de privilèges qui tue wg-quick en plein vol),
``wg-quick down`` échoue immédiatement et ``unset_dns`` n'est jamais appelé.
L'entrée DNS reste enregistrée et pointe vers un résolveur devenu injoignable :
plus aucune résolution de nom ne fonctionne, donc « plus d'internet ».

Ce module retire cette entrée de façon inconditionnelle et idempotente, quel
que soit l'état de l'interface.
"""
from __future__ import annotations

import os
import shutil
import socket

from app.utils.platform_utils import is_windows, run_command, run_privileged

# Domaine utilisé pour vérifier que la résolution DNS fonctionne encore.
_PROBE_HOSTS = ("cloudflare.com", "example.com")


def resolvconf_iface(iface: str) -> str:
    """Nom d'interface tel que wg-quick l'enregistre auprès de resolvconf.

    Reproduit ``resolvconf_iface_prefix()`` de wg-quick : le préfixe
    (``tun.`` sur Debian/openresolv) n'est utilisé que si
    /etc/resolvconf/interface-order existe ET que resolvconf n'est pas un lien
    symbolique (sur Ubuntu, /usr/sbin/resolvconf -> resolvectl, donc pas de
    préfixe).
    """
    order = "/etc/resolvconf/interface-order"
    path = shutil.which("resolvconf")
    if not path or not os.path.isfile(order) or os.path.islink(path):
        return iface
    try:
        with open(order, "r", encoding="utf-8") as f:
            for raw in f:
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                # wg-quick prend le premier motif se terminant par '*'
                if line.endswith("*"):
                    return line[:-1] + iface
    except OSError:
        pass
    return iface


def has_dns_backend() -> bool:
    """True si wg-quick pourra appliquer la directive DNS = du fichier .conf."""
    if is_windows():
        return True
    return bool(shutil.which("resolvconf") or shutil.which("resolvectl"))


def interface_exists(iface: str) -> bool:
    """True si le périphérique réseau existe encore (indépendamment de wg)."""
    if is_windows():
        # Le service du tunnel survit à la fenêtre WireGuard : c'est lui qui
        # fait foi, pas la présence de l'adaptateur.
        code, out, _ = run_command(["sc", "query", f"WireGuardTunnel${iface}"], 10)
        return code == 0 and "1060" not in out
    return os.path.isdir(f"/sys/class/net/{iface}")


def interface_is_up(iface: str) -> bool:
    """True si l'interface existe et porte le drapeau IFF_UP.

    Lecture directe de sysfs : contrairement à `wg show`, cela ne demande
    aucun privilège. `operstate` vaut « unknown » sur une interface WireGuard
    (point à point sans porteuse), il faut donc lire `flags`.
    """
    try:
        with open(f"/sys/class/net/{iface}/flags", encoding="utf-8") as f:
            return bool(int(f.read().strip(), 16) & 0x1)   # IFF_UP
    except (OSError, ValueError):
        return False


def list_wireguard_interfaces() -> list[str]:
    """Interfaces WireGuard présentes sur la machine, sans privilèges.

    `wg show` exige CAP_NET_ADMIN ; sysfs expose DEVTYPE=wireguard à tous.
    """
    if is_windows():
        return []
    found: list[str] = []
    try:
        for name in sorted(os.listdir("/sys/class/net")):
            try:
                with open(f"/sys/class/net/{name}/uevent", encoding="utf-8") as f:
                    if "DEVTYPE=wireguard" in f.read():
                        found.append(name)
            except OSError:
                continue
    except OSError:
        pass
    return found


def _resolvconf_run_files(rc_iface: str) -> list[str]:
    """Chemins où openresolv/resolvconf stocke l'entrée d'une interface."""
    return [
        f"/run/resolvconf/interface/{rc_iface}",
        f"/etc/resolvconf/run/interface/{rc_iface}",
        f"/var/run/resolvconf/interface/{rc_iface}",
    ]


def has_stale_entry(iface: str, tunnel_dns: str = "") -> bool:
    """True s'il reste une trace DNS du tunnel alors que l'interface est morte.

    Vérification purement locale : aucune commande privilégiée n'est lancée,
    pour ne pas déclencher de dialogue d'authentification inutile au démarrage.
    """
    if is_windows():
        return _has_stale_nrpt_rule()
    if interface_exists(iface):
        return False

    # 1. Entrée resolvconf/openresolv orpheline.
    rc_iface = resolvconf_iface(iface)
    for path in _resolvconf_run_files(rc_iface):
        if os.path.exists(path):
            return True

    # 2. /etc/resolv.conf remplacé par un fichier simple pointant sur le DNS
    #    du tunnel (cas des systèmes sans systemd-resolved).
    tunnel_dns = (tunnel_dns or "").split(",")[0].strip()
    if tunnel_dns and not os.path.islink("/etc/resolv.conf"):
        try:
            with open("/etc/resolv.conf", encoding="utf-8") as f:
                for line in f:
                    parts = line.split()
                    if len(parts) >= 2 and parts[0] == "nameserver" \
                            and parts[1] == tunnel_dns:
                        return True
        except OSError:
            pass

    return False


_NRPT_LIST_PS = (
    "$r = Get-DnsClientNrptRule -ErrorAction SilentlyContinue | "
    "Where-Object { $_.Comment -like '*WireGuard*' }; "
    "if ($r) { 'STALE' }"
)

_NRPT_CLEAN_PS = (
    "Get-DnsClientNrptRule -ErrorAction SilentlyContinue | "
    "Where-Object { $_.Comment -like '*WireGuard*' } | "
    "ForEach-Object { Remove-DnsClientNrptRule -Name $_.Name -Force "
    "-ErrorAction SilentlyContinue }"
)


def _powershell(script: str, timeout: int = 20) -> tuple[int, str, str]:
    return run_command(
        ["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
        timeout,
    )


def _has_stale_nrpt_rule() -> bool:
    """True s'il reste une règle NRPT posée par WireGuard.

    Équivalent Windows de l'entrée resolvconf orpheline : quand le tunnel est
    arrêté brutalement, la règle NRPT redirige toujours la résolution vers le
    DNS du VPN, devenu injoignable — plus aucune résolution ne fonctionne.
    """
    code, out, _ = _powershell(_NRPT_LIST_PS, 15)
    return code == 0 and "STALE" in out


def _cleanup_windows(iface: str, force: bool) -> list[str]:
    done: list[str] = []

    if force or _has_stale_nrpt_rule():
        code, _, _ = _powershell(_NRPT_CLEAN_PS, 30)
        if code == 0:
            done.append("Remove-DnsClientNrptRule (WireGuard)")

    # Purge du cache : les échecs mis en cache pendant la panne survivraient.
    run_command(["ipconfig", "/flushdns"], 15)
    return done


def cleanup(iface: str, force: bool = False) -> list[str]:
    """Retire toute entrée DNS résiduelle laissée par le tunnel `iface`.

    Par défaut, aucune commande privilégiée n'est lancée tant qu'une trace
    résiduelle n'a pas été constatée sans privilèges : sur un système
    systemd-resolved, la configuration DNS disparaît avec le lien réseau, et
    déclencher un dialogue polkit à chaque déconnexion serait inutile.
    Passer ``force=True`` pour forcer le retrait (interface encore présente).

    Idempotent. Retourne la liste des actions réellement effectuées.
    """
    if is_windows():
        return _cleanup_windows(iface, force)

    done: list[str] = []

    if force or has_stale_entry(iface):
        rc_iface = resolvconf_iface(iface)

        # 1. resolvconf : c'est le chemin qu'emprunte wg-quick lui-même.
        if shutil.which("resolvconf"):
            # -f : ne pas échouer si l'entrée n'existe pas.
            code, _, _ = run_privileged(["resolvconf", "-d", rc_iface, "-f"],
                                        timeout=20)
            if code == 0:
                done.append(f"resolvconf -d {rc_iface}")

        # 2. systemd-resolved : purge la configuration portée par le lien.
        #    `resolvectl revert` échoue sans dommage si le lien a disparu.
        if shutil.which("resolvectl") and interface_exists(iface):
            code, _, _ = run_privileged(["resolvectl", "revert", iface], timeout=20)
            if code == 0:
                done.append(f"resolvectl revert {iface}")

    # 3. Vider le cache : sans cela les NXDOMAIN obtenus pendant la panne
    #    restent servis pendant leur TTL négatif. Non privilégié.
    if shutil.which("resolvectl"):
        run_command(["resolvectl", "flush-caches"], timeout=10)

    return done


def resolution_works(timeout: float = 3.0) -> bool:
    """True si la résolution DNS système répond."""
    old = socket.getdefaulttimeout()
    socket.setdefaulttimeout(timeout)
    try:
        for host in _PROBE_HOSTS:
            try:
                socket.getaddrinfo(host, None)
                return True
            except OSError:
                continue
        return False
    finally:
        socket.setdefaulttimeout(old)


def repair_if_broken(iface: str, tunnel_dns: str = "") -> tuple[bool, str]:
    """Répare le DNS si le tunnel est absent mais une entrée résiduelle traîne.

    Appelé au démarrage de l'application : c'est le cas typique après un crash
    ou un arrêt brutal de la session précédente, où ``wg-quick down`` n'a
    jamais tourné et où l'entrée DNS du tunnel reste active alors que le
    résolveur correspondant est injoignable.

    Retourne (réparation_effectuée, message).
    """
    if interface_exists(iface):
        # Le tunnel tourne : ne pas toucher à sa configuration DNS.
        return False, ""
    if not has_stale_entry(iface, tunnel_dns):
        return False, ""
    if resolution_works():
        # Entrée orpheline mais résolution fonctionnelle : nettoyage discret,
        # sans escalade de privilèges susceptible d'ouvrir un dialogue.
        return False, ""

    actions = cleanup(iface)
    if not actions:
        return False, "Entrée DNS résiduelle détectée, nettoyage impossible"
    if resolution_works():
        return True, "DNS restauré (entrée résiduelle du tunnel retirée)"
    return True, "Entrée DNS résiduelle retirée, mais la résolution échoue encore"


def active_resolvers() -> list[str]:
    """Liste les résolveurs DNS effectivement utilisés par le système."""
    if is_windows():
        import re
        code, out, _ = run_command(["ipconfig", "/all"], 15)
        if code != 0:
            return []
        found: list[str] = []
        for block in re.findall(
            r"DNS Servers[ .]*:\s*(.*?)(?=\n\s*[A-Za-z][^\n]*[ .]+:|\Z)",
            out, re.DOTALL,
        ):
            for line in block.splitlines():
                ip = line.strip()
                if re.fullmatch(r"[0-9a-fA-F.:]+", ip or ""):
                    found.append(ip)
        return found

    # systemd-resolved : /etc/resolv.conf ne contient que le stub 127.0.0.53,
    # les vrais résolveurs ne sont visibles que via resolvectl.
    resolvers: list[str] = []
    if shutil.which("resolvectl"):
        code, out, _ = run_command(["resolvectl", "dns"], 10)
        if code == 0:
            for line in out.splitlines():
                if ":" not in line:
                    continue
                for ip in line.split(":", 1)[1].split():
                    if ip and ip not in resolvers:
                        resolvers.append(ip)

    if not resolvers:
        try:
            with open("/etc/resolv.conf", encoding="utf-8") as f:
                for line in f:
                    parts = line.split()
                    if len(parts) >= 2 and parts[0] == "nameserver":
                        if parts[1] not in resolvers:
                            resolvers.append(parts[1])
        except OSError:
            pass
    return resolvers