Johnny/WGSecure
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: DNS coupée après arrêt du tunnel, gel UI, dialogues sudo à répétition

Browse Source 
- wg-quick down échoue silencieusement si l'interface a déjà disparu
  (crash, kill, veille) et ne retire donc jamais l'entrée DNS posée par
  wg-quick up : la résolution reste pointée sur un résolveur mort.
  Ajout d'un nettoyage DNS inconditionnel (app/core/dns.py) déclenché à
  la connexion, la déconnexion, la fermeture et au démarrage suivant.
- is_connected() se basait sur `wg show`, qui rend un code 0 même en cas
  d'échec de permission pour un utilisateur non-root : l'app se croyait
  déconnectée en permanence et ne démontait donc jamais le tunnel à la
  fermeture. Détection réécrite via sysfs (sans privilèges).
- Connexion/déconnexion et ping tournaient sur le thread Qt : gel de
  l'UI pendant les élévations de privilèges. Déportés dans des
  QThread (app/ui/worker.py).
- Restauration réseau garantie sur toutes les sorties (croix, tray,
  SIGINT/SIGTERM, atexit, exception) via app/core/shutdown.py.
- run_privileged() : la détection "sudo veut un mot de passe" ratait en
  session non-anglophone et sur les comptes ALL=(ALL) ALL ; réécrite
  pour se fier au message d'erreur réel plutôt qu'à `sudo -l`.
- Écriture de la config WireGuard dans /etc/wireguard (root:root 0700)
  passe maintenant par install -D via la chaîne de privilèges complète
  au lieu d'un tee non autorisé par les sudoers.
- Ajout d'un diagnostic des droits (make check-privileges, bouton
  Admin) et mise à jour de setup-sudoers pour couvrir toutes les
  commandes désormais utilisées.
