From 1f555eafc7c9f8944e2dcd209267f8753d152d82 Mon Sep 17 00:00:00 2001 From: tuxgyver Date: Tue, 1 Sep 2026 15:09:28 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20test=20de=20handshake=20r=C3=A9el=20san?= =?UTF-8?q?s=20monter=20le=20tunnel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Le diagnostic s'arrêtait sur « le port UDP ne rejette rien ». C'est le maximum qu'une sonde aveugle puisse dire : un serveur WireGuard ignore silencieusement tout paquet non authentifié, si bien qu'un port non redirigé, une clé inconnue du serveur et un serveur en parfaite santé produisent exactement la même absence de réponse. Le tunnel n'étant pas monté, les étapes suivantes étaient sautées et le rapport concluait « chaîne validée » sur une configuration qui n'avait aucune chance de négocier. WGSecure construit désormais un vrai message d'initiation (Noise_IKpsk2_25519_ ChaChaPoly_BLAKE2s) et interprète la réponse. Le serveur ne répond que si notre clé publique statique lui est connue et si sa propre clé est celle que nous croyons : une réponse prouve la configuration, et son déchiffrement dit quelle clé pré-partagée est en jeu. Aucun tunnel n'est monté, aucun privilège demandé, aucun état conservé. Quand le serveur reste muet, une seconde tentative part depuis le port du serveur. S'il répond alors, c'est qu'un équipement en chemin n'accepte le trafic que si le port source égale le port de destination — une règle courante sur FortiGate, dont l'objet Service porte un udp-portrange 51820:51820. Un client WireGuard émettant depuis un port aléatoire ne peut jamais la satisfaire, et rien nulle part ne le disait. Corrige aussi le verdict rendu quand le tunnel est inactif : « chaîne validée » laissait croire que tout avait été vérifié, alors que les étapes sautées sont précisément celles qui échouent. Le rapport les nomme désormais. La fenêtre d'administration revient à 720x600 et les encarts posés au-dessus des sous-onglets disparaissent, où ils répétaient ce que les sous-onglets annoncent déjà. --- app/core/wg_handshake.py | 282 +++++++++++++++++++++++++++++++++++++++ app/core/wireguard.py | 39 ++++++ app/ui/admin_window.py | 52 ++++---- 3 files changed, 344 insertions(+), 29 deletions(-) create mode 100644 app/core/wg_handshake.py diff --git a/app/core/wg_handshake.py b/app/core/wg_handshake.py new file mode 100644 index 0000000..77f70bb --- /dev/null +++ b/app/core/wg_handshake.py @@ -0,0 +1,282 @@ +"""Sonde de handshake WireGuard, sans monter le tunnel. + +Le diagnostic s'arrêtait au constat « le port UDP ne rejette rien ». C'est le +maximum qu'une sonde aveugle puisse dire : un serveur WireGuard ignore +silencieusement tout paquet non authentifié, si bien qu'un port fermé par un +pare-feu, un serveur écoutant sur un autre port et un serveur en bonne santé +produisent exactement la même absence de réponse. Quand le tunnel n'était pas +monté, le rapport concluait donc « chaîne validée, reste à monter le tunnel » +sur une configuration qui n'avait aucune chance de négocier. + +Ce module construit un vrai message d'initiation (Noise_IKpsk2_25519_ +ChaChaPoly_BLAKE2s, protocole WireGuard v1) et interprète la réponse. Le +serveur ne répond que si notre clé publique statique lui est connue et si sa +propre clé publique est celle que nous croyons : une réponse prouve donc la +configuration, et son déchiffrement dit si une clé pré-partagée est en jeu. + +Aucun état n'est conservé : la session négociée est jetée aussitôt. La sonde +n'ouvre pas de tunnel et ne perturbe pas une session existante — le serveur +traite l'initiation comme celle d'un pair qui se reconnecte. +""" + +import base64 +import hashlib +import hmac +import os +import socket +import struct +import time + +from cryptography.hazmat.primitives.asymmetric.x25519 import ( + X25519PrivateKey, X25519PublicKey, +) +from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 + +_CONSTRUCTION = b"Noise_IKpsk2_25519_ChaChaPoly_BLAKE2s" +_IDENTIFIER = b"WireGuard v1 zx2c4 Jason@zx2c4.com" +_LABEL_MAC1 = b"mac1----" + +_MSG_INITIATION = 1 +_MSG_RESPONSE = 2 +_MSG_COOKIE = 3 + +_INITIATION_LEN = 148 +_RESPONSE_LEN = 92 +_ZERO_PSK = b"\x00" * 32 + + +def _hash(*parts: bytes) -> bytes: + h = hashlib.blake2s(digest_size=32) + for p in parts: + h.update(p) + return h.digest() + + +def _mac(key: bytes, data: bytes) -> bytes: + return hashlib.blake2s(data, digest_size=16, key=key).digest() + + +def _hmac(key: bytes, data: bytes) -> bytes: + return hmac.new(key, data, hashlib.blake2s).digest() + + +def _kdf(key: bytes, data: bytes, n: int) -> list[bytes]: + """KDF de Noise : dérive `n` clés de 32 octets.""" + tau0 = _hmac(key, data) + out = [_hmac(tau0, b"\x01")] + for i in range(2, n + 1): + out.append(_hmac(tau0, out[-1] + bytes([i]))) + return out + + +def _aead_encrypt(key: bytes, counter: int, plain: bytes, ad: bytes) -> bytes: + nonce = b"\x00" * 4 + struct.pack(" bytes: + nonce = b"\x00" * 4 + struct.pack(" bytes: + """Horodatage TAI64N, tel que WireGuard l'attend (12 octets).""" + now = time.time() + secs = int(now) + nanos = int((now - secs) * 1e9) + return struct.pack(">QI", 0x400000000000000A + secs, nanos) + + +class _Initiation: + """Message 1 construit, et l'état de Noise nécessaire à lire le message 2.""" + + def __init__(self, static_priv: X25519PrivateKey, server_pub: X25519PublicKey): + self._static_priv = static_priv + server_pub_raw = server_pub.public_bytes_raw() + + c = _hash(_CONSTRUCTION) + h = _hash(c, _IDENTIFIER) + h = _hash(h, server_pub_raw) + + self._eph_priv = X25519PrivateKey.generate() + eph_pub = self._eph_priv.public_key().public_bytes_raw() + + c = _kdf(c, eph_pub, 1)[0] + h = _hash(h, eph_pub) + + c, k = _kdf(c, self._eph_priv.exchange(server_pub), 2) + enc_static = _aead_encrypt( + k, 0, static_priv.public_key().public_bytes_raw(), h) + h = _hash(h, enc_static) + + c, k = _kdf(c, static_priv.exchange(server_pub), 2) + enc_ts = _aead_encrypt(k, 0, _tai64n(), h) + h = _hash(h, enc_ts) + + self._c, self._h = c, h + self.sender_index = int.from_bytes(os.urandom(4), "little") + + body = (struct.pack(" bool: + """Le message 2 se déchiffre-t-il avec cette clé pré-partagée ?""" + eph_pub_r = packet[12:44] + enc_empty = packet[44:60] + + c, h = self._c, self._h + c = _kdf(c, eph_pub_r, 1)[0] + h = _hash(h, eph_pub_r) + + peer_eph = X25519PublicKey.from_public_bytes(eph_pub_r) + c = _kdf(c, self._eph_priv.exchange(peer_eph), 1)[0] + c = _kdf(c, self._static_priv.exchange(peer_eph), 1)[0] + + c, tau, k = _kdf(c, psk, 3) + h = _hash(h, tau) + try: + _aead_decrypt(k, 0, enc_empty, h) + return True + except Exception: + return False + + +#: Issues possibles de la sonde. +OK_NO_PSK = "ok" # handshake complet, sans clé pré-partagée +OK_PSK = "ok_psk" # handshake complet avec la PSK configurée +PSK_REQUIRED = "psk_required" # le serveur répond, mais la PSK ne convient pas +NO_REPLY = "no_reply" # aucune réponse : clé, port ou filtrage +SRC_PORT_FILTERED = "src_port" # accepté seulement si le port source = port serveur +COOKIE = "cookie" # le serveur exige un cookie (charge/anti-DoS) +BAD_CONFIG = "bad_config" # clés illisibles côté client +NET_ERROR = "net_error" # résolution ou socket en échec + + +def probe_handshake(server_endpoint: str, server_port: int, + server_public_key: str, client_private_key: str, + preshared_key: str = "", timeout: float = 5.0, + attempts: int = 2) -> tuple[str, str]: + """Tente un handshake réel et retourne (issue, message). + + Ne monte aucun tunnel, n'écrit aucun fichier et ne demande aucun + privilège : un simple datagramme UDP sortant. + """ + try: + server_pub = X25519PublicKey.from_public_bytes( + base64.b64decode(server_public_key)) + static_priv = X25519PrivateKey.from_private_bytes( + base64.b64decode(client_private_key)) + except Exception as e: + return BAD_CONFIG, f"Clés illisibles : {e}" + + psk = _ZERO_PSK + if preshared_key.strip(): + try: + psk = base64.b64decode(preshared_key.strip()) + if len(psk) != 32: + return BAD_CONFIG, "Clé pré-partagée : 32 octets attendus" + except Exception: + return BAD_CONFIG, "Clé pré-partagée : base64 invalide" + + try: + infos = socket.getaddrinfo(server_endpoint, server_port, + type=socket.SOCK_DGRAM) + except socket.gaierror as e: + return NET_ERROR, f"« {server_endpoint} » non résolu : {e.strerror or e}" + family, socktype, proto, _, addr = infos[0] + + def _attempt(source_port: int | None) -> tuple[str, str] | None: + """Une tentative. None = pas de réponse dans le délai imparti.""" + init = _Initiation(static_priv, server_pub) + sock = socket.socket(family, socktype, proto) + sock.settimeout(timeout) + try: + if source_port: + sock.bind(("", source_port)) + start = time.monotonic() + sock.sendto(init.packet, addr) + while True: + data, _src = sock.recvfrom(1024) + ms = int((time.monotonic() - start) * 1000) + if len(data) < 4: + continue + kind = struct.unpack(" tuple[list[dict], str]: return steps, server_ip +def _diag_handshake(cfg: Config, timeout: float = 5.0) -> list[dict]: + """Négocie un vrai handshake, sans monter le tunnel. + + Sans cette étape, le diagnostic s'arrêtait sur « le port ne rejette rien » + puis, tunnel non monté, concluait « chaîne validée — reste à le monter ». + Un serveur WireGuard étant silencieux avec les paquets non authentifiés, + ce verdict tombait aussi bien sur une configuration saine que sur un port + non redirigé ou une clé inconnue du serveur. C'est le seul test qui + distingue les deux. + """ + wg = cfg.wg + issue, detail = wg_handshake.probe_handshake( + server_endpoint=(wg.get("server_endpoint") or "").strip(), + server_port=int(wg.get("server_port", 51820)), + server_public_key=(wg.get("server_public_key") or "").strip(), + client_private_key=(wg.get("client_private_key") or "").strip(), + preshared_key=((wg.get("preshared_key") or "").strip() + if uses_preshared_key(wg) else ""), + timeout=timeout, + ) + state = { + wg_handshake.OK_NO_PSK: _OK, + wg_handshake.OK_PSK: _OK, + wg_handshake.COOKIE: _WARN, + wg_handshake.SRC_PORT_FILTERED: _FAIL, + wg_handshake.PSK_REQUIRED: _FAIL, + wg_handshake.NO_REPLY: _FAIL, + wg_handshake.BAD_CONFIG: _FAIL, + wg_handshake.NET_ERROR: _FAIL, + }.get(issue, _WARN) + return [_step(state, "Négociation du handshake", detail)] + + def _diag_tunnel(cfg: Config) -> list[dict]: """Étapes portant sur le tunnel lui-même : montage, handshake, trafic.""" wg = cfg.wg @@ -815,6 +849,11 @@ def diagnose_connection(cfg: Config, timeout: int = 5) -> list[dict]: if any(s["state"] == _FAIL for s in net_steps): return steps + hs_steps = _diag_handshake(cfg, timeout) + steps += hs_steps + if any(s["state"] == _FAIL for s in hs_steps): + return steps + steps += _diag_tunnel(cfg) return steps diff --git a/app/ui/admin_window.py b/app/ui/admin_window.py index 211e3e9..e395a9a 100644 --- a/app/ui/admin_window.py +++ b/app/ui/admin_window.py @@ -24,10 +24,10 @@ class AdminWindow(QDialog): | Qt.WindowType.WindowTitleHint | Qt.WindowType.WindowCloseButtonHint ) - # Taille figée auparavant : le rapport de diagnostic, long d'une - # dizaine d'étapes détaillées, était comprimé en une bande illisible. - self.setMinimumSize(720, 620) - self.resize(880, 740) + # Le découpage en sous-onglets a supprimé l'empilement qui imposait une + # grande fenêtre : chaque page n'affiche plus qu'une section à la fois. + self.setMinimumSize(640, 520) + self.resize(720, 600) self._build_ui() self._load_values() # L'aperçu s'ouvrait vide : il fallait cliquer « Rafraîchir » pour @@ -191,12 +191,7 @@ class AdminWindow(QDialog): tunnel, les clés, puis les fichiers. """ w, lay = self._dark_page("🛡️ WireGuard", theme.BANNER["wireguard"]) - lay.setContentsMargins(10, 10, 10, 0) - lay.addWidget(self._hint( - "Tout ce qui définit le tunnel : les paramètres du pair, vos clés, " - "et les profils enregistrés. Les modifications ne prennent effet " - "qu'après « Enregistrer et fermer »." - )) + lay.setContentsMargins(0, 0, 0, 0) sub = QTabWidget() sub.setStyleSheet(theme.TABBAR_CSS) @@ -453,7 +448,7 @@ class AdminWindow(QDialog): self._config_preview = QTextEdit() self._config_preview.setReadOnly(True) self._config_preview.setFont(QFont("Courier", 9)) - self._config_preview.setMinimumHeight(170) + self._config_preview.setMinimumHeight(140) self._config_preview.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) @@ -512,7 +507,7 @@ class AdminWindow(QDialog): v2 = QVBoxLayout(grp2) self._qr_label = QLabel("(générez un secret pour afficher le QR Code)") self._qr_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - self._qr_label.setFixedHeight(230) + self._qr_label.setFixedHeight(190) v2.addWidget(self._qr_label) self._mfa_uri_label = QLabel("") self._mfa_uri_label.setWordWrap(True) @@ -537,12 +532,7 @@ class AdminWindow(QDialog): dispose maintenant de toute la hauteur. """ w, lay = self._dark_page("📡 Tests & diagnostic", theme.BANNER["test"]) - lay.setContentsMargins(10, 10, 10, 0) - lay.addWidget(self._hint( - "Trois vérifications indépendantes : où la chaîne de connexion se " - "rompt, si le DNS fuit hors du tunnel, et si les droits " - "nécessaires au montage sont réunis." - )) + lay.setContentsMargins(0, 0, 0, 0) sub = QTabWidget() sub.setStyleSheet(theme.TABBAR_CSS) @@ -590,7 +580,7 @@ class AdminWindow(QDialog): # entier quelle que soit la taille de la fenêtre. self._test_result = QTextEdit() self._test_result.setReadOnly(True) - self._test_result.setMinimumHeight(300) + self._test_result.setMinimumHeight(200) self._test_result.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) @@ -626,7 +616,7 @@ class AdminWindow(QDialog): self._dns_result = QTextEdit() self._dns_result.setReadOnly(True) - self._dns_result.setMinimumHeight(220) + self._dns_result.setMinimumHeight(150) self._dns_result.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) @@ -684,7 +674,7 @@ class AdminWindow(QDialog): self._priv_result = QTextEdit() self._priv_result.setReadOnly(True) - self._priv_result.setMinimumHeight(220) + self._priv_result.setMinimumHeight(150) self._priv_result.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) @@ -819,11 +809,7 @@ class AdminWindow(QDialog): qui en comptait sept. """ w, lay = self._dark_page("🔒 Sécurité", theme.BANNER["security"]) - lay.setContentsMargins(10, 10, 10, 0) - lay.addWidget(self._hint( - "Qui a le droit de monter le tunnel, et qui a le droit " - "d'administrer WGSecure." - )) + lay.setContentsMargins(0, 0, 0, 0) sub = QTabWidget() sub.setStyleSheet(theme.TABBAR_CSS) @@ -896,7 +882,7 @@ class AdminWindow(QDialog): name_lbl.setStyleSheet("color: white;") layout.addWidget(name_lbl) - version_lbl = QLabel("Version 0.5.0") + version_lbl = QLabel("Version 0.6.0") version_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter) version_lbl.setStyleSheet("color: rgba(255,255,255,0.7); font-size: 13px;") layout.addWidget(version_lbl) @@ -1143,9 +1129,17 @@ class AdminWindow(QDialog): verdict = (f'
' f'⛔ Blocage : {escape(failed["label"])}
') elif any(s["state"] == "skip" for s in steps): + # « Chaîne validée » auparavant : le mot laissait croire que tout + # avait été vérifié, alors que les étapes sautées sont justement + # celles qui échouent quand un tunnel refuse de négocier. + skipped = ", ".join(s["label"] for s in steps + if s["state"] == "skip") verdict = ('
' - '➡️ Chaîne validée jusqu\'au tunnel — reste à le monter' - '
') + '➡️ Rien de bloquant détecté, mais le tunnel est ' + 'inactif' + '
Non vérifié faute de ' + f'tunnel monté : {escape(skipped)}. Connectez-vous, ' + 'puis relancez ce test.
') else: verdict = ('
' '✅ Chaîne de connexion complète
')