Files
WGSecure/app/core/dns.py
T
tuxgyverandClaude Sonnet 5 8ec7dc6aaf chore(release): v0.7.13
Corrige Add-DnsClientNrptRule qui échouait systématiquement : ce
cmdlet n'a pas de paramètre -Name (contrairement à Get-/Set-/
Remove-), résolu comme abréviation ambiguë entre -Namespace/
-NameServers/-NameEncoding. Identification de la règle via -Comment
à la place, comme le nettoyage générique existant. Bénéficie au
chemin élevé classique et au service wgsecure-helper.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-04 19:13:50 +02:00

577 lines
22 KiB
Python

"""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.helper_ipc import call_helper
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(
["powershell", "-NoProfile", "-NonInteractive", "-Command",
f"Get-Service -Name 'WireGuardTunnel${iface}' -ErrorAction SilentlyContinue"],
10
)
return code == 0 and out.strip()
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 }"
)
# Split-DNS : rule name template for per-tunnel DNS rules
_SPLIT_DNS_RULE_PREFIX = "WGSecure-SplitDNS-"
def _nrpt_add_split_dns_ps(server: str, domains: list[str], rule_name: str) -> str:
"""PowerShell script to add a NRPT rule for split-DNS.
Args:
server: DNS server IP address
domains: List of domain suffixes (e.g., ["local", "internal"])
rule_name: Name for the NRPT rule
"""
if not domains:
return ""
clean = [d.strip().strip(".") for d in domains if d.strip()]
# NRPT : un namespace sans point de tête ('h3adm.lan') ne matche que ce
# nom exact, jamais ses sous-domaines ; un namespace avec point de tête
# ('.h3adm.lan') ne matche que les sous-domaines, jamais le domaine lui-
# même. Les deux formes sont nécessaires pour couvrir toute la zone —
# comme `~domaine` sous systemd-resolved (Linux) couvre déjà les deux.
namespace_list = ", ".join(f"'{d}', '.{d}'" for d in clean)
# Add-DnsClientNrptRule n'a PAS de paramètre -Name (contrairement à
# Get-/Set-/Remove-) : la règle créée reçoit un identifiant GUID auto-
# généré, jamais le nôtre. "-Name" était donc résolu par PowerShell
# comme une abréviation ambiguë entre -Namespace/-NameServers/
# -NameEncoding (« le nom du paramètre "name" est ambigu ») — la pose de
# la règle échouait donc systématiquement, masqué jusqu'ici par
# l'ancienne incapacité à remonter le texte d'une erreur élevée. On
# identifie/retrouve donc la règle via -Comment (qui, lui, existe bien
# sur Add-), comme le fait déjà le nettoyage générique plus bas
# (_NRPT_CLEAN_PS, filtré sur .Comment).
return (
f"Get-DnsClientNrptRule -ErrorAction SilentlyContinue | "
f"Where-Object {{ $_.Comment -eq '{rule_name}' }} | "
f"ForEach-Object {{ Remove-DnsClientNrptRule -Name $_.Name -Force -ErrorAction SilentlyContinue }}; "
f"Add-DnsClientNrptRule -Namespace @({namespace_list}) "
f"-NameServers '{server}' -Comment '{rule_name}'; "
# Sans ce contrôle, une erreur réelle d'Add-DnsClientNrptRule restait
# invisible : le script se terminait quand même avec le code 0, et
# `setup_split_dns` rapportait un succès alors que la règle n'avait
# jamais été posée.
"if (-not $?) { exit 1 }"
)
def _nrpt_remove_split_dns_ps(rule_name: str) -> str:
"""PowerShell script to remove a NRPT rule for split-DNS.
La règle n'a pas de -Name pris en charge à la création (cf.
`_nrpt_add_split_dns_ps`) : on la retrouve par son -Comment plutôt que
par un -Name qui ne correspondrait jamais à rien.
"""
return (
f"Get-DnsClientNrptRule -ErrorAction SilentlyContinue | "
f"Where-Object {{ $_.Comment -eq '{rule_name}' }} | "
f"ForEach-Object {{ Remove-DnsClientNrptRule -Name $_.Name -Force -ErrorAction SilentlyContinue }}"
)
def _systemd_resolved_available() -> bool:
"""Vérifie si systemd-resolved est actif (Linux)."""
if is_windows():
return False
import subprocess
try:
# Vérifier que systemd-resolved est actif
code = subprocess.run(
["systemctl", "is-active", "systemd-resolved"],
capture_output=True, text=True, timeout=5
).returncode
return code == 0
except Exception:
return False
def _setup_split_dns_linux(server: str, domains: list[str], iface: str) -> tuple[bool, str]:
"""Configure le split-DNS sous Linux via systemd-resolved, par lien.
Écrivait auparavant un fichier global dans /etc/systemd/resolved.conf.d/
et redémarrait tout le service — deux privilèges que `make setup-sudoers`
ne couvrait pas (échec silencieux ou dialogue à chaque connexion), pour
un réglage qui entrait en concurrence avec la configuration éventuelle
d'autres liens et survivait à un arrêt brutal du tunnel.
`resolvectl dns`/`resolvectl domain` posent la même route mais rattachée
à l'interface WireGuard elle-même — comme `wg-quick` le fait déjà pour le
DNS non split. `resolvectl revert <iface>` (déjà autorisé sans mot de
passe, cf. `make setup-sudoers`) l'annule d'un coup, et systemd-resolved
la retire de lui-même dès que l'interface disparaît : plus de résidu à
surveiller après un crash.
"""
if not shutil.which("resolvectl"):
return False, "resolvectl introuvable : split-DNS indisponible sur ce système"
if not _systemd_resolved_available():
return False, "systemd-resolved non actif : split-DNS indisponible"
code, out, err = run_privileged(["resolvectl", "dns", iface, server], timeout=10)
if code != 0:
return False, f"Échec de configuration du DNS distant : {err or out or 'erreur inconnue'}"
routing_domains = [f"~{d}" for d in domains]
code, out, err = run_privileged(["resolvectl", "domain", iface, *routing_domains], timeout=10)
if code != 0:
return False, f"Échec de configuration du domaine distant : {err or out or 'erreur inconnue'}"
return True, f"Split-DNS configuré pour {server} (domaines : {', '.join(domains)})"
def setup_split_dns(cfg: Config) -> tuple[bool, str]:
"""Configure le split-DNS pour Windows (NRPT) ou Linux (systemd-resolved).
Args:
cfg: Configuration contenant split_dns_server et split_dns_domains
Returns:
(success, message)
"""
server = (cfg.wg.get("split_dns_server") or "").strip()
domains = cfg.wg.get("split_dns_domains") or []
name = cfg.wg.get("interface_name", "wgs0")
if not server or not domains:
return True, "Aucune configuration split-DNS"
# Nettoyer les domaines
domains = [d.strip() for d in domains if d.strip()]
if not domains:
return True, "Aucun domaine valide à configurer"
if is_windows():
rule_name = f"{_SPLIT_DNS_RULE_PREFIX}{name}"
# Délègue à wgsecure-helper (service SYSTEM) s'il est installé : la
# règle NRPT est posée sans invite d'authentification. `None` =
# service absent/indisponible, on retombe sur le chemin élevé.
resp = call_helper({
"cmd": "set_split_dns", "server": server,
"domains": domains, "rule_name": rule_name,
})
if resp is not None:
if resp.get("ok"):
return True, f"Règle split-DNS (NRPT) configurée pour {server} (domaines: {', '.join(domains)})"
return False, f"Échec de la configuration split-DNS (NRPT): {resp.get('message') or 'Erreur inconnue'}"
script = _nrpt_add_split_dns_ps(server, domains, rule_name)
if not script:
return True, "Aucun domaine à configurer"
# Exécuter avec élévation (nécessite admin pour modifier les règles NRPT)
code, out, err = run_privileged(
["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
timeout=15
)
if code == 0:
return True, f"Règle split-DNS (NRPT) configurée pour {server} (domaines: {', '.join(domains)})"
else:
return False, f"Échec de la configuration split-DNS (NRPT): {err or out or 'Erreur inconnue'}"
else:
# Linux : utiliser systemd-resolved
return _setup_split_dns_linux(server, domains, name)
def _cleanup_split_dns_linux(iface: str) -> tuple[bool, str]:
"""Annule la config split-DNS par-lien de cette interface.
`force_cleanup()` appelle ceci après avoir déjà supprimé l'interface la
plupart du temps (systemd-resolved retire alors sa config par-lien tout
seul) : `resolvectl revert` échoue simplement sur une interface déjà
partie, ce qui n'est pas une erreur — rien à nettoyer, pas un échec.
"""
if not shutil.which("resolvectl"):
return True, ""
code, out, err = run_privileged(["resolvectl", "revert", iface], timeout=10)
if code == 0:
return True, "Configuration split-DNS retirée"
return True, ""
def cleanup_split_dns(cfg: Config) -> tuple[bool, str]:
"""Supprime la configuration split-DNS pour Windows (NRPT) ou Linux.
Args:
cfg: Configuration contenant l'interface name
Returns:
(success, message)
"""
if is_windows():
name = cfg.wg.get("interface_name", "wgs0")
rule_name = f"{_SPLIT_DNS_RULE_PREFIX}{name}"
resp = call_helper({"cmd": "clear_split_dns", "rule_name": rule_name})
if resp is not None:
if resp.get("ok"):
return True, "Règle split-DNS (NRPT) supprimée"
msg = resp.get("message") or ""
if "cannot find" in msg.lower() or "not found" in msg.lower():
return True, ""
return False, f"Échec de la suppression split-DNS (NRPT): {msg or 'Erreur inconnue'}"
script = _nrpt_remove_split_dns_ps(rule_name)
# Exécuter avec élévation
code, out, err = run_privileged(
["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
timeout=15
)
if code == 0:
return True, "Règle split-DNS (NRPT) supprimée"
else:
# Ignorer l'erreur si la règle n'existait pas
if "cannot find" in (err or out or "").lower() or "not found" in (err or out or "").lower():
return True, ""
return False, f"Échec de la suppression split-DNS (NRPT): {err or out or 'Erreur inconnue'}"
else:
# Linux : supprimer le fichier de config systemd-resolved
name = cfg.wg.get("interface_name", "wgs0")
return _cleanup_split_dns_linux(name)
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 resolve_host(host: str, timeout: float = 3.0) -> tuple[bool, list[str]]:
"""Résout un nom via le résolveur système. Retourne (succès, IPs uniques).
Sert au split-DNS : `resolution_works()` ne confirme qu'un résolveur
quelconque répond, jamais que les noms du réseau distant (le domaine
configuré en split-DNS, ou un hôte à l'intérieur) résolvent bien vers ce
réseau plutôt que vers un NXDOMAIN ou une réponse publique inattendue.
"""
old = socket.getdefaulttimeout()
socket.setdefaulttimeout(timeout)
try:
infos = socket.getaddrinfo(host, None)
ips = list(dict.fromkeys(info[4][0] for info in infos))
return True, ips
except OSError:
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