This commit is contained in:
tuxgyver
2026-08-31 11:16:04 +02:00
parent 2fa50a5873
commit 1b74719204
11 changed files with 1295 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+49 -17
View File
@@ -27,17 +27,29 @@ PI_OPTS := \
--hidden-import cryptography.hazmat.primitives.asymmetric.x25519 \
--collect-submodules PyQt6
.PHONY: all linux windows release install install-gnome uninstall-gnome run run-admin icon reset-password setup-sudoers clean clean-all help venv
.PHONY: all linux windows release install install-gnome uninstall-gnome run run-admin icon reset-password setup-sudoers check-privileges clean clean-all help venv
# ── Cible par défaut ────────────────────────────────────────────────────────
all: linux
# ── Environnement virtuel ────────────────────────────────────────────────────
venv: $(VENV)/bin/python3
$(VENV)/bin/python3:
python3 -m venv $(VENV)
$(PIP) install --upgrade pip --quiet
# Le venv est reconstruit si la version de Python du système a changé : un venv
# créé pour python3.13 pointe vers .venv/lib/python3.13/site-packages, que
# python3.14 n'ouvre pas. L'interpréteur démarrait alors sans aucune de ses
# dépendances (« No module named PyQt6 ») sans que make ne le remarque.
venv:
@if [ -x $(PYTHON) ] && \
[ "$$($(PYTHON) -c 'import sys;print("%d.%d"%sys.version_info[:2])' 2>/dev/null)" \
= "$$(python3 -c 'import sys;print("%d.%d"%sys.version_info[:2])')" ] && \
$(PYTHON) -c 'import PyQt6' >/dev/null 2>&1; then \
echo " ✅ Environnement virtuel valide ($(VENV))"; \
else \
echo " ♻️ (Re)création de l'environnement virtuel $(VENV)…"; \
rm -rf $(VENV); \
python3 -m venv $(VENV); \
$(PIP) install --upgrade pip --quiet; \
$(PIP) install -r requirements.txt --quiet; \
fi
# ── Génération de l'icône PNG (utilisée par PyInstaller) ────────────────────
icon:
@@ -147,23 +159,43 @@ install: venv
$(PIP) install pyinstaller
@echo " ✅ Dépendances installées dans $(VENV)"
# ── Diagnostic des droits ────────────────────────────────────────────────────
check-privileges:
@echo ""
@echo " 🔐 Droits requis par WGSecure :"
@$(PYTHON) -c "\
import sys; sys.path.insert(0, '.'); \
from app.utils.platform_utils import privilege_report; \
rows = privilege_report(); \
print(' (exécution en root : aucun droit à configurer)') if not rows else None; \
[print(' ' + ('✅' if ok else '⚠️ ') + ' ' + name.ljust(12) + ('sans mot de passe' if ok else 'dialogue polkit à chaque appel').ljust(32) + '— ' + role) for name, ok, role in rows]; \
print() if not rows else print(chr(10) + ' ⚠️ → lancez make setup-sudoers pour supprimer les dialogues.' if any(not ok for _, ok, _ in rows) else chr(10) + ' ✅ Tout est configuré.')"
@echo ""
# ── Droits sudo wg-quick sans dialogue de mot de passe ───────────────────────
setup-sudoers:
@echo ""
@echo " 🔧 Configuration sudoers pour wg-quick (sans mot de passe)…"
@echo " 🔧 Configuration sudoers (wg-quick + nettoyage réseau, sans mot de passe)…"
@$(PYTHON) -c "\
import os, subprocess; \
import os, subprocess, tempfile, shutil; \
user = os.environ.get('USER', os.path.basename(os.path.expanduser('~'))); \
rules = ( \
user + ' ALL=(ALL) NOPASSWD: /usr/bin/wg-quick\n' \
+ user + ' ALL=(ALL) NOPASSWD: /usr/bin/tee /etc/wireguard/*\n' \
+ user + ' ALL=(ALL) NOPASSWD: /bin/chmod 600 /etc/wireguard/*\n' \
+ user + ' ALL=(ALL) NOPASSWD: /usr/bin/chmod 600 /etc/wireguard/*\n' \
); \
tmp = '/tmp/wgsecure_sudoers_tmp'; \
open(tmp, 'w').write(rules); \
r = subprocess.run(['pkexec', 'bash', '-c', 'cp ' + tmp + ' /etc/sudoers.d/wgsecure && chmod 440 /etc/sudoers.d/wgsecure']); \
w = lambda n: shutil.which(n) or ''; \
cmds = [c for c in ( \
w('wg-quick'), \
w('wg'), \
w('ip') + ' link delete dev *' if w('ip') else '', \
w('resolvconf') + ' -d *' if w('resolvconf') else '', \
w('resolvectl') + ' revert *' if w('resolvectl') else '', \
w('install') + ' -D -m 600 -o root -g root * /etc/wireguard/*' if w('install') else '', \
) if c]; \
rules = ''.join(user + ' ALL=(ALL) NOPASSWD: ' + c + chr(10) for c in cmds); \
fd, tmp = tempfile.mkstemp(prefix='wgsecure_sudoers_', dir=os.path.expanduser('~')); \
os.write(fd, rules.encode()); os.close(fd); os.chmod(tmp, 0o600); \
chk = subprocess.run(['visudo', '-cqf', tmp], capture_output=True); \
ok = chk.returncode == 0; \
r = subprocess.run(['pkexec', 'bash', '-c', 'install -m 440 -o root -g root ' + tmp + ' /etc/sudoers.d/wgsecure']) if ok else None; \
os.unlink(tmp); \
print(' ❌ Règles sudoers invalides, rien installé : ' + chk.stderr.decode().strip()) if not ok else \
print(' ✅ /etc/sudoers.d/wgsecure configuré — plus de dialogue admin.' if r.returncode == 0 else ' ❌ Échec.')"
@echo ""
app/core/config.py
+6 -8
View File
@@ -196,11 +196,10 @@ class Config:
self.save()
def _apply_autostart(self, enabled: bool) -> None:
import sys
from app.utils.platform_utils import is_linux, is_windows
script = os.path.abspath(os.path.join(
os.path.dirname(__file__), "..", "..", "main.py"
))
from app.utils.platform_utils import is_linux, is_windows, launch_command
# launch_command() gère le cas du binaire PyInstaller, où main.py
# n'existe pas : l'ancien « python3 <…>/main.py » ne démarrait jamais.
command = launch_command()
if is_linux():
autostart_dir = os.path.expanduser("~/.config/autostart")
desktop = os.path.join(autostart_dir, "wgsecure.desktop")
@@ -208,7 +207,7 @@ class Config:
os.makedirs(autostart_dir, exist_ok=True)
content = (
"[Desktop Entry]\nType=Application\nName=WGSecure\n"
f"Exec=python3 {script}\nIcon=wgsecure\n"
f"Exec={command}\nIcon=wgsecure\n"
"Hidden=false\nX-GNOME-Autostart-enabled=true\n"
)
with open(desktop, "w") as f:
@@ -225,8 +224,7 @@ class Config:
reg = winreg.OpenKey(winreg.HKEY_CURRENT_USER, key, 0,
winreg.KEY_SET_VALUE)
if enabled:
winreg.SetValueEx(reg, "WGSecure", 0, winreg.REG_SZ,
f'pythonw "{script}"')
winreg.SetValueEx(reg, "WGSecure", 0, winreg.REG_SZ, command)
else:
winreg.DeleteValue(reg, "WGSecure")
winreg.CloseKey(reg)
app/core/dns.py
+327
View File
@@ -0,0 +1,327 @@
"""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
app/core/shutdown.py
+131
View File
@@ -0,0 +1,131 @@
"""Restauration garantie de l'état réseau à la fermeture de l'application.
Le tunnel WireGuard modifie deux choses globales au système : la table de
routage et la résolution DNS. Tant que l'application ne les rend pas, la
machine reste dans un état où « il n'y a plus internet » — c'est vrai sous
Linux (entrée resolvconf orpheline) comme sous Windows (service de tunnel
resté en place et règle NRPT résiduelle).
Ce module branche la même routine de restauration sur *toutes* les sorties :
* clic sur « Quitter » dans le systray ou fermeture de la fenêtre ;
* signal SIGINT / SIGTERM (Ctrl-C, arrêt de session, `kill`) ;
* SIGBREAK sous Windows (fermeture de la console) ;
* `atexit`, qui couvre `sys.exit()` et la fin normale de l'interpréteur ;
* exception non rattrapée, via `sys.excepthook`.
La restauration est idempotente : la déclencher plusieurs fois est sans effet.
"""
from __future__ import annotations
import atexit
import signal
import sys
import threading
from typing import Callable
from app.core import wireguard as wg_core
from app.core.config import Config
_lock = threading.Lock()
_restored = False
_cfg: Config | None = None
_report: Callable[[str, str], None] | None = None
_installed = False
def _log(msg: str, kind: str = "info") -> None:
if _report is not None:
try:
_report(msg, kind)
except Exception:
pass
def restore_network(reason: str = "") -> list[str]:
"""Rend le réseau à son état d'origine. Idempotent et sans exception.
Retourne la liste des actions effectuées (vide si rien n'était à faire).
"""
global _restored
with _lock:
if _restored or _cfg is None:
return []
_restored = True
cfg = _cfg
actions: list[str] = []
try:
if wg_core.is_connected(cfg):
ok, msg = wg_core.disconnect(cfg)
actions.append("tunnel arrêté" if ok else f"échec arrêt tunnel : {msg}")
_log(f"Fermeture{(' (' + reason + ')') if reason else ''} :"
f" {actions[-1]}", "info" if ok else "error")
else:
# Le tunnel n'est plus là, mais sa configuration DNS peut avoir
# survécu : c'est précisément le cas qui coupe la résolution.
actions += wg_core.force_cleanup(cfg)
if actions:
_log("Résidus réseau nettoyés à la fermeture : "
+ ", ".join(actions), "warning")
except Exception as e: # noqa: BLE001
_log(f"Restauration réseau incomplète : {e}", "error")
return actions
def _signal_handler(signum, _frame):
restore_network(f"signal {signum}")
# Réinstaller le comportement par défaut puis se renvoyer le signal :
# le code de sortie reste celui attendu par l'appelant (shell, systemd).
try:
signal.signal(signum, signal.SIG_DFL)
except (ValueError, OSError):
pass
sys.exit(128 + int(signum))
def _excepthook(exc_type, exc, tb):
restore_network("exception non rattrapée")
sys.__excepthook__(exc_type, exc, tb)
def install(cfg: Config, report: Callable[[str, str], None] | None = None,
app=None) -> None:
"""Branche la restauration sur toutes les voies de sortie du programme."""
global _cfg, _report, _installed
_cfg = cfg
_report = report
if _installed:
return
_installed = True
atexit.register(restore_network, "atexit")
sys.excepthook = _excepthook
for name in ("SIGINT", "SIGTERM", "SIGBREAK", "SIGHUP"):
sig = getattr(signal, name, None)
if sig is None:
continue
try:
signal.signal(sig, _signal_handler)
except (ValueError, OSError):
# Signal indisponible sur la plateforme ou hors thread principal.
pass
if app is not None:
app.aboutToQuit.connect(lambda: restore_network("aboutToQuit"))
# Qt bloque dans du code C : sans réveil périodique, les gestionnaires
# de signaux Python ne s'exécutent jamais pendant la boucle d'événements.
from PyQt6.QtCore import QTimer
heartbeat = QTimer(app)
heartbeat.timeout.connect(lambda: None)
heartbeat.start(400)
app._wgs_signal_heartbeat = heartbeat # garder une référence vivante
def arm() -> None:
"""Réarme la restauration après une reconnexion volontaire."""
global _restored
with _lock:
_restored = False
app/core/wireguard.py
+266 -79
View File
@@ -3,6 +3,7 @@ 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.config import Config
from app.utils.platform_utils import (
get_config_dir,
@@ -35,7 +36,15 @@ def build_client_config(cfg: Config) -> str:
"[Interface]",
f"PrivateKey = {wg['client_private_key']}",
f"Address = {wg['client_address']}",
f"DNS = {wg['dns']}",
]
# 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.
dns_value = (wg.get("dns") or "").strip()
if dns_value and dns_util.has_dns_backend():
lines.append(f"DNS = {dns_value}")
lines += [
"",
"[Peer]",
f"PublicKey = {wg['server_public_key']}",
@@ -54,8 +63,19 @@ def get_client_config_path(cfg: Config) -> str:
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:
@@ -63,39 +83,123 @@ def write_client_config(cfg: Config) -> tuple[bool, str]:
if not is_windows():
os.chmod(path, 0o600)
return True, path
except PermissionError:
# sudo -n tee (NOPASSWD si configuré, sinon pkexec cp)
import subprocess as _sp
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:
r = _sp.run(["sudo", "-n", "tee", path],
input=content, text=True,
capture_output=True, timeout=10)
if r.returncode == 0:
_sp.run(["sudo", "-n", "chmod", "600", path],
capture_output=True, timeout=5)
return True, path
except Exception:
pass
# Repli sur pkexec cp (dialogue graphique)
tmp = os.path.join(get_config_dir(), "wgs_tmp.conf")
with open(tmp, "w") as f:
# 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)
os.chmod(tmp, 0o600)
code, _, err = run_privileged(["cp", tmp, path])
os.unlink(tmp)
code, _, err = run_privileged(
["install", "-D", "-m", "600", "-o", "root", "-g", "root", tmp, path],
timeout=60,
)
if code == 0:
run_privileged(["chmod", "600", path])
return True, path
return False, err
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.
Attention : sans CAP_NET_ADMIN, `wg show` écrit « Unable to access
interface: Operation not permitted » sur stderr **et sort avec le code 0**.
Un simple test `code == 0` conclut donc « déconnecté » pour tout
utilisateur non root — c'était le cas de l'application entière.
"""
code, out, err = run_command(["wg", "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(["sc", "query", f"WireGuardTunnel${name}"])
return code == 0 and "RUNNING" in out
code, out, _ = run_command(["wg", "show", name])
return code == 0 and bool(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 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 ; s'il est
# bloqué, on le supprime avant de purger les règles NRPT résiduelles.
if dns_util.interface_exists(name):
svc = f"WireGuardTunnel${name}"
run_command(["wireguard", "/uninstalltunnel", name], 30)
if dns_util.interface_exists(name):
run_privileged(["sc", "stop", svc], timeout=30)
run_privileged(["sc", "delete", svc], timeout=30)
actions.append(f"sc delete {svc}")
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)
return actions
def connect(cfg: Config) -> tuple[bool, str]:
@@ -111,11 +215,25 @@ def connect(cfg: Config) -> tuple[bool, str]:
if is_windows():
code, _, err = run_command(["wireguard", "/installtunnel", result])
return (code == 0), (err or "Connecté")
else:
# 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])
code, _, err = run_privileged(["wg-quick", "up", name], timeout=90)
if code == 0:
return True, "Tunnel WireGuard activé"
# É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"
@@ -123,32 +241,78 @@ def disconnect(cfg: Config) -> tuple[bool, str]:
name = cfg.wg.get("interface_name", "wgs0")
if is_windows():
code, _, err = run_command(["wireguard", "/uninstalltunnel", name])
return (code == 0), (err or "Déconnecté")
code, _, err = run_command(["wireguard", "/uninstalltunnel", name], 30)
# Nettoyage inconditionnel : si le service reste ou qu'une règle NRPT
# survit, la résolution DNS resterait détournée vers le DNS du tunnel.
force_cleanup(cfg)
if code == 0 or not dns_util.interface_exists(name):
return True, "Déconnecté"
return False, err or "Erreur lors de la déconnexion"
err = ""
if is_connected(cfg):
code, _, err = run_privileged(["wg-quick", "down", name], timeout=60)
else:
code, _, err = run_privileged(["wg-quick", "down", name])
if code == 0:
# 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:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
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.connect((endpoint, port))
sock.close()
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:
return False, "Timeout — serveur inaccessible"
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é)"
def parse_conf_file(path: str) -> dict | None:
@@ -219,14 +383,26 @@ def export_conf_file(cfg: Config, dest_path: str) -> tuple[bool, str]:
def get_interface_bytes(iface: str) -> tuple[int, int] | None:
"""Retourne (rx_bytes, tx_bytes) depuis /proc/net/dev (Linux)."""
"""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 iface in line:
parts = line.split()
return int(parts[1]), int(parts[9])
except Exception:
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
@@ -258,54 +434,46 @@ def ping_server(host: str, timeout: int = 2) -> int | None:
def dns_leak_test(cfg: Config) -> dict:
"""
Teste si le DNS fuit en dehors du tunnel.
Retourne {"status": ok|leak|unknown, "resolvers": [...], "expected": str}
Retourne {"status": ok|leak|unknown|down, "resolvers": [...], "expected": str}
"""
expected_dns = cfg.wg.get("dns", "")
resolvers: list[str] = []
expected_dns = (cfg.wg.get("dns", "") or "").split(",")[0].strip()
# 1. Lire /etc/resolv.conf (Linux) ou ipconfig /all (Windows)
if not is_windows():
try:
with open("/etc/resolv.conf") as f:
for line in f:
if line.startswith("nameserver"):
parts = line.split()
if len(parts) >= 2:
resolvers.append(parts[1])
except Exception:
pass
else:
code, out, _ = run_command(["ipconfig", "/all"])
if code == 0:
import re
resolvers = re.findall(r"DNS Servers.*?:(.*?)(?:\n\S|\Z)",
out, re.DOTALL)
# 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
# 2. Résoudre un domaine test via dig/nslookup pour voir quel serveur répond
probe_ip: str = ""
code, out, _ = run_command(["dig", "+short", "+time=3", "whoami.akamai.net"], 6)
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 = [l.strip() for l in out.splitlines() if l.strip()]
lines = [line.strip() for line in out.splitlines() if line.strip()]
if lines:
probe_ip = lines[-1]
if not resolvers:
return {"status": "unknown", "resolvers": [], "expected": expected_dns,
"probe_ip": probe_ip}
# 3. Comparer avec le DNS configuré
if expected_dns and any(expected_dns in r for r in resolvers):
status = "ok"
elif expected_dns:
status = "leak"
else:
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,
}
@@ -363,6 +531,13 @@ def generate_server_config(
}
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 get_status_info(cfg: Config) -> dict:
name = cfg.wg.get("interface_name", "wgs0")
info = {
@@ -377,11 +552,24 @@ def get_status_info(cfg: Config) -> dict:
info["connected"] = is_connected(cfg)
return info
code, out, _ = run_command(["wg", "show", name])
if code != 0 or not out:
info["connected"] = is_connected(cfg)
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])
out = _wg_show(name)
if not out:
# Détails du pair inaccessibles sans CAP_NET_ADMIN — l'état de la
# connexion, lui, reste correct.
info["last_handshake"] = "—"
return info
info["connected"] = True
for line in out.splitlines():
line = line.strip()
if line.startswith("peer:"):
@@ -389,10 +577,9 @@ def get_status_info(cfg: Config) -> dict:
elif line.startswith("transfer:"):
parts = line.split(":", 1)[1].strip().split(",")
try:
rx = parts[0].strip().split()[0]
tx = parts[1].strip().split()[0]
info["rx_bytes"] = rx
info["tx_bytes"] = tx
# « 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:"):
app/ui/admin_window.py
+97
View File
@@ -9,6 +9,7 @@ from PyQt6.QtGui import QFont
from app.core.config import Config
from app.core import wireguard as wg_core
from app.core import dns as dns_util
from app.core import mfa as mfa_core
# CSS sombre commun à tous les onglets
@@ -465,6 +466,13 @@ class AdminWindow(QDialog):
btn_dns = QPushButton("🔍 Analyser le DNS")
btn_dns.clicked.connect(self._run_dns_test)
g2.addWidget(btn_dns)
btn_dns_fix = QPushButton("🛠️ Réparer le DNS (tunnel arrêté)")
btn_dns_fix.setToolTip(
"Retire la configuration DNS laissée par un tunnel arrêté "
"brutalement, qui empêche toute résolution de noms."
)
btn_dns_fix.clicked.connect(self._repair_dns)
g2.addWidget(btn_dns_fix)
self._dns_result = QLabel("")
self._dns_result.setWordWrap(True)
self._dns_result.setStyleSheet(
@@ -472,6 +480,24 @@ class AdminWindow(QDialog):
)
g2.addWidget(self._dns_result)
lay.addWidget(grp2)
# Droits d'exécution
grp3 = QGroupBox("🔐 Droits d'exécution")
g3 = QVBoxLayout(grp3)
g3.addWidget(QLabel(
"Sans règle sudo, chaque connexion ouvre un dialogue d'authentification."
))
btn_priv = QPushButton("🔐 Vérifier les droits")
btn_priv.clicked.connect(self._check_privileges)
g3.addWidget(btn_priv)
self._priv_result = QLabel("")
self._priv_result.setWordWrap(True)
self._priv_result.setStyleSheet(
"color: white; padding: 8px; border-radius: 6px; font-size: 11px;"
)
g3.addWidget(self._priv_result)
lay.addWidget(grp3)
lay.addStretch()
return w
@@ -948,6 +974,12 @@ class AdminWindow(QDialog):
elif status == "leak":
lines.append("⚠️ Fuite DNS potentielle !")
css = "background: rgba(231,76,60,0.2); color: #f1948a;"
elif status == "down":
lines.append("🚫 Résolution DNS hors service — utilisez « Réparer le DNS »")
css = "background: rgba(231,76,60,0.2); color: #f1948a;"
elif not r.get("tunnel_up", False):
lines.append("ℹ️ Tunnel inactif : la comparaison n'a pas de sens")
css = "color: #aed6f1;"
else:
lines.append("❓ Statut inconnu")
css = "color: #aed6f1;"
@@ -962,6 +994,71 @@ class AdminWindow(QDialog):
f"padding: 8px; border-radius: 6px; font-size: 11px; {css}"
)
def _check_privileges(self):
"""Liste les commandes privilégiées et celles qui exigeront un dialogue."""
from PyQt6.QtWidgets import QApplication
from app.utils.platform_utils import privilege_report
self._priv_result.setText("Vérification en cours…")
QApplication.processEvents()
rows = privilege_report()
if not rows:
self._priv_result.setText(
"✅ Exécution avec les privilèges root : rien à configurer."
)
css = "background: rgba(39,174,96,0.2); color: #a9dfbf;"
else:
lines = [
("✅ " if ok else "⚠️ ") + f"{name} — {role}"
+ ("" if ok else " (dialogue à chaque appel)")
for name, ok, role in rows
]
missing = [n for n, ok, _ in rows if not ok]
if missing:
lines.append("")
lines.append("Pour supprimer ces dialogues : make setup-sudoers")
css = "background: rgba(241,196,15,0.2); color: #f9e79f;"
else:
css = "background: rgba(39,174,96,0.2); color: #a9dfbf;"
self._priv_result.setText("\n".join(lines))
self._priv_result.setStyleSheet(
f"padding: 8px; border-radius: 6px; font-size: 11px; {css}"
)
def _repair_dns(self):
"""Retire la configuration DNS résiduelle d'un tunnel mal arrêté.
`wg-quick down` refuse de s'exécuter dès que l'interface a disparu et
ne retire donc jamais l'entrée DNS qu'il avait posée : la résolution
reste dirigée vers un serveur injoignable.
"""
from PyQt6.QtWidgets import QApplication
self._dns_result.setText("Réparation en cours…")
QApplication.processEvents()
actions = wg_core.force_cleanup(self._cfg)
works = dns_util.resolution_works()
if works:
msg = "✅ Résolution DNS fonctionnelle"
if actions:
msg += "\nActions : " + ", ".join(actions)
css = "background: rgba(39,174,96,0.2); color: #a9dfbf;"
elif actions:
msg = ("⚠️ Résidus retirés (" + ", ".join(actions) +
") mais la résolution échoue toujours.\n"
"Vérifiez la connexion réseau physique.")
css = "background: rgba(241,196,15,0.2); color: #f9e79f;"
else:
msg = ("🚫 Aucun résidu du tunnel détecté et la résolution échoue :"
" le problème vient du réseau, pas de WGSecure.")
css = "background: rgba(231,76,60,0.2); color: #f1948a;"
self._dns_result.setText(msg)
self._dns_result.setStyleSheet(
f"padding: 8px; border-radius: 6px; font-size: 11px; {css}"
)
def _set_admin_password(self):
p1 = self._admin_pw1.text()
p2 = self._admin_pw2.text()
app/ui/main_window.py
+193 -67
View File
@@ -10,8 +10,11 @@ from PyQt6.QtGui import QFont, QCloseEvent, QColor
from app.core.config import Config
from app.core import wireguard as wg_core
from app.core import dns as dns_util
from app.core import log as conn_log
from app.core import history as hist
from app.core import shutdown as shutdown_guard
from app.ui.worker import TaskWorker, ValueWorker
from app.ui.mfa_dialog import MFADialog
from app.ui.admin_window import AdminWindow
from app.ui.bw_graph import BandwidthGraph
@@ -47,6 +50,12 @@ class MainWindow(QMainWindow):
self._prev_tx: int | None = None
self._auto_reconnect_active = False
self._session_id: int | None = None
self._worker: TaskWorker | None = None
self._ping_worker: ValueWorker | None = None
self._is_connected = False
self._reconnect_failures = 0
self._quitting = False
self._warned_foreign: set[str] = set()
self.setWindowTitle("WGSecure")
self.setFixedSize(430, 680)
@@ -82,6 +91,7 @@ class MainWindow(QMainWindow):
self._refresh_status()
self._reload_log()
self._reload_profiles()
self._warn_foreign_interfaces()
if self._cfg.get("ui", "auto_connect_on_startup") and self._cfg.configured:
QTimer.singleShot(500, self._on_connect)
@@ -227,7 +237,7 @@ class MainWindow(QMainWindow):
self._log_list.setFixedHeight(78)
self._log_list.setStyleSheet(
f"QListWidget {{ background: {_DARK2}; color: #aed6f1;"
" border: none; font-size: 10px; font-family: Courier; }}"
" border: none; font-size: 10px; font-family: Courier; }"
"QListWidget::item { padding: 1px 6px; }"
)
self._log_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
@@ -390,47 +400,68 @@ class MainWindow(QMainWindow):
rx_bps = max(0, rx_now - self._prev_rx) / dt
tx_bps = max(0, tx_now - self._prev_tx) / dt
self._prev_rx, self._prev_tx = rx_now, tx_now
if wg_core.is_connected(self._cfg):
self._bw_graph.push(rx_bps, tx_bps)
else:
self._bw_graph.push(0, 0)
# `_is_connected` est rafraîchi par _refresh_status : inutile de
# relancer un `wg show` (un sous-processus) toutes les 3 secondes.
self._bw_graph.push(rx_bps if self._is_connected else 0,
tx_bps if self._is_connected else 0)
def _update_ping(self):
host = self._cfg.wg.get("server_endpoint", "")
if not host or not wg_core.is_connected(self._cfg):
self._ping_badge.setText("— ms")
self._ping_badge.setStyleSheet(
f"border-radius: 11px; font-size: 10px; font-weight: bold; {_PING_CSS['offline']}"
)
if not host or not self._is_connected:
self._set_ping_badge(None, "— ms", "offline")
return
ms = wg_core.ping_server(host)
if ms is None:
css = _PING_CSS["bad"]
txt = "hors ligne"
if self._ping_worker is not None and self._ping_worker.isRunning():
return
# `ping` bloque jusqu'à 4 s : exécuté dans le thread Qt, il faisait
# saccader toute l'interface toutes les 10 secondes.
self._ping_worker = ValueWorker(wg_core.ping_server, host, 2, parent=self)
self._ping_worker.done.connect(self._on_ping_done)
# Détruire le QThread depuis `finished` et non depuis `done` : `done`
# est émis avant la sortie de run(), un deleteLater() à ce moment
# provoque « QThread: Destroyed while thread is still running ».
self._ping_worker.finished.connect(self._ping_worker.deleteLater)
self._ping_worker.start()
def _on_ping_done(self, ms):
self._ping_worker = None
if not self._is_connected:
self._set_ping_badge(None, "— ms", "offline")
elif ms is None:
self._set_ping_badge(None, "hors ligne", "bad")
elif ms < 50:
css = _PING_CSS["good"]
txt = f"{ms} ms"
self._set_ping_badge(ms, f"{ms} ms", "good")
elif ms < 200:
css = _PING_CSS["medium"]
txt = f"{ms} ms"
self._set_ping_badge(ms, f"{ms} ms", "medium")
else:
css = _PING_CSS["bad"]
txt = f"{ms} ms"
self._ping_badge.setText(txt)
self._set_ping_badge(ms, f"{ms} ms", "bad")
def _set_ping_badge(self, _ms, text: str, css_key: str):
self._ping_badge.setText(text)
self._ping_badge.setStyleSheet(
f"border-radius: 11px; font-size: 10px; font-weight: bold; {css}"
"border-radius: 11px; font-size: 10px; font-weight: bold; "
+ _PING_CSS[css_key]
)
def _check_reconnect(self):
if not self._cfg.get("ui", "auto_reconnect"):
if not self._cfg.get("ui", "auto_reconnect") or self._quitting:
return
if self._worker is not None:
return # une (dé)connexion est déjà en cours
if self._cfg.mfa_enabled:
# Rouvrir une fenêtre MFA modale toutes les 30 s rendrait
# l'application inutilisable : la reconnexion doit rester manuelle.
return
if not self._cfg.configured or self._is_connected:
self._reconnect_failures = 0
return
if self._reconnect_failures >= 5:
# Arrêt du harcèlement : chaque tentative demande des privilèges.
if self._reconnect_failures == 5:
self._add_log("Auto-reconnexion suspendue après 5 échecs", "error")
self._reconnect_failures += 1
return
try:
connected = wg_core.is_connected(self._cfg)
except Exception:
connected = False
if not connected and self._cfg.configured:
self._add_log("Auto-reconnexion…", "warning")
self._on_connect()
self._start_connect(silent=True)
# ------------------------------------------------------------------ #
# Statut WireGuard
@@ -461,6 +492,9 @@ class MainWindow(QMainWindow):
except Exception:
connected = False
info = {}
self._is_connected = connected
if connected:
self._reconnect_failures = 0
if connected:
if self._connected_since is None:
@@ -519,27 +553,20 @@ class MainWindow(QMainWindow):
self._add_log("Clé publique copiée dans le presse-papiers", "info")
def _on_connect(self):
if self._connecting:
return
try:
connected = wg_core.is_connected(self._cfg)
except Exception:
connected = False
"""Bascule connexion/déconnexion. Le travail réel part dans un thread."""
if self._worker is not None:
return # opération déjà en cours
if self._is_connected or wg_core.is_connected(self._cfg):
self._start_disconnect()
else:
self._start_connect(silent=False)
if connected:
ok, msg = wg_core.disconnect(self._cfg)
if ok:
if self._session_id is not None:
rx = self._prev_rx or 0
tx = self._prev_tx or 0
hist.end_session(self._session_id, rx, tx)
self._session_id = None
self._add_log("Tunnel WireGuard désactivé", "warning")
else:
self._add_log(f"Erreur déconnexion : {msg}", "error")
QMessageBox.warning(self, "Erreur", msg)
else:
# ── Connexion ────────────────────────────────────────────────────────
def _start_connect(self, silent: bool = False):
if self._worker is not None:
return
if not self._cfg.configured:
if not silent:
QMessageBox.information(
self, "Configuration manquante",
"WireGuard n'est pas encore configuré.\n"
@@ -549,37 +576,106 @@ class MainWindow(QMainWindow):
if self._cfg.mfa_enabled:
if not self._cfg.mfa_secret:
if not silent:
QMessageBox.warning(self, "MFA non configuré",
"MFA activé mais aucun secret défini.")
return
if silent:
return # jamais de fenêtre MFA sur minuterie
dlg = MFADialog(self._cfg.mfa_secret, self)
if dlg.exec() != MFADialog.DialogCode.Accepted or not dlg.is_verified():
self._add_log("MFA annulé ou échoué", "warning")
return
self._add_log("Authentification MFA réussie ✓", "success")
self._btn_connect.setEnabled(False)
self._btn_connect.setText("Connexion…")
self._connecting = True
QApplication.processEvents()
self._begin_task("Connexion…")
shutdown_guard.arm() # le tunnel devra être rendu à la sortie
self._worker = TaskWorker(wg_core.connect, self._cfg,
tag="connect" if not silent else "connect-silent",
parent=self)
self._worker.done.connect(self._on_task_done)
self._worker.finished.connect(self._worker.deleteLater)
self._worker.start()
ok, msg = wg_core.connect(self._cfg)
# ── Déconnexion ──────────────────────────────────────────────────────
def _start_disconnect(self):
if self._worker is not None:
return
self._begin_task("Déconnexion…")
self._worker = TaskWorker(wg_core.disconnect, self._cfg,
tag="disconnect", parent=self)
self._worker.done.connect(self._on_task_done)
self._worker.finished.connect(self._worker.deleteLater)
self._worker.start()
def _begin_task(self, label: str):
self._connecting = True
self._btn_connect.setEnabled(False)
self._btn_connect.setText(label)
def _on_task_done(self, ok: bool, msg: str, tag: str):
self._worker = None
self._connecting = False
self._btn_connect.setEnabled(True)
silent = tag.endswith("-silent")
if tag.startswith("connect"):
if ok:
self._reconnect_failures = 0
self._connected_since = QDateTime.currentDateTime()
endpoint = self._cfg.wg.get("server_endpoint", "")
profile = self._cfg.active_profile
self._session_id = hist.start_session(endpoint, profile)
self._session_id = hist.start_session(endpoint,
self._cfg.active_profile)
self._add_log(f"Connexion établie → {endpoint}", "success")
QTimer.singleShot(5000, self._update_ping)
else:
self._reconnect_failures += 1
self._add_log(f"Erreur : {msg}", "error")
if not silent:
QMessageBox.warning(self, "Erreur de connexion", msg)
else: # disconnect
if ok:
if self._session_id is not None:
hist.end_session(self._session_id,
self._prev_rx or 0, self._prev_tx or 0)
self._session_id = None
self._add_log("Tunnel WireGuard désactivé", "warning")
else:
self._add_log(f"Erreur déconnexion : {msg}", "error")
if not silent:
QMessageBox.warning(self, "Erreur", msg)
self._check_dns_health()
self._warn_foreign_interfaces()
self._refresh_status()
def _warn_foreign_interfaces(self):
"""Signale un tunnel WireGuard monté hors de WGSecure.
Il impose son propre DNS au système et WGSecure ne le démontera pas :
c'est une cause fréquente de « plus d'accès internet » après coup.
"""
try:
foreign = wg_core.foreign_wg_interfaces(self._cfg)
except Exception:
return
for iface in foreign:
if iface in self._warned_foreign:
continue
self._warned_foreign.add(iface)
self._add_log(
f"Tunnel « {iface} » actif hors de WGSecure "
f"(interface configurée : {self._cfg.wg.get('interface_name','wgs0')})",
"warning",
)
def _check_dns_health(self):
"""Signale — et répare — une résolution DNS restée détournée."""
iface = self._cfg.wg.get("interface_name", "wgs0")
repaired, msg = dns_util.repair_if_broken(iface, self._cfg.wg.get("dns", ""))
if msg:
self._add_log(msg, "warning" if repaired else "error")
def _open_history(self):
HistoryDialog(self).exec()
@@ -600,19 +696,49 @@ class MainWindow(QMainWindow):
self._reload_profiles()
self._refresh_status()
# ------------------------------------------------------------------ #
# Fermeture — restitution de l'état réseau
# ------------------------------------------------------------------ #
def prepare_quit(self) -> None:
"""Arrête minuteries et threads, puis rend le réseau à son état initial.
Appelé par toutes les voies de sortie (croix de la fenêtre, « Quitter »
du systray, signal système). Synchrone : la restauration doit être
terminée avant que le processus ne disparaisse, sinon la configuration
DNS du tunnel survit à l'application et la machine perd la résolution
de noms.
"""
if self._quitting:
return
self._quitting = True
for timer in (self._status_timer, self._clock_timer, self._bw_timer,
self._ping_timer, self._reconnect_timer):
timer.stop()
# Laisser les threads en cours se terminer avant de démonter le tunnel :
# une connexion et une déconnexion simultanées laisseraient un état mixte.
for worker in (self._worker, self._ping_worker):
try:
if worker is not None and worker.isRunning():
worker.wait(15000)
except RuntimeError:
pass # objet Qt déjà détruit
self._worker = None
self._ping_worker = None
if self._session_id is not None:
hist.end_session(self._session_id, self._prev_rx or 0,
self._prev_tx or 0)
self._session_id = None
shutdown_guard.restore_network("fermeture de l'application")
def closeEvent(self, event: QCloseEvent):
if self._cfg.get("ui", "minimize_to_tray"):
if self._cfg.get("ui", "minimize_to_tray") and not self._quitting:
# Simple masquage : ce n'est pas une fermeture, le tunnel reste actif.
event.ignore()
self.hide()
else:
self._status_timer.stop()
self._clock_timer.stop()
self._bw_timer.stop()
self._ping_timer.stop()
self._reconnect_timer.stop()
try:
if wg_core.is_connected(self._cfg):
wg_core.disconnect(self._cfg)
except Exception:
pass
return
self.prepare_quit()
event.accept()
app/ui/systray.py
+8 -11
View File
@@ -104,18 +104,15 @@ class SystemTray(QSystemTrayIcon):
def _toggle_connection(self):
self._win._on_connect()
self._update_status()
# La (dé)connexion est désormais asynchrone : l'icône sera mise à jour
# par la scrutation périodique une fois l'opération terminée.
def _quit(self):
self._poll.stop()
try:
if wg_core.is_connected(self._cfg):
wg_core.disconnect(self._cfg)
except Exception:
pass
self._win._status_timer.stop()
self._win._clock_timer.stop()
self._win._bw_timer.stop()
self._win._ping_timer.stop()
self._win._reconnect_timer.stop()
self.hide()
# prepare_quit() arrête les minuteries, attend les threads en cours,
# démonte le tunnel et retire la configuration DNS qu'il avait posée.
# Sans cela, quitter en étant connecté laissait la machine sans
# résolution de noms.
self._win.prepare_quit()
QApplication.quit()
app/ui/worker.py
+54
View File
@@ -0,0 +1,54 @@
"""Exécution de tâches bloquantes hors du thread graphique.
`wg-quick up/down` passe par sudo ou pkexec : entre l'attente du dialogue
d'authentification et le montage du tunnel, l'appel peut durer plusieurs
dizaines de secondes. Lancé depuis le thread Qt, il gèle toute l'interface
(fenêtre grise, « ne répond pas »). Ces workers déportent l'appel.
"""
from __future__ import annotations
from typing import Any, Callable
from PyQt6.QtCore import QThread, pyqtSignal
class TaskWorker(QThread):
"""Exécute `fn(*args)` dans un thread et émet son résultat.
`done` porte (ok, message, tag) ; `tag` permet à l'appelant de savoir de
quelle tâche il s'agit lorsqu'un même slot en traite plusieurs.
"""
done = pyqtSignal(bool, str, str)
def __init__(self, fn: Callable[..., tuple[bool, str]], *args: Any,
tag: str = "", parent=None):
super().__init__(parent)
self._fn = fn
self._args = args
self._tag = tag
def run(self) -> None:
try:
ok, msg = self._fn(*self._args)
except Exception as e: # noqa: BLE001 — remonté à l'UI
ok, msg = False, str(e)
self.done.emit(bool(ok), str(msg), self._tag)
class ValueWorker(QThread):
"""Variante pour une fonction qui retourne une valeur quelconque."""
done = pyqtSignal(object)
def __init__(self, fn: Callable[..., Any], *args: Any, parent=None):
super().__init__(parent)
self._fn = fn
self._args = args
def run(self) -> None:
try:
result = self._fn(*self._args)
except Exception: # noqa: BLE001
result = None
self.done.emit(result)
app/utils/platform_utils.py
+108 -24
View File
@@ -39,13 +39,15 @@ def has_root_privileges() -> bool:
return os.geteuid() == 0
def run_command(cmd: list[str], timeout: int = 10) -> tuple[int, str, str]:
def run_command(cmd: list[str], timeout: int = 10,
env: dict | None = None) -> tuple[int, str, str]:
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
env=env,
)
return result.returncode, result.stdout.strip(), result.stderr.strip()
except subprocess.TimeoutExpired:
@@ -56,46 +58,86 @@ def run_command(cmd: list[str], timeout: int = 10) -> tuple[int, str, str]:
return -1, "", str(e)
def _c_locale_env() -> dict:
"""Environnement forçant les messages système en anglais.
Sans cela, la détection « sudo demande un mot de passe » échoue sur une
session en français (« mot de passe » au lieu de « password »), et l'appel
repart avec le code d'erreur de sudo au lieu de basculer sur pkexec.
"""
env = os.environ.copy()
env["LC_ALL"] = "C"
env["LANG"] = "C"
return env
# Messages que sudo émet quand il refuse *avant* de lancer la commande.
# Comparés en anglais : _c_locale_env() force LC_ALL=C pour que cette
# détection ne dépende pas de la langue de la session.
_SUDO_AUTH_MARKERS = (
"a password is required",
"no askpass program specified",
"no tty present",
"must have a tty",
"a terminal is required",
"is not allowed to run",
"may not run sudo",
"authentication failure",
)
def _sudo_refused_auth(err: str) -> bool:
"""True si sudo a refusé faute d'authentification, sans lancer la commande.
Distinguer ce cas d'un échec de la commande elle-même est indispensable :
dans le premier, il faut basculer sur askpass/pkexec ; dans le second, il
faut remonter l'erreur telle quelle à l'utilisateur.
Interroger `sudo -l` ne convient pas : pour un compte membre du groupe
sudo (`(ALL : ALL) ALL`), `sudo -n -l -- <cmd>` répond « autorisé » pour
*toutes* les commandes, y compris celles qui exigeront un mot de passe.
"""
low = err.lower()
return any(marker in low for marker in _SUDO_AUTH_MARKERS)
def run_privileged(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
if is_windows():
return run_command(cmd, timeout)
if has_root_privileges():
return run_command(cmd, timeout)
# 1. sudo -n (NOPASSWD configuré dans sudoers — sans aucun dialogue)
code, out, err = run_command(["sudo", "-n"] + cmd, timeout)
# 1. sudo -n : réussit sans aucun dialogue si une règle NOPASSWD couvre la
# commande (cf. `make setup-sudoers`).
code, out, err = run_command(["sudo", "-n"] + cmd, timeout,
env=_c_locale_env())
if code == 0:
return code, out, err
# Si sudo a tourné mais la commande a échoué (pas un problème d'auth), retourner l'erreur
if code != 1 or ("password" not in err.lower() and "passwd" not in err.lower()):
if "sudo:" not in err.lower() and code not in (-1,):
if not _sudo_refused_auth(err):
# sudo a bien lancé la commande : c'est elle qui a échoué. Inutile de
# redemander des privilèges, l'erreur est remontée telle quelle.
return code, out, err
# 2. sudo avec programme askpass graphique (pas de TTY dans une app Qt)
askpass = _find_askpass()
if askpass:
env = os.environ.copy()
env = _c_locale_env()
env["SUDO_ASKPASS"] = askpass
try:
result = subprocess.run(
["sudo", "-A"] + cmd,
capture_output=True, text=True,
timeout=timeout, env=env,
)
if result.returncode == 0:
return result.returncode, result.stdout.strip(), result.stderr.strip()
except (subprocess.TimeoutExpired, Exception):
pass
code, out, err = run_command(["sudo", "-A"] + cmd, timeout, env=env)
if code == 0:
return code, out, err
# 3. pkexec — dialogue graphique polkit (GNOME/KDE)
code, out, err = run_command(["pkexec"] + cmd, timeout)
if code != -1 or "introuvable" not in err:
code, out, err = run_command(["pkexec"] + cmd, timeout, env=_c_locale_env())
# 126 = authentification refusée/annulée, 127 = pkexec absent : on préfère
# le message d'aide au code opaque.
if code not in (-1, 126, 127):
return code, out, err
return (
-1, "",
"Élévation de privilèges impossible.\n"
"Exécutez make setup-sudoers pour configurer wg-quick sans dialogue.\n"
"Exécutez make setup-sudoers pour autoriser WGSecure sans dialogue.\n"
"Ou manuellement :\n"
f" sudo bash -c \"echo '{os.environ.get('USER','<user>')} ALL=(ALL) NOPASSWD: /usr/bin/wg-quick'"
" > /etc/sudoers.d/wgsecure && chmod 440 /etc/sudoers.d/wgsecure\""
@@ -123,6 +165,35 @@ def _find_askpass() -> str:
return ""
# Commandes que WGSecure lance avec des privilèges, et ce qu'elles servent.
_PRIVILEGED_COMMANDS = (
(["wg-quick", "--help"], "monter / démonter le tunnel"),
(["install", "--version"], "écrire la config dans /etc/wireguard"),
(["ip", "-V"], "supprimer une interface orpheline"),
(["resolvconf", "--version"], "retirer l'entrée DNS du tunnel"),
(["resolvectl", "--version"], "purger le DNS systemd-resolved"),
)
def privilege_report() -> list[tuple[str, bool, str]]:
"""État des droits pour chaque commande privilégiée utilisée.
Retourne [(commande, sans_mot_de_passe, rôle)]. Une commande absente du
système est ignorée. Sert à expliquer pourquoi un dialogue polkit apparaît
à chaque connexion plutôt que de laisser l'utilisateur deviner.
"""
import shutil as _sh
report: list[tuple[str, bool, str]] = []
if is_windows() or has_root_privileges():
return report
for probe, role in _PRIVILEGED_COMMANDS:
if not _sh.which(probe[0]):
continue
code, _, err = run_command(["sudo", "-n"] + probe, 8, env=_c_locale_env())
report.append((probe[0], not _sudo_refused_auth(err), role))
return report
def wg_available() -> bool:
code, _, _ = run_command(["wg", "--version"])
return code == 0
@@ -136,6 +207,22 @@ def wg_quick_available() -> bool:
return code == 0
def launch_command() -> str:
"""Commande à écrire dans .desktop / registre Run pour relancer l'appli.
Dans un binaire PyInstaller, `main.py` n'est pas déployé : écrire
« python3 <chemin>/main.py » produisait un lanceur et un démarrage
automatique qui ne fonctionnaient jamais depuis un paquet installé.
"""
if getattr(sys, "frozen", False):
return f'"{sys.executable}"' if " " in sys.executable else sys.executable
script = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "main.py")
)
exe = "pythonw" if is_windows() else "python3"
return f'{exe} "{script}"' if " " in script else f"{exe} {script}"
def install_desktop_entry(icon_png_path: str) -> None:
"""Installe l'entrée .desktop et l'icône PNG pour la barre des tâches GNOME/KDE.
Appelé une fois au démarrage ; sans effet sur Windows."""
@@ -152,16 +239,13 @@ def install_desktop_entry(icon_png_path: str) -> None:
# Fichier .desktop
app_dir = os.path.expanduser("~/.local/share/applications")
os.makedirs(app_dir, exist_ok=True)
script = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "main.py")
)
desktop = os.path.join(app_dir, "wgsecure.desktop")
content = (
"[Desktop Entry]\n"
"Type=Application\n"
"Name=WGSecure\n"
"Comment=WireGuard GUI avec MFA TOTP\n"
f"Exec=python3 {script}\n"
f"Exec={launch_command()}\n"
"Icon=wgsecure\n"
"Categories=Network;Security;\n"
"StartupWMClass=main\n"
main.py
+29
View File
@@ -10,6 +10,9 @@ from PyQt6.QtWidgets import QApplication, QMessageBox, QInputDialog, QLineEdit,
from PyQt6.QtCore import Qt
from app.core.config import Config
from app.core import dns as dns_util
from app.core import log as conn_log
from app.core import shutdown as shutdown_guard
from app.ui.main_window import MainWindow
from app.ui.systray import SystemTray
from app.ui import icons
@@ -27,6 +30,23 @@ def _install_taskbar_icon(app_icon) -> None:
pass
def recover_dns_state(config: Config) -> None:
"""Répare la résolution DNS laissée détournée par une session précédente.
Si l'application (ou la machine) s'est arrêtée sans exécuter
`wg-quick down`, la configuration DNS du tunnel reste active alors que le
résolveur correspondant est injoignable : plus aucun nom ne se résout.
Le nettoyage n'a lieu que si un résidu est effectivement constaté.
"""
iface = config.wg.get("interface_name", "wgs0")
try:
repaired, msg = dns_util.repair_if_broken(iface, config.wg.get("dns", ""))
except Exception as e: # noqa: BLE001
repaired, msg = False, f"Vérification DNS impossible : {e}"
if msg:
conn_log.log_event(msg, "warning" if repaired else "error")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="wgsecure",
@@ -85,6 +105,15 @@ def main():
config = Config()
# Reste d'une session précédente mal terminée : DNS détourné vers un
# résolveur mort. À faire avant d'ouvrir la fenêtre pour rendre l'accès
# au réseau tout de suite.
recover_dns_state(config)
# Restauration de l'état réseau garantie sur toutes les sorties :
# Quitter, croix, SIGINT/SIGTERM, atexit, exception non rattrapée.
shutdown_guard.install(config, report=conn_log.log_event, app=app)
# Mode Admin direct (--admin)
if args.admin:
if not require_admin_auth(config, app):