Johnny/WGSecure
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
60e85b1c20f14fec5588572ad2170aeb0f61fe92
WGSecure/app/core/wireguard.py
T
tuxgyver 1b74719204 fix: DNS coupée après arrêt du tunnel, gel UI, dialogues sudo à répétition 
- 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.
2026-08-31 11:16:04 +02:00

589 lines
22 KiB
Python
Raw Blame History

import base64
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,
get_wg_config_dir,
is_windows,
run_command,
run_privileged,
wg_available,
)
def generate_keypair() -> tuple[str, str]:
"""Retourne (private_key_b64, public_key_b64) au format WireGuard."""
private = X25519PrivateKey.generate()
private_bytes = private.private_bytes_raw()
public_bytes = private.public_key().public_bytes_raw()
return (
base64.b64encode(private_bytes).decode(),
base64.b64encode(public_bytes).decode(),
)
def generate_preshared_key() -> str:
return base64.b64encode(os.urandom(32)).decode()
def build_client_config(cfg: Config) -> str:
wg = cfg.wg
lines = [
"[Interface]",
f"PrivateKey = {wg['client_private_key']}",
f"Address = {wg['client_address']}",
]
# Une directive DNS = sans backend resolvconf/resolvectl fait échouer
# `wg-quick up` en plein milieu de son exécution : l'interface est déjà
# créée, le trap de nettoyage laisse un état partiel. Mieux vaut monter le
# tunnel sans toucher au DNS système.
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']}",
f"AllowedIPs = {wg['allowed_ips']}",
f"Endpoint = {wg['server_endpoint']}:{wg['server_port']}",
f"PersistentKeepalive = {wg['keepalive']}",
]
return "\n".join(lines) + "\n"
def get_client_config_path(cfg: Config) -> str:
name = cfg.wg.get("interface_name", "wgs0")
if is_windows():
return os.path.join(get_config_dir(), f"{name}.conf")
return os.path.join(get_wg_config_dir(), f"{name}.conf") # /etc/wireguard/
def write_client_config(cfg: Config) -> tuple[bool, str]:
"""Écrit la config client, en élevant les privilèges si nécessaire.
/etc/wireguard est en 0700 root:root : l'écriture directe échoue pour un
utilisateur normal. On passe alors par un fichier temporaire dans le
répertoire de configuration de l'utilisateur, installé ensuite par
`install` — une seule commande privilégiée, qui pose contenu, propriétaire
et permissions d'un coup, et qui traverse toute la chaîne
sudo -n → askpass → pkexec de run_privileged().
"""
content = build_client_config(cfg)
path = get_client_config_path(cfg)
# Chemin direct : Windows, ou exécution en root.
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write(content)
if not is_windows():
os.chmod(path, 0o600)
return True, path
except (PermissionError, FileNotFoundError, OSError):
if is_windows():
return False, f"Écriture impossible dans {path}"
tmp = os.path.join(get_config_dir(), ".wgs_tmp.conf")
try:
# Créer en 0600 dès l'origine : la clé privée ne doit jamais être
# lisible par les autres utilisateurs, même brièvement.
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as f:
f.write(content)
code, _, err = run_privileged(
["install", "-D", "-m", "600", "-o", "root", "-g", "root", tmp, path],
timeout=60,
)
if code == 0:
return True, path
return False, err or f"Écriture impossible dans {path}"
except OSError as e:
return False, str(e)
finally:
try:
os.unlink(tmp)
except OSError:
pass
def _wg_show(name: str) -> str:
"""Sortie de `wg show <iface>`, ou "" si les privilèges manquent.
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
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]:
if not cfg.configured:
return False, "WireGuard non configuré. Ouvrez le panneau Admin."
ok, result = write_client_config(cfg)
if not ok:
return False, f"Impossible d'écrire la config : {result}"
name = cfg.wg.get("interface_name", "wgs0")
if is_windows():
code, _, err = run_command(["wireguard", "/installtunnel", result])
return (code == 0), (err or "Connecté")
# Déjà monté et fonctionnel : rien à faire.
if is_connected(cfg):
return True, "Tunnel WireGuard déjà actif"
# Interface fantôme d'une session précédente : `wg-quick up` mourrait sur
# « `wgs0' already exists ». On nettoie avant de remonter.
if dns_util.interface_exists(name):
force_cleanup(cfg)
# Passer le nom d'interface (pas le chemin) : AppArmor autorise /etc/wireguard/ seulement
code, _, err = run_privileged(["wg-quick", "up", name], timeout=90)
if code == 0:
return True, "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"
def disconnect(cfg: Config) -> tuple[bool, str]:
name = cfg.wg.get("interface_name", "wgs0")
if is_windows():
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:
# wg-quick refuserait de tourner ici ; on passe directement au
# démontage manuel, sinon l'entrée DNS du tunnel resterait active.
code = -1
# Nettoyage inconditionnel : `wg-quick down` peut avoir réussi partiellement
# (échec d'un PostDown, interface déjà supprimée) et laissé l'entrée DNS.
force_cleanup(cfg)
if code == 0 or not dns_util.interface_exists(name):
return True, "Tunnel WireGuard désactivé"
return False, err or "Erreur lors de la déconnexion"
def test_connection(cfg: Config, timeout: int = 5) -> tuple[bool, str]:
"""Teste la joignabilité du serveur WireGuard.
Note : `socket.connect()` sur un socket UDP n'émet aucun paquet et réussit
donc toujours — l'ancien test renvoyait « joignable » même serveur éteint.
On résout le nom, on envoie un datagramme et on interprète l'absence de
réponse ICMP « port unreachable » comme un port ouvert/filtré.
"""
endpoint = cfg.wg.get("server_endpoint", "")
port = int(cfg.wg.get("server_port", 51820))
if not endpoint:
return False, "Aucun serveur configuré"
# 1. Résolution du nom — échoue aussi si le DNS système est cassé.
try:
infos = socket.getaddrinfo(endpoint, port, type=socket.SOCK_DGRAM)
except socket.gaierror as e:
return False, f"Nom « {endpoint} » non résolu : {e.strerror or e}"
family, socktype, proto, _, addr = infos[0]
# 2. Datagramme sonde : un ICMP port unreachable remonte en
# ConnectionRefusedError, ce qui prouve que rien n'écoute.
sock = socket.socket(family, socktype, proto)
sock.settimeout(timeout)
try:
start = time.monotonic()
sock.sendto(b"\x00" * 32, addr)
try:
sock.recvfrom(1024)
latency = int((time.monotonic() - start) * 1000)
return True, f"Serveur joignable ({latency} ms)"
except socket.timeout:
pass
except ConnectionRefusedError:
return False, f"Port UDP {port} fermé sur {addr[0]}"
except OSError as e:
return False, str(e)
finally:
sock.close()
# 3. Pas de rejet ICMP : le port est ouvert ou filtré. On complète avec un
# ping pour distinguer « hôte injoignable » de « serveur silencieux ».
ms = ping_server(addr[0], timeout=2)
if ms is None:
return True, (f"Port UDP {port} ouvert ou filtré sur {addr[0]} "
"(hôte ne répond pas au ping)")
return True, f"Serveur joignable ({ms} ms, port UDP {port} non rejeté)"
def parse_conf_file(path: str) -> dict | None:
"""Parse un fichier .conf WireGuard et retourne les valeurs extraites."""
result: dict = {}
section = None
try:
with open(path, "r", encoding="utf-8") as f:
for raw in f:
line = raw.strip()
if not line or line.startswith("#"):
continue
if line.startswith("["):
section = line.strip("[]").lower()
continue
if "=" not in line:
continue
key, val = (x.strip() for x in line.split("=", 1))
if section == "interface":
if key == "PrivateKey":
result["client_private_key"] = val
elif key == "Address":
result["client_address"] = val.split(",")[0].strip()
elif key == "DNS":
result["dns"] = val.split(",")[0].strip()
elif section == "peer":
if key == "PublicKey":
result["server_public_key"] = val
elif key == "Endpoint" and ":" in val:
host, port = val.rsplit(":", 1)
result["server_endpoint"] = host.strip("[]")
try:
result["server_port"] = int(port)
except ValueError:
pass
elif key == "AllowedIPs":
result["allowed_ips"] = val
elif key == "PersistentKeepalive":
try:
result["keepalive"] = int(val)
except ValueError:
pass
# Dériver la clé publique depuis la clé privée importée
if "client_private_key" in result and "client_public_key" not in result:
try:
priv_bytes = base64.b64decode(result["client_private_key"])
priv = X25519PrivateKey.from_private_bytes(priv_bytes)
pub_bytes = priv.public_key().public_bytes_raw()
result["client_public_key"] = base64.b64encode(pub_bytes).decode()
except Exception:
pass
return result or None
except Exception:
return None
def export_conf_file(cfg: Config, dest_path: str) -> tuple[bool, str]:
"""Exporte la configuration courante vers un fichier .conf."""
try:
content = build_client_config(cfg)
with open(dest_path, "w") as f:
f.write(content)
if not is_windows():
os.chmod(dest_path, 0o600)
return True, dest_path
except Exception as e:
return False, str(e)
def get_interface_bytes(iface: str) -> tuple[int, int] | None:
"""Retourne (rx_bytes, tx_bytes) depuis /proc/net/dev (Linux).
Le nom d'interface est comparé exactement : un `in` sur la ligne entière
confondait « wg0 » avec « wg0bis » et pouvait aussi matcher un compteur.
Les colonnes sont lues après le « : », car sur un compteur très large
/proc/net/dev colle le nom et la valeur (« wgs0:123456789 »).
"""
try:
with open("/proc/net/dev") as f:
for line in f:
if ":" not in line:
continue
name, _, counters = line.partition(":")
if name.strip() != iface:
continue
parts = counters.split()
if len(parts) < 9:
return None
return int(parts[0]), int(parts[8])
except (OSError, ValueError):
pass
return None
def ping_server(host: str, timeout: int = 2) -> int | None:
"""Ping ICMP du serveur. Retourne la latence en ms, ou None si inaccessible."""
if not host:
return None
if is_windows():
code, out, _ = run_command(
["ping", "-n", "1", "-w", str(timeout * 1000), host], timeout + 2
)
else:
code, out, _ = run_command(
["ping", "-c", "1", "-W", str(timeout), host], timeout + 2
)
if code != 0:
return None
# Parser "time=X.X ms" ou "temps=X.X ms"
import re
m = re.search(r"[Tt]ime[<=]([\d.]+)\s*ms", out)
if m:
return int(float(m.group(1)))
return None
# ── Test DNS leak ─────────────────────────────────────────────────────────
def dns_leak_test(cfg: Config) -> dict:
"""
Teste si le DNS fuit en dehors du tunnel.
Retourne {"status": ok|leak|unknown|down, "resolvers": [...], "expected": str}
"""
expected_dns = (cfg.wg.get("dns", "") or "").split(",")[0].strip()
# Sous systemd-resolved, /etc/resolv.conf ne contient que le stub
# 127.0.0.53 : lire ce fichier signalait donc toujours une « fuite ».
# dns_util.active_resolvers() interroge resolvectl quand il est présent.
resolvers = dns_util.active_resolvers()
resolvers = [r for r in resolvers if r not in ("127.0.0.53", "127.0.0.1")] or resolvers
tunnel_up = is_connected(cfg)
# Sonde : quel résolveur récursif sort réellement sur le réseau.
probe_ip = ""
code, out, _ = run_command(["dig", "+short", "+time=3", "whoami.akamai.net"], 8)
if code == 0:
lines = [line.strip() for line in out.splitlines() if line.strip()]
if lines:
probe_ip = lines[-1]
if not dns_util.resolution_works():
status = "down"
elif not resolvers:
status = "unknown"
elif not tunnel_up:
# Tunnel inactif : parler de fuite n'a pas de sens.
status = "unknown"
elif not expected_dns:
status = "unknown"
elif any(expected_dns == r or expected_dns in r for r in resolvers):
status = "ok"
else:
status = "leak"
return {
"status": status,
"resolvers": resolvers,
"expected": expected_dns,
"probe_ip": probe_ip,
"tunnel_up": tunnel_up,
}
# ── Génération config serveur ─────────────────────────────────────────────
def generate_server_config(
server_port: int = 51820,
server_address: str = "10.8.0.1/24",
client_address: str = "10.8.0.2/24",
client_allowed_ips: str = "10.8.0.2/32",
) -> dict:
"""
Génère une paire complète (server_conf, client_conf) avec de nouvelles clés.
Retourne un dict avec server_priv/pub, client_priv/pub, server_conf, client_conf.
"""
srv_priv, srv_pub = generate_keypair()
cli_priv, cli_pub = generate_keypair()
psk = generate_preshared_key()
server_conf = "\n".join([
"[Interface]",
f"PrivateKey = {srv_priv}",
f"Address = {server_address}",
f"ListenPort = {server_port}",
"PostUp = iptables -A FORWARD -i %i -j ACCEPT; iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE",
"PostDown = iptables -D FORWARD -i %i -j ACCEPT; iptables -t nat -D POSTROUTING -o eth0 -j MASQUERADE",
"",
"# === Peer client ===",
"[Peer]",
f"PublicKey = {cli_pub}",
f"PresharedKey = {psk}",
f"AllowedIPs = {client_allowed_ips}",
])
client_conf = "\n".join([
"[Interface]",
f"PrivateKey = {cli_priv}",
f"Address = {client_address}",
"DNS = 1.1.1.1",
"",
"[Peer]",
f"PublicKey = {srv_pub}",
f"PresharedKey = {psk}",
"AllowedIPs = 10.8.0.0/24",
"Endpoint = <SERVER_IP>:" + str(server_port),
"PersistentKeepalive = 25",
])
return {
"server_priv": srv_priv, "server_pub": srv_pub,
"client_priv": cli_priv, "client_pub": cli_pub,
"psk": psk,
"server_conf": server_conf,
"client_conf": client_conf,
}
def _fmt_bytes(n: int) -> str:
for unit, div in (("GiB", 1 << 30), ("MiB", 1 << 20), ("KiB", 1 << 10)):
if n >= div:
return f"{n / div:.2f} {unit}"
return f"{n} B"
def 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": "",
}
if is_windows():
info["connected"] = is_connected(cfg)
return info
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
for line in out.splitlines():
line = line.strip()
if line.startswith("peer:"):
info["peer"] = line.split(":", 1)[1].strip()[:16] + "…"
elif line.startswith("transfer:"):
parts = line.split(":", 1)[1].strip().split(",")
try:
# « 1.23 MiB received » → « 1.23 MiB » (l'unité était perdue)
info["rx_bytes"] = " ".join(parts[0].split()[:2])
info["tx_bytes"] = " ".join(parts[1].split()[:2])
except Exception:
pass
elif line.startswith("latest handshake:"):
info["last_handshake"] = line.split(":", 1)[1].strip()
return info
Reference in New Issue View Git Blame Copy Permalink