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
')