chore(release): v0.4.5

fix: la clé pré-partagée n'était écrite dans aucune config client

L'assistant de génération produisait une PresharedKey affichée à l'écran mais
stockée nulle part, et l'import d'un .conf ignorait la ligne. Face à un serveur
qui en attend une, le tunnel montait et l'interface existait — sans qu'aucun
handshake n'aboutisse jamais. La PSK devient un champ à part entière.

fix: options de ligne de commande wireguard.exe incorrectes sous Windows

/installtunnel et /uninstalltunnel n'existent pas ; les vrais verbes sont
/installtunnelservice et /uninstalltunnelservice. S'y ajoutaient l'absence de
résolution du chemin d'installation (WireGuard n'est pas dans le PATH),
l'absence totale d'élévation UAC alors que ces commandes l'exigent, et une
détection d'installation qui sondait une application graphique avec --help.

feat: diagnostic de bout en bout de la chaîne de connexion

Le bouton de test s'arrêtait à la joignabilité UDP et concluait « serveur
joignable » sur un tunnel qui n'échangeait rien. Il déroule désormais quatorze
étapes, de la configuration locale jusqu'au DNS dans le tunnel, et désigne
l'étape qui bloque.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tuxgyver
2026-08-31 15:41:43 +02:00
co-authored by Claude Opus 5
parent 09724f4e4f
commit 9d785c00cc
10 changed files with 777 additions and 64 deletions
+56
View File
@@ -6,6 +6,62 @@ Ce projet suit le [Versionnage Sémantique](https://semver.org/lang/fr/).
--- ---
## [0.4.5] — 2026-08-31
### Corrigé
- **Tunnel « connecté » sans jamais négocier de handshake** : la clé pré-partagée (PresharedKey)
n'était écrite dans aucune configuration client. L'assistant « Générer config serveur + client »
en produisait une, visible à l'écran, mais impossible à enregistrer ; et l'import d'un fichier
`.conf` l'ignorait purement et simplement. Face à un serveur qui en attend une, le tunnel montait,
l'interface existait, l'application affichait « Connecté » — et pas un paquet ne passait. La clé
pré-partagée est désormais un champ à part entière : saisie, importée, exportée et écrite dans la
configuration du tunnel.
- **Latence du serveur jamais mesurée hors système anglophone** : la sortie de `ping` est traduite
(« temps= » en français), et seule la forme anglaise était reconnue. Un serveur qui répondait
correctement était rapporté comme injoignable.
### Ajouté
- **Diagnostic complet de la chaîne de connexion** : le bouton « Lancer le test de connexion »
ne se contentait plus d'un simple test UDP. Il déroule maintenant toute la chaîne et désigne
l'étape exacte qui bloque : présence de WireGuard, complétude de la configuration, cohérence de
la paire de clés du client, clé pré-partagée, résolution du nom du serveur, route empruntée,
cohérence des « IPs autorisées », ping ICMP, port UDP, montage du tunnel, handshake, trafic
chiffré, joignabilité du pair distant et résolution DNS dans le tunnel.
- **Détection des deux pièges de routage classiques** : l'adresse du serveur incluse dans les
« IPs autorisées » (le tunnel se coupe l'herbe sous le pied dès qu'il monte) et le réseau local
couvert par ces mêmes plages (imprimante, NAS et box basculent dans le VPN).
- **Champ MTU** : réglable, laissé automatique par défaut, importé et exporté avec les fichiers `.conf`.
- **Bouton « Appliquer le volet client »** dans l'assistant de génération : reporte clés et clé
pré-partagée directement dans la configuration, au lieu de les laisser dans une fenêtre à recopier.
### Modifié
- **Statistiques du tunnel** : le dernier handshake est lu via `wg show latest-handshakes`, ce qui
permet d'en donner l'ancienneté exacte et de signaler un pair qui a cessé de répondre.
## [0.4.4] — 2026-08-31
### Corrigé
- **Tunnel impossible à monter sous Windows** : les commandes envoyées à `wireguard.exe`
utilisaient des options qui n'existent pas (`/installtunnel`, `/uninstalltunnel`). WireGuard
affichait sa boîte d'aide et sortait sans rien faire : le bouton « Connecter » ne montait jamais
le tunnel. Les vraies options `/installtunnelservice` et `/uninstalltunnelservice` sont
désormais utilisées.
- **WireGuard déclaré absent sous Windows** : l'installateur WireGuard n'ajoute pas son
répertoire au PATH, et l'application cherchait `wireguard.exe` et `wg.exe` sans chemin. Ils sont
maintenant localisés dans `C:\Program Files\WireGuard`.
- **« Accès refusé » silencieux sous Windows** : installer ou retirer un tunnel exige les droits
administrateur, qui n'étaient jamais demandés. Une invite UAC apparaît désormais au moment
voulu, et l'annulation de cette invite est signalée clairement.
- **Détection de WireGuard toujours négative sous Windows** : le test d'installation lançait
`wireguard --help`, une option inexistante sur une application graphique. La présence du
logiciel est maintenant vérifiée sur le disque.
- **Échec annoncé sur un tunnel qui montait** : le service du tunnel démarre de façon asynchrone
et l'état était vérifié trop tôt. L'application attend désormais son démarrage effectif.
### Ajouté
- **Statistiques du tunnel sous Windows** : volumes envoyés/reçus, pair et dernier handshake,
jusqu'ici affichés vides.
## [0.4.3] — 2026-08-31 ## [0.4.3] — 2026-08-31
### Corrigé ### Corrigé
+1 -1
View File
@@ -4,7 +4,7 @@
## ────────────────────────────────────────────── ## ──────────────────────────────────────────────
APP := wgsecure APP := wgsecure
VERSION := 0.4.3 VERSION := 0.4.5
VENV := .venv VENV := .venv
PYTHON := $(VENV)/bin/python3 PYTHON := $(VENV)/bin/python3
PIP := $(VENV)/bin/pip PIP := $(VENV)/bin/pip
+2 -2
View File
@@ -1,6 +1,6 @@
# 🛡️ WGSecure (WGS) # 🛡️ WGSecure (WGS)
🚀 v0.4.3 · 🐍 · 🪟🐧 · 🔐 MFA TOTP · 🛡️ WireGuard 🚀 v0.4.5 · 🐍 · 🪟🐧 · 🔐 MFA TOTP · 🛡️ WireGuard
**WGSecure** est une interface graphique multiplateforme (Windows & Linux) pour gérer une connexion WireGuard avec une surcouche MFA (authentification à deux facteurs TOTP). **WGSecure** est une interface graphique multiplateforme (Windows & Linux) pour gérer une connexion WireGuard avec une surcouche MFA (authentification à deux facteurs TOTP).
@@ -188,4 +188,4 @@ WGSecure/
## 👤 Auteur ## 👤 Auteur
Développé par **Johnny** — [JT-Tools](https://github.com/JT-Tools) Développé par **Johnny** — [JT-Tools](https://github.com/JT-Tools)
Version : ![v0.4.3](https://img.shields.io/badge/v0.4.3-ao%C3%BBt%202026-2980b9?style=flat-square) — Août 2026 Version : ![v0.4.5](https://img.shields.io/badge/v0.4.5-ao%C3%BBt%202026-2980b9?style=flat-square) — Août 2026
+1 -1
View File
@@ -1,3 +1,3 @@
__version__ = "0.4.3" __version__ = "0.4.5"
APP_NAME = "WGSecure" APP_NAME = "WGSecure"
APP_SHORT = "WGS" APP_SHORT = "WGS"
+3 -1
View File
@@ -15,16 +15,18 @@ _WG_DEFAULT: dict[str, Any] = {
"server_endpoint": "", "server_endpoint": "",
"server_port": 51820, "server_port": 51820,
"server_public_key": "", "server_public_key": "",
"preshared_key": "",
"client_private_key": "", "client_private_key": "",
"client_public_key": "", "client_public_key": "",
"client_address": "10.8.0.2/24", "client_address": "10.8.0.2/24",
"dns": "1.1.1.1", "dns": "1.1.1.1",
"allowed_ips": "10.8.0.0/24", "allowed_ips": "10.8.0.0/24",
"keepalive": 25, "keepalive": 25,
"mtu": 0,
} }
_DEFAULT: dict[str, Any] = { _DEFAULT: dict[str, Any] = {
"version": "0.4.3", "version": "0.4.5",
"admin_password_hash": "", "admin_password_hash": "",
"admin_salt": "", "admin_salt": "",
"mfa_enabled": False, "mfa_enabled": False,
+470 -36
View File
@@ -1,4 +1,5 @@
import base64 import base64
import ipaddress
import os import os
import socket import socket
import time import time
@@ -7,11 +8,15 @@ from app.core import dns as dns_util
from app.core.config import Config from app.core.config import Config
from app.utils.platform_utils import ( from app.utils.platform_utils import (
get_config_dir, get_config_dir,
has_root_privileges,
get_wg_config_dir, get_wg_config_dir,
is_windows, is_windows,
run_command, run_command,
run_privileged, run_privileged,
wg_available, wg_available,
wg_exe,
wg_quick_available,
wireguard_exe,
) )
@@ -44,10 +49,23 @@ def build_client_config(cfg: Config) -> str:
dns_value = (wg.get("dns") or "").strip() dns_value = (wg.get("dns") or "").strip()
if dns_value and dns_util.has_dns_backend(): if dns_value and dns_util.has_dns_backend():
lines.append(f"DNS = {dns_value}") lines.append(f"DNS = {dns_value}")
mtu = int(wg.get("mtu") or 0)
if mtu:
lines.append(f"MTU = {mtu}")
lines += [ lines += [
"", "",
"[Peer]", "[Peer]",
f"PublicKey = {wg['server_public_key']}", f"PublicKey = {wg['server_public_key']}",
]
# Sans cette ligne, un serveur configuré avec une clé pré-partagée rejette
# silencieusement toutes nos tentatives : le service monte, l'interface
# existe, l'application affiche « Connecté » — et le handshake n'aboutit
# jamais. La PSK était générée par l'assistant puis perdue, ni écrite dans
# la config client ni relue à l'import d'un .conf.
psk = (wg.get("preshared_key") or "").strip()
if psk:
lines.append(f"PresharedKey = {psk}")
lines += [
f"AllowedIPs = {wg['allowed_ips']}", f"AllowedIPs = {wg['allowed_ips']}",
f"Endpoint = {wg['server_endpoint']}:{wg['server_port']}", f"Endpoint = {wg['server_endpoint']}:{wg['server_port']}",
f"PersistentKeepalive = {wg['keepalive']}", f"PersistentKeepalive = {wg['keepalive']}",
@@ -119,7 +137,7 @@ def _wg_show(name: str) -> str:
Un simple test `code == 0` conclut donc « déconnecté » pour tout Un simple test `code == 0` conclut donc « déconnecté » pour tout
utilisateur non root — c'était le cas de l'application entière. utilisateur non root — c'était le cas de l'application entière.
""" """
code, out, err = run_command(["wg", "show", name]) code, out, err = run_command([wg_exe(), "show", name])
if code != 0 or not out: if code != 0 or not out:
return "" return ""
if "not permitted" in err.lower() or "unable to access" in err.lower(): if "not permitted" in err.lower() or "unable to access" in err.lower():
@@ -178,7 +196,8 @@ def force_cleanup(cfg: Config) -> list[str]:
# bloqué, on le supprime avant de purger les règles NRPT résiduelles. # bloqué, on le supprime avant de purger les règles NRPT résiduelles.
if dns_util.interface_exists(name): if dns_util.interface_exists(name):
svc = f"WireGuardTunnel${name}" svc = f"WireGuardTunnel${name}"
run_command(["wireguard", "/uninstalltunnel", name], 30) run_privileged([wireguard_exe(), "/uninstalltunnelservice", name],
timeout=60)
if dns_util.interface_exists(name): if dns_util.interface_exists(name):
run_privileged(["sc", "stop", svc], timeout=30) run_privileged(["sc", "stop", svc], timeout=30)
run_privileged(["sc", "delete", svc], timeout=30) run_privileged(["sc", "delete", svc], timeout=30)
@@ -202,6 +221,17 @@ def force_cleanup(cfg: Config) -> list[str]:
return actions return actions
def _wait_connected(cfg: Config, timeout: int = 15) -> bool:
"""Attend que le service du tunnel passe à RUNNING, ou expire."""
deadline = time.monotonic() + timeout
while True:
if is_connected(cfg):
return True
if time.monotonic() >= deadline:
return False
time.sleep(0.5)
def connect(cfg: Config) -> tuple[bool, str]: def connect(cfg: Config) -> tuple[bool, str]:
if not cfg.configured: if not cfg.configured:
return False, "WireGuard non configuré. Ouvrez le panneau Admin." return False, "WireGuard non configuré. Ouvrez le panneau Admin."
@@ -213,8 +243,28 @@ def connect(cfg: Config) -> tuple[bool, str]:
name = cfg.wg.get("interface_name", "wgs0") name = cfg.wg.get("interface_name", "wgs0")
if is_windows(): if is_windows():
code, _, err = run_command(["wireguard", "/installtunnel", result]) # Verbes réels de wireguard.exe : /installtunnelservice <chemin .conf>
return (code == 0), (err or "Connecté") # et /uninstalltunnelservice <nom>. Les formes /installtunnel et
# /uninstalltunnel n'existent pas — wireguard.exe affichait sa boîte
# d'aide et sortait sans jamais monter le tunnel.
#
# Le nom du tunnel est le nom du fichier sans extension : il doit
# correspondre à interface_name pour que is_connected() retrouve le
# service WireGuardTunnel$<nom>.
if is_connected(cfg):
return True, "Tunnel WireGuard déjà actif"
if dns_util.interface_exists(name):
force_cleanup(cfg)
code, _, err = run_privileged(
[wireguard_exe(), "/installtunnelservice", result], timeout=90
)
# /installtunnelservice enregistre le service et rend la main : le
# démarrage effectif du tunnel est asynchrone. Tester l'état
# immédiatement conclurait « échec » sur un tunnel qui monte.
if code == 0 and _wait_connected(cfg, timeout=15):
return True, "Tunnel WireGuard activé"
force_cleanup(cfg)
return False, err or f"Échec de l'installation du tunnel (code {code})"
# Déjà monté et fonctionnel : rien à faire. # Déjà monté et fonctionnel : rien à faire.
if is_connected(cfg): if is_connected(cfg):
@@ -241,7 +291,9 @@ def disconnect(cfg: Config) -> tuple[bool, str]:
name = cfg.wg.get("interface_name", "wgs0") name = cfg.wg.get("interface_name", "wgs0")
if is_windows(): if is_windows():
code, _, err = run_command(["wireguard", "/uninstalltunnel", name], 30) code, _, err = run_privileged(
[wireguard_exe(), "/uninstalltunnelservice", name], timeout=60
)
# Nettoyage inconditionnel : si le service reste ou qu'une règle NRPT # 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. # survit, la résolution DNS resterait détournée vers le DNS du tunnel.
force_cleanup(cfg) force_cleanup(cfg)
@@ -315,6 +367,366 @@ def test_connection(cfg: Config, timeout: int = 5) -> tuple[bool, str]:
return True, f"Serveur joignable ({ms} ms, port UDP {port} non rejeté)" return True, f"Serveur joignable ({ms} ms, port UDP {port} non rejeté)"
# ── Diagnostic de bout en bout ────────────────────────────────────────────
# États d'une étape : ok = validée, warn = suspecte mais non bloquante,
# fail = point de rupture, skip = non évaluable (dépend d'une étape échouée).
_OK, _WARN, _FAIL, _SKIP = "ok", "warn", "fail", "skip"
def _step(state: str, label: str, detail: str = "") -> dict:
return {"state": state, "label": label, "detail": detail}
def last_handshake_age(name: str) -> int | None:
"""Secondes écoulées depuis le dernier handshake.
Retourne -1 si aucun handshake n'a jamais eu lieu, None si l'information
est inaccessible (privilèges insuffisants, `wg` absent).
"""
code, out, _ = run_command([wg_exe(), "show", name, "latest-handshakes"], 10)
if code != 0 or not out.strip():
return None
newest = 0
for line in out.splitlines():
parts = line.split()
if len(parts) >= 2:
try:
newest = max(newest, int(parts[-1]))
except ValueError:
continue
if newest == 0:
return -1
return max(0, int(time.time()) - newest)
def _networks(value: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
"""Découpe une liste « a/24, b/16 » en réseaux, en ignorant les entrées
invalides."""
nets = []
for part in (value or "").split(","):
part = part.strip()
if not part:
continue
try:
nets.append(ipaddress.ip_network(part, strict=False))
except ValueError:
continue
return nets
def _tunnel_gateway(client_address: str, allowed_ips: str = "") -> str:
"""Première IP utilisable du réseau du tunnel (typiquement le serveur).
Une adresse client en /32 — la forme la plus courante — ne décrit aucun
réseau : on se replie alors sur le premier réseau des « IPs autorisées »,
qui contient le pair distant.
"""
try:
iface = ipaddress.ip_interface(client_address.split(",")[0].strip())
except ValueError:
return ""
candidates = []
if iface.network.prefixlen < iface.network.max_prefixlen - 1:
candidates.append(iface.network)
candidates += [n for n in _networks(allowed_ips)
if n.prefixlen < n.max_prefixlen - 1
and n.version == iface.ip.version]
for net in candidates:
for host in net.hosts():
if host != iface.ip:
return str(host)
return ""
def _local_route_source(ip: str, port: int) -> str:
"""IP source que le système emploierait pour joindre `ip`.
Aucun paquet n'est émis : `connect()` sur un socket UDP ne fait que fixer
la route. Sert à détecter le cas où l'endpoint du serveur est lui-même
routé dans le tunnel — le tunnel a alors besoin de lui-même pour se
maintenir et le handshake s'arrête net après la connexion.
"""
try:
family = socket.AF_INET6 if ":" in ip else socket.AF_INET
with socket.socket(family, socket.SOCK_DGRAM) as sock:
sock.connect((ip, port or 51820))
return sock.getsockname()[0]
except OSError:
return ""
def _diag_config(cfg: Config) -> list[dict]:
wg = cfg.wg
steps: list[dict] = []
missing = [
label for label, key in (
("adresse du serveur", "server_endpoint"),
("clé publique du serveur", "server_public_key"),
("clé privée du client", "client_private_key"),
("adresse IP du client", "client_address"),
) if not (wg.get(key) or "").strip()
]
if missing:
steps.append(_step(_FAIL, "Configuration",
"Champs manquants : " + ", ".join(missing)))
return steps
steps.append(_step(_OK, "Configuration", "Tous les champs requis sont remplis"))
# Clé publique client cohérente avec la clé privée : une clé publique
# obsolète est déclarée au serveur et le handshake est refusé.
try:
priv = X25519PrivateKey.from_private_bytes(
base64.b64decode(wg["client_private_key"])
)
derived = base64.b64encode(
priv.public_key().public_bytes_raw()
).decode()
except Exception:
steps.append(_step(_FAIL, "Clé privée du client",
"Clé illisible : ce n'est pas une clé X25519 base64"))
return steps
declared = (wg.get("client_public_key") or "").strip()
if declared and declared != derived:
steps.append(_step(
_FAIL, "Paire de clés du client",
f"La clé publique enregistrée ne correspond pas à la clé privée. "
f"Le serveur doit connaître : {derived}"))
else:
steps.append(_step(_OK, "Paire de clés du client",
f"Clé publique déclarée au serveur : {derived[:20]}"))
psk = (wg.get("preshared_key") or "").strip()
if psk:
try:
valid = len(base64.b64decode(psk)) == 32
except Exception:
valid = False
steps.append(_step(_OK if valid else _FAIL, "Clé pré-partagée",
"Présente (doit être identique côté serveur)" if valid
else "Format invalide : 32 octets en base64 attendus"))
else:
steps.append(_step(_WARN, "Clé pré-partagée",
"Absente — à renseigner si le serveur en attend une, "
"sinon le handshake n'aboutira jamais"))
return steps
def _diag_reachability(cfg: Config, timeout: int) -> tuple[list[dict], str]:
"""Étapes réseau jusqu'au serveur. Retourne (étapes, ip_résolue)."""
wg = cfg.wg
endpoint = (wg.get("server_endpoint") or "").strip()
port = int(wg.get("server_port", 51820))
steps: list[dict] = []
try:
infos = socket.getaddrinfo(endpoint, port, type=socket.SOCK_DGRAM)
except socket.gaierror as e:
steps.append(_step(_FAIL, "Résolution du nom du serveur",
f"« {endpoint} » non résolu : {e.strerror or e}"))
return steps, ""
family, socktype, proto, _, addr = infos[0]
server_ip = addr[0]
steps.append(_step(_OK, "Résolution du nom du serveur",
f"{endpoint}{server_ip}"))
# Route empruntée : si elle sort par l'adresse du tunnel, le trafic du
# tunnel repasse par le tunnel et rien ne circule plus.
src = _local_route_source(server_ip, port)
client_ip = (wg.get("client_address") or "").split("/")[0].strip()
if not src:
steps.append(_step(_FAIL, "Route vers le serveur",
f"Aucune route vers {server_ip}"))
elif src == client_ip:
steps.append(_step(
_FAIL, "Route vers le serveur",
f"Le serveur ({server_ip}) est routé dans le tunnel lui-même "
f"(source {src}). Excluez l'IP du serveur des « IPs autorisées »."))
else:
steps.append(_step(_OK, "Route vers le serveur",
f"Sortie par {src}"))
allowed = _networks(wg.get("allowed_ips", ""))
# L'endpoint capturé par AllowedIPs : le tunnel se coupe l'herbe sous le
# pied dès qu'il monte, et le handshake suivant n'arrive jamais.
try:
server_addr = ipaddress.ip_address(server_ip)
except ValueError:
server_addr = None
if server_addr and any(server_addr in n for n in allowed):
steps.append(_step(
_FAIL, "Cohérence des « IPs autorisées »",
f"{server_ip} (le serveur) est inclus dans les IPs autorisées : "
f"une fois monté, le tunnel s'auto-route et le handshake cesse. "
f"Retirez cette plage ou excluez {server_ip}/32."))
elif src and allowed:
try:
local_addr = ipaddress.ip_address(src)
except ValueError:
local_addr = None
clash = [str(n) for n in allowed if local_addr and local_addr in n]
if clash:
steps.append(_step(
_WARN, "Cohérence des « IPs autorisées »",
f"Votre réseau local ({src}) est couvert par {', '.join(clash)} : "
f"une fois le tunnel monté, imprimante, NAS et box locaux "
f"passeront par le VPN"))
else:
steps.append(_step(_OK, "Cohérence des « IPs autorisées »",
f"{len(allowed)} plage(s), sans conflit avec le "
f"réseau local"))
ms = ping_server(server_ip, timeout=2)
if ms is None:
steps.append(_step(_WARN, "Ping du serveur (ICMP)",
"Pas de réponse — normal si l'ICMP est filtré"))
else:
steps.append(_step(_OK, "Ping du serveur (ICMP)", f"{ms} ms"))
sock = socket.socket(family, socktype, proto)
sock.settimeout(timeout)
try:
sock.sendto(b"\x00" * 32, addr)
try:
sock.recvfrom(1024)
steps.append(_step(_OK, f"Port UDP {port}", "Réponse reçue"))
except socket.timeout:
steps.append(_step(
_OK if ms is not None else _WARN, f"Port UDP {port}",
"Ouvert ou filtré : aucun rejet ICMP (un serveur WireGuard "
"ignore silencieusement les paquets non authentifiés, c'est "
"le résultat attendu)"))
except ConnectionRefusedError:
steps.append(_step(_FAIL, f"Port UDP {port}",
f"Fermé sur {server_ip} : rien n'écoute, ou un "
f"pare-feu rejette le trafic"))
except OSError as e:
steps.append(_step(_FAIL, f"Port UDP {port}", str(e)))
finally:
sock.close()
return steps, server_ip
def _diag_tunnel(cfg: Config) -> list[dict]:
"""Étapes portant sur le tunnel lui-même : montage, handshake, trafic."""
wg = cfg.wg
name = wg.get("interface_name", "wgs0")
steps: list[dict] = []
if not is_connected(cfg):
steps.append(_step(_SKIP, "Tunnel monté",
"Tunnel inactif — connectez-vous pour tester la "
"suite de la chaîne"))
return steps
steps.append(_step(_OK, "Tunnel monté",
f"Service actif sur l'interface {name}"))
age = last_handshake_age(name)
if age is None:
steps.append(_step(
_WARN, "Handshake",
"État inaccessible : `wg show` exige des droits "
+ ("administrateur" if is_windows() else "root")))
return steps
if age < 0:
steps.append(_step(
_FAIL, "Handshake",
"Jamais négocié. Le tunnel est monté mais le serveur ne répond "
"pas à nos paquets. Causes usuelles : clé publique du client "
"absente côté serveur, clé publique du serveur erronée, clé "
"pré-partagée manquante ou différente, port UDP bloqué."))
return steps
if age > 180:
steps.append(_step(_WARN, "Handshake",
f"Dernier échange il y a {age} s — le pair ne "
f"répond plus (keepalive recommandé : 25 s)"))
else:
steps.append(_step(_OK, "Handshake", f"Négocié il y a {age} s"))
info = get_status_info(cfg)
rx, tx = info.get("rx_bytes"), info.get("tx_bytes")
if rx in (0, "0", "", None) and tx in (0, "0", "", None):
steps.append(_step(_WARN, "Trafic chiffré", "Aucun octet échangé"))
else:
steps.append(_step(_OK, "Trafic chiffré", f"{rx}{tx}"))
gateway = _tunnel_gateway(wg.get("client_address", ""),
wg.get("allowed_ips", ""))
if not gateway:
steps.append(_step(_SKIP, "Pair joignable dans le tunnel",
"Adresse du tunnel non exploitable"))
else:
ms = ping_server(gateway, timeout=2)
if ms is None:
steps.append(_step(
_WARN, "Pair joignable dans le tunnel",
f"{gateway} ne répond pas — ICMP filtré côté serveur, ou "
f"« IPs autorisées » ne couvre pas ce réseau"))
else:
steps.append(_step(_OK, "Pair joignable dans le tunnel",
f"{gateway} répond en {ms} ms"))
if not dns_util.resolution_works():
steps.append(_step(_FAIL, "Résolution DNS dans le tunnel",
"Plus aucun nom ne se résout — utilisez "
"« Réparer le DNS »"))
else:
steps.append(_step(_OK, "Résolution DNS dans le tunnel", "Fonctionnelle"))
return steps
def diagnose_connection(cfg: Config, timeout: int = 5) -> list[dict]:
"""Vérifie toute la chaîne de connexion et désigne le point de rupture.
Le test précédent s'arrêtait à la joignabilité UDP du serveur : il
concluait « serveur joignable » sur un tunnel qui n'échangeait rien, sans
jamais dire où ça coinçait. Les étapes vont ici de la configuration locale
jusqu'à la résolution DNS à l'intérieur du tunnel.
Retourne une liste de {"state", "label", "detail"}, dans l'ordre de la
chaîne. Une étape `fail` arrête la progression : les suivantes seraient
ininterprétables.
"""
steps: list[dict] = []
if not wg_quick_available():
steps.append(_step(
_FAIL, "WireGuard installé",
"wireguard.exe introuvable — installez WireGuard for Windows"
if is_windows() else
"wg-quick introuvable — installez le paquet wireguard-tools"))
return steps
if not wg_available():
steps.append(_step(_WARN, "Outils WireGuard",
"`wg` introuvable : l'état du handshake ne pourra "
"pas être lu"))
else:
steps.append(_step(_OK, "WireGuard installé", "Binaires présents"))
if not is_windows() and not has_root_privileges():
steps.append(_step(_WARN, "Privilèges",
"Exécution sans root : certains détails du tunnel "
"resteront masqués"))
steps += _diag_config(cfg)
if any(s["state"] == _FAIL for s in steps):
return steps
net_steps, _ = _diag_reachability(cfg, timeout)
steps += net_steps
if any(s["state"] == _FAIL for s in net_steps):
return steps
steps += _diag_tunnel(cfg)
return steps
def parse_conf_file(path: str) -> dict | None: def parse_conf_file(path: str) -> dict | None:
"""Parse un fichier .conf WireGuard et retourne les valeurs extraites.""" """Parse un fichier .conf WireGuard et retourne les valeurs extraites."""
result: dict = {} result: dict = {}
@@ -338,9 +750,18 @@ def parse_conf_file(path: str) -> dict | None:
result["client_address"] = val.split(",")[0].strip() result["client_address"] = val.split(",")[0].strip()
elif key == "DNS": elif key == "DNS":
result["dns"] = val.split(",")[0].strip() result["dns"] = val.split(",")[0].strip()
elif key == "MTU":
try:
result["mtu"] = int(val)
except ValueError:
pass
elif section == "peer": elif section == "peer":
if key == "PublicKey": if key == "PublicKey":
result["server_public_key"] = val result["server_public_key"] = val
elif key == "PresharedKey":
# Ignorée jusqu'ici : le .conf importé perdait sa PSK
# et le tunnel ne négociait plus rien.
result["preshared_key"] = val
elif key == "Endpoint" and ":" in val: elif key == "Endpoint" and ":" in val:
host, port = val.rsplit(":", 1) host, port = val.rsplit(":", 1)
result["server_endpoint"] = host.strip("[]") result["server_endpoint"] = host.strip("[]")
@@ -423,10 +844,18 @@ def ping_server(host: str, timeout: int = 2) -> int | None:
return None return None
# Parser "time=X.X ms" ou "temps=X.X ms" # Parser "time=X.X ms" ou "temps=X.X ms"
import re import re
m = re.search(r"[Tt]ime[<=]([\d.]+)\s*ms", out) # `ping` est traduit : « temps=12ms » sur un Windows français, « tempo »
# sur un italien. Ne matcher que « time » faisait conclure « hôte
# injoignable » alors que la commande avait réussi (code 0).
m = re.search(r"(?:time|temps|tempo|Zeit|tiempo)[<=]\s*([\d.,]+)\s*m?s",
out, re.IGNORECASE)
if m: if m:
return int(float(m.group(1))) try:
return None return int(float(m.group(1).replace(",", ".")))
except ValueError:
pass
# Réponse reçue mais latence non analysable : 0 ms plutôt que « injoignable ».
return 0
# ── Test DNS leak ───────────────────────────────────────────────────────── # ── Test DNS leak ─────────────────────────────────────────────────────────
@@ -538,37 +967,15 @@ def _fmt_bytes(n: int) -> str:
return f"{n} B" return f"{n} B"
def get_status_info(cfg: Config) -> dict: def _fill_peer_stats(info: dict, out: str) -> None:
name = cfg.wg.get("interface_name", "wgs0") """Complète `info` avec la sortie de `wg show <iface>`.
info = {
"connected": False,
"interface": name,
"peer": "",
"rx_bytes": 0,
"tx_bytes": 0,
"last_handshake": "",
}
if is_windows():
info["connected"] = is_connected(cfg)
return info
info["connected"] = is_connected(cfg) Sans privilèges (Linux) ou sans droits administrateur (Windows), `wg show`
if not info["connected"]: ne renvoie rien : l'état de la connexion, lui, reste correct.
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: if not out:
# Détails du pair inaccessibles sans CAP_NET_ADMIN — l'état de la
# connexion, lui, reste correct.
info["last_handshake"] = "" info["last_handshake"] = ""
return info return
for line in out.splitlines(): for line in out.splitlines():
line = line.strip() line = line.strip()
@@ -585,4 +992,31 @@ def get_status_info(cfg: Config) -> dict:
elif line.startswith("latest handshake:"): elif line.startswith("latest handshake:"):
info["last_handshake"] = line.split(":", 1)[1].strip() info["last_handshake"] = line.split(":", 1)[1].strip()
def get_status_info(cfg: Config) -> dict:
name = cfg.wg.get("interface_name", "wgs0")
info = {
"connected": False,
"interface": name,
"peer": "",
"rx_bytes": 0,
"tx_bytes": 0,
"last_handshake": "",
}
info["connected"] = is_connected(cfg)
if is_windows():
if info["connected"]:
_fill_peer_stats(info, _wg_show(name))
return info
if not info["connected"]:
return info
# Compteurs : /proc/net/dev est lisible sans privilèges, contrairement à
# `wg show`. Sert de repli quand l'application tourne en simple utilisateur.
counters = get_interface_bytes(name)
if counters:
info["rx_bytes"] = _fmt_bytes(counters[0])
info["tx_bytes"] = _fmt_bytes(counters[1])
_fill_peer_stats(info, _wg_show(name))
return info return info
+119 -17
View File
@@ -283,6 +283,16 @@ class AdminWindow(QDialog):
self._srv_pubkey = QLineEdit() self._srv_pubkey = QLineEdit()
self._srv_pubkey.setPlaceholderText("Clé publique du serveur (base64)") self._srv_pubkey.setPlaceholderText("Clé publique du serveur (base64)")
self._row(form, "Clé publique serveur :", self._srv_pubkey) self._row(form, "Clé publique serveur :", self._srv_pubkey)
self._psk = QLineEdit()
self._psk.setEchoMode(QLineEdit.EchoMode.Password)
self._psk.setPlaceholderText("(vide si le serveur n'en utilise pas)")
self._psk.setToolTip(
"Clé pré-partagée (PresharedKey). Doit être identique des deux côtés :\n"
"si le serveur en attend une et qu'elle manque ici, le tunnel monte\n"
"mais aucun handshake n'aboutit."
)
self._row(form, "Clé pré-partagée :", self._psk)
lay.addWidget(grp) lay.addWidget(grp)
# ── Interface client ───────────────────────────────────────────── # ── Interface client ─────────────────────────────────────────────
@@ -311,6 +321,16 @@ class AdminWindow(QDialog):
self._keepalive.setValue(25) self._keepalive.setValue(25)
self._keepalive.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self._keepalive.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
self._row(form2, "Keepalive (s) :", self._keepalive) self._row(form2, "Keepalive (s) :", self._keepalive)
self._mtu = QSpinBox()
self._mtu.setRange(0, 9000)
self._mtu.setSpecialValueText("automatique")
self._mtu.setToolTip(
"0 = laisser WireGuard décider. Abaisser à 12801420 si le tunnel\n"
"monte mais que les gros paquets ne passent pas."
)
self._mtu.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
self._row(form2, "MTU :", self._mtu)
lay.addWidget(grp2) lay.addWidget(grp2)
lay.addStretch() lay.addStretch()
@@ -439,9 +459,13 @@ class AdminWindow(QDialog):
w, lay = self._dark_page("📡 Test de connectivité réseau", "#1a4a7a") w, lay = self._dark_page("📡 Test de connectivité réseau", "#1a4a7a")
# Test UDP serveur # Test UDP serveur
grp1 = QGroupBox("Test UDP serveur") grp1 = QGroupBox("Diagnostic de la chaîne de connexion")
g1 = QVBoxLayout(grp1) g1 = QVBoxLayout(grp1)
g1.addWidget(QLabel("Vérifie la joignabilité UDP du serveur WireGuard.")) g1.addWidget(QLabel(
"Vérifie toute la chaîne : configuration, clés, route, port UDP,\n"
"montage du tunnel, handshake, trafic et DNS — et désigne l'étape\n"
"qui bloque."
))
btn_test = QPushButton("🚀 Lancer le test de connexion") btn_test = QPushButton("🚀 Lancer le test de connexion")
btn_test.setStyleSheet( btn_test.setStyleSheet(
"QPushButton { padding: 9px; background: #1e8449; color: white;" "QPushButton { padding: 9px; background: #1e8449; color: white;"
@@ -452,7 +476,13 @@ class AdminWindow(QDialog):
g1.addWidget(btn_test) g1.addWidget(btn_test)
self._test_result = QLabel("") self._test_result = QLabel("")
self._test_result.setWordWrap(True) self._test_result.setWordWrap(True)
self._test_result.setAlignment(Qt.AlignmentFlag.AlignCenter) self._test_result.setTextFormat(Qt.TextFormat.RichText)
self._test_result.setAlignment(
Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop
)
self._test_result.setTextInteractionFlags(
Qt.TextInteractionFlag.TextSelectableByMouse
)
self._test_result.setStyleSheet( self._test_result.setStyleSheet(
"color: white; padding: 8px; border-radius: 6px; font-size: 12px;" "color: white; padding: 8px; border-radius: 6px; font-size: 12px;"
) )
@@ -675,7 +705,7 @@ class AdminWindow(QDialog):
name_lbl.setStyleSheet("color: white;") name_lbl.setStyleSheet("color: white;")
layout.addWidget(name_lbl) layout.addWidget(name_lbl)
version_lbl = QLabel("Version 0.4.3") version_lbl = QLabel("Version 0.4.5")
version_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter) version_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
version_lbl.setStyleSheet("color: rgba(255,255,255,0.7); font-size: 13px;") version_lbl.setStyleSheet("color: rgba(255,255,255,0.7); font-size: 13px;")
layout.addWidget(version_lbl) layout.addWidget(version_lbl)
@@ -725,11 +755,13 @@ class AdminWindow(QDialog):
self._srv_endpoint.setText(wg.get("server_endpoint", "")) self._srv_endpoint.setText(wg.get("server_endpoint", ""))
self._srv_port.setValue(int(wg.get("server_port", 51820))) self._srv_port.setValue(int(wg.get("server_port", 51820)))
self._srv_pubkey.setText(wg.get("server_public_key", "")) self._srv_pubkey.setText(wg.get("server_public_key", ""))
self._psk.setText(wg.get("preshared_key", ""))
self._iface_name.setText(wg.get("interface_name", "wgs0")) self._iface_name.setText(wg.get("interface_name", "wgs0"))
self._client_addr.setText(wg.get("client_address", "10.8.0.2/24")) self._client_addr.setText(wg.get("client_address", "10.8.0.2/24"))
self._dns.setText(wg.get("dns", "1.1.1.1")) self._dns.setText(wg.get("dns", "1.1.1.1"))
self._allowed_ips.setText(wg.get("allowed_ips", "0.0.0.0/0")) self._allowed_ips.setText(wg.get("allowed_ips", "0.0.0.0/0"))
self._keepalive.setValue(int(wg.get("keepalive", 25))) self._keepalive.setValue(int(wg.get("keepalive", 25)))
self._mtu.setValue(int(wg.get("mtu") or 0))
self._priv_key.setText(wg.get("client_private_key", "")) self._priv_key.setText(wg.get("client_private_key", ""))
self._pub_key.setText(wg.get("client_public_key", "")) self._pub_key.setText(wg.get("client_public_key", ""))
self._mfa_enabled_cb.setChecked(self._cfg.mfa_enabled) self._mfa_enabled_cb.setChecked(self._cfg.mfa_enabled)
@@ -739,11 +771,13 @@ class AdminWindow(QDialog):
self._cfg.set("wg", "server_endpoint", self._srv_endpoint.text().strip()) self._cfg.set("wg", "server_endpoint", self._srv_endpoint.text().strip())
self._cfg.set("wg", "server_port", self._srv_port.value()) self._cfg.set("wg", "server_port", self._srv_port.value())
self._cfg.set("wg", "server_public_key", self._srv_pubkey.text().strip()) self._cfg.set("wg", "server_public_key", self._srv_pubkey.text().strip())
self._cfg.set("wg", "preshared_key", self._psk.text().strip())
self._cfg.set("wg", "interface_name", self._iface_name.text().strip() or "wgs0") self._cfg.set("wg", "interface_name", self._iface_name.text().strip() or "wgs0")
self._cfg.set("wg", "client_address", self._client_addr.text().strip()) self._cfg.set("wg", "client_address", self._client_addr.text().strip())
self._cfg.set("wg", "dns", self._dns.text().strip()) self._cfg.set("wg", "dns", self._dns.text().strip())
self._cfg.set("wg", "allowed_ips", self._allowed_ips.text().strip()) self._cfg.set("wg", "allowed_ips", self._allowed_ips.text().strip())
self._cfg.set("wg", "keepalive", self._keepalive.value()) self._cfg.set("wg", "keepalive", self._keepalive.value())
self._cfg.set("wg", "mtu", self._mtu.value())
self._cfg.set("wg", "client_private_key", self._priv_key.text().strip()) self._cfg.set("wg", "client_private_key", self._priv_key.text().strip())
self._cfg.set("wg", "client_public_key", self._pub_key.text().strip()) self._cfg.set("wg", "client_public_key", self._pub_key.text().strip())
self._cfg.mfa_enabled = self._mfa_enabled_cb.isChecked() self._cfg.mfa_enabled = self._mfa_enabled_cb.isChecked()
@@ -792,6 +826,7 @@ class AdminWindow(QDialog):
"allowed_ips": self._allowed_ips, "allowed_ips": self._allowed_ips,
"client_private_key": self._priv_key, "client_private_key": self._priv_key,
"client_public_key": self._pub_key, "client_public_key": self._pub_key,
"preshared_key": self._psk,
} }
for k, widget in mapping.items(): for k, widget in mapping.items():
if k in values: if k in values:
@@ -800,6 +835,8 @@ class AdminWindow(QDialog):
self._srv_port.setValue(int(values["server_port"])) self._srv_port.setValue(int(values["server_port"]))
if "keepalive" in values: if "keepalive" in values:
self._keepalive.setValue(int(values["keepalive"])) self._keepalive.setValue(int(values["keepalive"]))
if "mtu" in values:
self._mtu.setValue(int(values["mtu"]))
# Sauvegarder en tant que nouveau profil # Sauvegarder en tant que nouveau profil
self._save_values() self._save_values()
@@ -857,28 +894,66 @@ class AdminWindow(QDialog):
) )
self._mfa_uri_label.setText(mfa_core.get_provisioning_uri(secret)) self._mfa_uri_label.setText(mfa_core.get_provisioning_uri(secret))
#: Rendu d'une étape de diagnostic : puce, couleur du libellé.
_STEP_STYLE = {
"ok": ("", "#a9dfbf"),
"warn": ("⚠️", "#f7dc6f"),
"fail": ("", "#f1948a"),
"skip": ("⏭️", "rgba(255,255,255,0.45)"),
}
def _run_test(self): def _run_test(self):
from PyQt6.QtWidgets import QApplication
self._save_values() self._save_values()
self._test_result.setText("Test en cours…") self._test_result.setText("Diagnostic en cours…")
self._test_result.setStyleSheet( self._test_result.setStyleSheet(
"color: rgba(255,255,255,0.6); padding: 12px; border-radius: 6px; font-size: 13px;" "color: rgba(255,255,255,0.6); padding: 12px; border-radius: 6px; font-size: 13px;"
) )
from PyQt6.QtWidgets import QApplication
QApplication.processEvents() QApplication.processEvents()
ok, msg = wg_core.test_connection(self._cfg)
if ok: steps = wg_core.diagnose_connection(self._cfg)
self._test_result.setText(f"{msg}") self._test_result.setText(self._format_diagnostic(steps))
self._test_result.setStyleSheet(
"padding: 12px; border-radius: 6px; font-size: 13px; font-weight: bold;" states = [s["state"] for s in steps]
"background: rgba(39,174,96,0.25); color: #a9dfbf;" if "fail" in states:
) css = "background: rgba(231,76,60,0.20); color: #f1948a;"
elif "warn" in states or "skip" in states:
css = "background: rgba(241,196,15,0.15); color: #f7dc6f;"
else: else:
self._test_result.setText(f"{msg}") css = "background: rgba(39,174,96,0.20); color: #a9dfbf;"
self._test_result.setStyleSheet( self._test_result.setStyleSheet(
"padding: 12px; border-radius: 6px; font-size: 13px; font-weight: bold;" "padding: 12px; border-radius: 6px; font-size: 12px; " + css
"background: rgba(231,76,60,0.25); color: #f1948a;" )
def _format_diagnostic(self, steps: list[dict]) -> str:
"""Met en forme les étapes ; la première en échec porte le verdict."""
from html import escape
rows = []
for st in steps:
bullet, color = self._STEP_STYLE.get(st["state"], ("", "white"))
weight = "bold" if st["state"] == "fail" else "normal"
detail = escape(st["detail"])
rows.append(
f'<div style="margin-bottom:5px;">'
f'<span style="color:{color}; font-weight:{weight};">'
f'{bullet}&nbsp;{escape(st["label"])}</span>'
f'<br><span style="color:rgba(255,255,255,0.65); font-size:11px;">'
f'&nbsp;&nbsp;&nbsp;&nbsp;{detail}</span></div>'
) )
failed = next((s for s in steps if s["state"] == "fail"), None)
if failed:
verdict = (f'<div style="margin-bottom:9px; font-weight:bold;">'
f'⛔ Blocage : {escape(failed["label"])}</div>')
elif any(s["state"] == "skip" for s in steps):
verdict = ('<div style="margin-bottom:9px; font-weight:bold;">'
'➡️ Chaîne validée jusqu\'au tunnel — reste à le monter'
'</div>')
else:
verdict = ('<div style="margin-bottom:9px; font-weight:bold;">'
'✅ Chaîne de connexion complète</div>')
return verdict + "".join(rows)
def _export_qr(self): def _export_qr(self):
"""Exporte la config client comme QR Code pour import mobile.""" """Exporte la config client comme QR Code pour import mobile."""
self._save_values() self._save_values()
@@ -952,6 +1027,33 @@ class AdminWindow(QDialog):
note.setStyleSheet("color: rgba(255,255,255,0.7); font-size: 10px; padding: 6px;") note.setStyleSheet("color: rgba(255,255,255,0.7); font-size: 10px; padding: 6px;")
v.addWidget(note) v.addWidget(note)
# Sans ce bouton, la PSK générée n'existait que dans ce texte : elle
# n'était reportée nulle part, et le tunnel monté avec les clés
# affichées ne négociait jamais avec le serveur.
btn_apply = QPushButton("📥 Appliquer le volet client à cette configuration")
btn_apply.setStyleSheet(
"QPushButton { background:#1e8449; color:white; border-radius:5px;"
" padding:7px 16px; border:none; } QPushButton:hover { background:#27ae60; }"
)
def _apply():
self._priv_key.setText(result["client_priv"])
self._pub_key.setText(result["client_pub"])
self._srv_pubkey.setText(result["server_pub"])
self._psk.setText(result["psk"])
self._refresh_preview()
dlg.accept()
QMessageBox.information(
self, "Config client appliquée",
"Clés client, clé publique du serveur et clé pré-partagée "
"reportées dans la configuration.\n\n"
"⚠ Déployez le volet serveur sur la machine distante et "
"renseignez son adresse publique."
)
btn_apply.clicked.connect(_apply)
v.addWidget(btn_apply)
btn_close = QPushButton("Fermer") btn_close = QPushButton("Fermer")
btn_close.setStyleSheet( btn_close.setStyleSheet(
"QPushButton { background:#2471a3; color:white; border-radius:5px;" "QPushButton { background:#2471a3; color:white; border-radius:5px;"
+122 -3
View File
@@ -29,6 +29,45 @@ def get_wg_config_dir() -> str:
return "/etc/wireguard" return "/etc/wireguard"
def _windows_wg_dir() -> str:
"""Répertoire d'installation de WireGuard for Windows."""
for base in (os.environ.get("ProgramFiles", r"C:\Program Files"),
os.environ.get("ProgramW6432", r"C:\Program Files"),
os.environ.get("ProgramFiles(x86)", "")):
if not base:
continue
path = os.path.join(base, "WireGuard")
if os.path.isdir(path):
return path
return ""
def _windows_exe(name: str) -> str:
"""Chemin absolu de wireguard.exe / wg.exe.
L'installateur WireGuard n'ajoute pas son répertoire au PATH : invoquer
« wireguard » ou « wg » tel quel remonte « Commande introuvable » sur une
machine WireGuard est pourtant installé.
"""
import shutil as _sh
d = _windows_wg_dir()
if d:
path = os.path.join(d, name)
if os.path.isfile(path):
return path
return _sh.which(name) or name
def wireguard_exe() -> str:
"""Binaire du gestionnaire WireGuard (installe/désinstalle les tunnels)."""
return _windows_exe("wireguard.exe") if is_windows() else "wg-quick"
def wg_exe() -> str:
"""Binaire `wg` (interrogation d'un tunnel actif)."""
return _windows_exe("wg.exe") if is_windows() else "wg"
def has_root_privileges() -> bool: def has_root_privileges() -> bool:
if is_windows(): if is_windows():
try: try:
@@ -123,9 +162,84 @@ def _sudo_refused_auth(err: str) -> bool:
return any(marker in low for marker in _SUDO_AUTH_MARKERS) return any(marker in low for marker in _SUDO_AUTH_MARKERS)
def _run_elevated_windows(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
"""Relance `cmd` derrière une invite UAC et attend sa fin.
`wireguard.exe /installtunnelservice` et `sc create/delete` exigent les
droits administrateur : lancés depuis une session standard, ils échouaient
sur « Access is denied » sans que l'application le sache. ShellExecuteEx
avec le verbe « runas » est le seul moyen d'obtenir l'élévation depuis un
processus déjà démarré.
Un processus élevé n'hérite pas de nos tubes : stdout/stderr sont perdus,
seul le code de sortie remonte.
"""
import ctypes
from ctypes import wintypes
SEE_MASK_NOCLOSEPROCESS = 0x00000040
SEE_MASK_NOASYNC = 0x00000100
SW_HIDE = 0
ERROR_CANCELLED = 1223
WAIT_TIMEOUT = 0x00000102
class SHELLEXECUTEINFOW(ctypes.Structure):
_fields_ = [
("cbSize", wintypes.DWORD),
("fMask", ctypes.c_ulong),
("hwnd", wintypes.HWND),
("lpVerb", wintypes.LPCWSTR),
("lpFile", wintypes.LPCWSTR),
("lpParameters", wintypes.LPCWSTR),
("lpDirectory", wintypes.LPCWSTR),
("nShow", ctypes.c_int),
("hInstApp", wintypes.HINSTANCE),
("lpIDList", ctypes.c_void_p),
("lpClass", wintypes.LPCWSTR),
("hkeyClass", wintypes.HKEY),
("dwHotKey", wintypes.DWORD),
("hIconOrMonitor", wintypes.HANDLE),
("hProcess", wintypes.HANDLE),
]
info = SHELLEXECUTEINFOW()
info.cbSize = ctypes.sizeof(info)
info.fMask = SEE_MASK_NOCLOSEPROCESS | SEE_MASK_NOASYNC
info.lpVerb = "runas"
info.lpFile = cmd[0]
info.lpParameters = subprocess.list2cmdline(cmd[1:])
info.nShow = SW_HIDE
if not ctypes.windll.shell32.ShellExecuteExW(ctypes.byref(info)):
err = ctypes.get_last_error() or ctypes.GetLastError()
if err == ERROR_CANCELLED:
return -1, "", "Élévation refusée par l'utilisateur"
return -1, "", f"Élévation impossible (code {err}) : {cmd[0]}"
handle = info.hProcess
if not handle:
# Commande lancée sans handle exploitable : succès non vérifiable.
return 0, "", ""
try:
if ctypes.windll.kernel32.WaitForSingleObject(
handle, int(timeout * 1000)
) == WAIT_TIMEOUT:
return -1, "", "Timeout"
code = wintypes.DWORD()
ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(code))
return code.value, "", ""
finally:
ctypes.windll.kernel32.CloseHandle(handle)
def run_privileged(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]: def run_privileged(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
if is_windows(): if is_windows():
return run_command(cmd, timeout) if has_root_privileges():
return run_command(cmd, timeout)
try:
return _run_elevated_windows(cmd, timeout)
except Exception as e:
return -1, "", str(e)
if has_root_privileges(): if has_root_privileges():
return run_command(cmd, timeout) return run_command(cmd, timeout)
@@ -217,14 +331,19 @@ def privilege_report() -> list[tuple[str, bool, str]]:
def wg_available() -> bool: def wg_available() -> bool:
if is_windows():
return os.path.isfile(wg_exe())
code, _, _ = run_command(["wg", "--version"]) code, _, _ = run_command(["wg", "--version"])
return code == 0 return code == 0
def wg_quick_available() -> bool: def wg_quick_available() -> bool:
# wireguard.exe est une application graphique : elle n'accepte pas
# « --help » et n'écrit rien sur stdout. Le sonder comme un outil console
# concluait « WireGuard absent » sur toute machine Windows. Sa présence sur
# le disque est le seul test fiable.
if is_windows(): if is_windows():
code, _, _ = run_command(["wireguard", "--help"]) return os.path.isfile(wireguard_exe())
return code == 0
code, _, _ = run_command(["wg-quick", "--help"]) code, _, _ = run_command(["wg-quick", "--help"])
return code == 0 return code == 0
+2 -2
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
WGSecure (WGS) Interface graphique WireGuard avec MFA WGSecure (WGS) Interface graphique WireGuard avec MFA
Version 0.4.3 Version 0.4.5
""" """
import sys import sys
import argparse import argparse
@@ -92,7 +92,7 @@ def main():
) )
app = QApplication(sys.argv) app = QApplication(sys.argv)
app.setApplicationName("WGSecure") app.setApplicationName("WGSecure")
app.setApplicationVersion("0.4.3") app.setApplicationVersion("0.4.5")
app.setOrganizationName("WGS") app.setOrganizationName("WGS")
# Icône globale multi-tailles — barre des tâches + alt-tab # Icône globale multi-tailles — barre des tâches + alt-tab
app_icon = icons.icon_app() app_icon = icons.icon_app()
+1 -1
View File
@@ -4,7 +4,7 @@
; ────────────────────────────────────────────── ; ──────────────────────────────────────────────
#define MyAppName "WGSecure" #define MyAppName "WGSecure"
#define MyAppVersion "0.4.3" #define MyAppVersion "0.4.5"
#define MyAppPublisher "WGSecure" #define MyAppPublisher "WGSecure"
#define MyAppExeName "wgsecure.exe" #define MyAppExeName "wgsecure.exe"