from __future__ import annotations import os import subprocess from datetime import datetime, timedelta, timezone from PyQt6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QFrame, QMessageBox, QInputDialog, QLineEdit, QListWidget, QListWidgetItem, QComboBox, QProgressBar, ) from PyQt6.QtCore import Qt, QTimer, QDateTime, QEvent, pyqtSignal from PyQt6.QtGui import QFont, QCloseEvent, QColor from app.core.config import Config from app.core import wireguard as wg_core from app.core import dns as dns_util from app.core import log as conn_log from app.core import history as hist from app.core import shutdown as shutdown_guard from app.core import vpn_session from app.core import updater from app.utils.platform_utils import is_windows from app import __version__ as app_version from app.ui.worker import TaskWorker, ValueWorker from app.ui.access_code_dialog import AccessCodeDialog from app.ui.admin_window import AdminWindow from app.ui.bw_graph import BandwidthGraph from app.ui.history_dialog import HistoryDialog from app.ui import icons from app.ui import theme # Intervalle entre deux re-vérifications de l'accès auprès du serveur, tant # qu'il est cru ouvert. Assez court pour qu'une révocation ne laisse pas # l'utilisateur devant un réseau muet plus d'une minute, assez long pour ne # pas transformer la scrutation de statut (3 s) en martèlement de l'API. _ACCESS_RECHECK = timedelta(seconds=60) # Nombre de sondages sans réponse d'affilée avant de renoncer à croire l'accès # encore ouvert. Un refus venant du serveur, lui, fait autorité et referme du # premier coup ; ceci ne vaut que pour le doute — serveur injoignable, réponse # illisible — où refermer aussitôt ferait clignoter l'état entre « Ouvert » et # « Quarantaine » au moindre hoquet réseau. Trois sondages, soit trois minutes # de silence. _ACCESS_UNKNOWN_BEFORE_CLOSE = 3 # Délai avant de resonder quand le serveur n'a rien dit alors qu'une saisie # devait être proposée. Le tunnel vient de monter : ses routes peuvent n'être # pas encore posées, et la fenêtre surgirait pour un code déjà valide. _PROBE_RETRY_MS = 3000 # Intervalle entre deux interrogations du serveur de mise à jour. La question # ne se posait qu'à l'instant précis où le tunnel montait : un serveur pas # encore joignable à cette seconde-là — routes non posées, quarantaine en # cours — et plus rien n'était proposé de toute la session, sans que rien ne # le dise. Un intervalle unique plutôt que deux rythmes : `check` rend `None` # aussi bien quand le serveur se tait que lorsqu'il n'a rien de neuf, et les # distinguer demanderait de changer sa signature pour un gain de cadence nul. _UPDATE_RECHECK = timedelta(minutes=30) class MainWindow(QMainWindow): # Émis depuis le thread de téléchargement : passer par un signal est la # seule façon sûre de toucher un widget depuis un autre thread que celui # de l'interface. update_progress = pyqtSignal(int, int) def __init__(self, config: Config, parent=None): super().__init__(parent) self._cfg = config self._connecting = False self._connected_since: QDateTime | None = None self._prev_rx: int | None = None self._prev_tx: int | None = None self._auto_reconnect_active = False self._session_id: int | None = None self._worker: TaskWorker | None = None self._ping_worker: ValueWorker | None = None self._is_connected = False self._access_open = False # Échéance annoncée par le serveur pour l'autorisation en cours. None # signifie « aucune échéance connue » — soit aucun accès ouvert, soit # un serveur qui n'en annonce pas — et jamais « expire maintenant ». self._access_until: datetime | None = None # Sondage de l'état de l'accès auprès du serveur. `_probe_prompt` # retient qu'une saisie devra être proposée si le sondage en cours # revient « fermé » : la demande peut arriver alors qu'un sondage # déclenché sans invite est déjà parti. self._probe_worker: ValueWorker | None = None self._probe_prompt = False # Vérification, après un code accepté, que le trafic atteint bien le # réseau distant. Distincte du sondage : celui-ci demande au serveur # ce qu'il a enregistré, celle-là observe ce qui passe réellement. self._reach_worker: ValueWorker | None = None # Date du dernier sondage abouti, et nombre de sondages consécutifs # ayant trouvé fermé un accès qu'on croyait ouvert (cf. # `_access_stale` et `_on_probe_reply`). `_access_message` évite de # répéter la même ligne de journal à chaque re-vérification. self._access_probed_at: datetime | None = None self._access_denials = 0 self._access_message = "" # Le serveur connaît-il un pair à notre adresse de tunnel ? `None` # tant qu'il n'a pas répondu. Ce constat remplace l'ancien « un # identifiant VPN est-il configuré ? », qui prétendait répondre à la # question depuis le poste alors que seul le serveur la tranche. self._access_enrolled: bool | None = None # Un seul nouvel essai quand le serveur reste muet alors qu'une saisie # devait être proposée : au-delà, le bouton prend le relais plutôt que # de faire tourner un sondage en boucle. self._probe_retried = False # Conservée après fermeture : la fenêtre reste parentée à celle-ci, et # son thread doit être attendu avant destruction (cf. prepare_quit). self._access_dialog: AccessCodeDialog | None = None # Mise à jour : proposée, jamais imposée. `_update_dismissed` ne dure # que la session — une version écartée d'un clic distrait se # represente au prochain démarrage plutôt que d'être perdue. self._update_worker: ValueWorker | None = None self._update = None self._update_dismissed = False # Date du dernier échange abouti avec le serveur de mise à jour, et # non de la dernière tentative : un échec doit être réessayé vite, une # réponse obtenue peut attendre. self._update_checked_at: datetime | None = None self._reconnect_failures = 0 self._quitting = False self._warned_foreign: set[str] = set() self.setWindowTitle("WGSecure") self.setFixedWidth(430) self.setWindowFlags( Qt.WindowType.Window | Qt.WindowType.WindowTitleHint | Qt.WindowType.WindowCloseButtonHint | Qt.WindowType.WindowMinimizeButtonHint ) self.setWindowIcon(icons.icon_app()) self._build_ui() # Repliée par défaut : au lancement, seuls statut et action de # connexion sont utiles — le détail ne sert qu'à qui va le chercher. self._details_expanded = False self._details.setVisible(False) self._sync_window_height() # Timers self._status_timer = QTimer(self) self._status_timer.timeout.connect(self._refresh_status) self._status_timer.start(3000) self._clock_timer = QTimer(self) self._clock_timer.timeout.connect(self._tick_duration) self._clock_timer.start(1000) self._bw_timer = QTimer(self) self._bw_timer.timeout.connect(self._update_bw) self._bw_timer.start(3000) self._ping_timer = QTimer(self) self._ping_timer.timeout.connect(self._update_ping) self._ping_timer.start(10000) self._reconnect_timer = QTimer(self) self._reconnect_timer.timeout.connect(self._check_reconnect) self._refresh_status() self._reload_log() self._reload_profiles() self._warn_foreign_interfaces() if self._cfg.get("ui", "auto_connect_on_startup") and self._cfg.configured: QTimer.singleShot(500, self._on_connect) # ------------------------------------------------------------------ # # Construction UI # # Ordre pensé pour l'usage réel : statut et action de connexion sont # immédiats (pas besoin de traverser la config ou le journal pour les # atteindre) ; viennent ensuite les valeurs dynamiques (transfert, # handshake), puis la config statique — consultée rarement une fois # réglée — et enfin bande passante / journal, les plus techniques. # ------------------------------------------------------------------ # def _build_ui(self): central = QWidget() central.setStyleSheet(f"background: {theme.BG};") self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) root.addWidget(self._build_header()) root.addWidget(self._build_update_banner()) root.addWidget(self._build_status_action()) root.addWidget(self._build_details()) # ── Pied de page ───────────────────────────────────────────────── root.addWidget(self._build_footer()) def _build_update_banner(self) -> QWidget: """Bandeau proposant une mise à jour, masqué le reste du temps. Placé sous l'en-tête et non dans le panneau de détails : une version disponible se rate si elle attend qu'on déplie quelque chose. Il n'apparaît qu'une fois le serveur consulté, ne se réaffiche pas dans la session si l'utilisateur l'écarte, et n'installe rien sans clic. """ self._update_banner = QWidget() self._update_banner.setStyleSheet( f"background: {theme.WARN_BG}; border-radius: 4px;") # Non parenté au bandeau : il reçoit plus bas un layout vertical qui # empile cette ligne et la barre de progression. row = QHBoxLayout() row.setContentsMargins(10, 6, 10, 6) row.setSpacing(6) self._update_label = QLabel("") self._update_label.setWordWrap(True) # Blanc plutôt que le jaune du bandeau : sur ce fond, le texte # d'avertissement passait pour du décor et se lisait mal. self._update_label.setStyleSheet( f"color: {theme.TEXT}; font-size: 11px; background: transparent;") row.addWidget(self._update_label, 1) self._btn_update = QPushButton("MAJ") self._btn_update.setFixedHeight(24) self._btn_update.setStyleSheet( f"QPushButton {{ background: {theme.OK_SOLID}; color: white;" " font-weight: bold; border: none; border-radius: 4px;" " padding: 0 14px; }" f"QPushButton:hover {{ background: {theme.OK_HOVER}; }}" f"QPushButton:disabled {{ background: {theme.BG_RAISED};" f" color: {theme.TEXT_FAINT}; }}") self._btn_update.clicked.connect(self._install_update) row.addWidget(self._btn_update) # Report : une croix, pas un mot. Elle dit « pas maintenant » sans # entrer en concurrence de lecture avec l'action principale. btn_later = QPushButton("✕") btn_later.setFixedSize(24, 24) btn_later.setToolTip("Reporter — la proposition reviendra au prochain " "démarrage") btn_later.setStyleSheet( f"QPushButton {{ background: {theme.FAIL_SOLID}; color: white;" " font-weight: bold; border: none; border-radius: 4px; }" f"QPushButton:hover {{ background: {theme.FAIL_HOVER}; }}") btn_later.clicked.connect(self._dismiss_update) row.addWidget(btn_later) # Le téléchargement porte sur une centaine de mégaoctets : sans rien # à l'écran, l'application paraissait figée et l'utilisateur ne # pouvait pas distinguer une attente d'un échec. self._update_progress = QProgressBar() self._update_progress.setFixedHeight(6) self._update_progress.setTextVisible(False) self._update_progress.setStyleSheet( f"QProgressBar {{ border: none; border-radius: 3px;" f" background: {theme.BG}; }}" f"QProgressBar::chunk {{ background: {theme.OK_SOLID};" " border-radius: 3px; }") self._update_progress.setVisible(False) outer = QVBoxLayout() outer.setContentsMargins(0, 0, 0, 0) outer.setSpacing(4) inner = QWidget() inner.setLayout(row) inner.setStyleSheet("background: transparent;") outer.addWidget(inner) outer.addWidget(self._update_progress) self._update_banner.setLayout(outer) self.update_progress.connect(self._on_download_progress) self._update_banner.setVisible(False) return self._update_banner def _on_download_progress(self, recu: int, total: int): """Avancement du téléchargement, reçu du thread qui le mène.""" if total > 0: self._update_progress.setRange(0, total) self._update_progress.setValue(recu) self._btn_update.setText(f"{100 * recu // total} %") else: # Total inconnu : barre en va-et-vient plutôt qu'une jauge figée # à zéro, qui se lit comme une panne. self._update_progress.setRange(0, 0) def _update_due(self) -> bool: """Faut-il (re)demander au serveur s'il propose une version ? Deux rythmes : rapide tant que le serveur n'a rien répondu, lent une fois qu'il a parlé. Une mise à jour déjà proposée, ou écartée pour la session, ne se redemande pas. """ if self._update is not None or self._update_dismissed: return False if self._update_checked_at is None: return True return (datetime.now(timezone.utc) - self._update_checked_at) >= _UPDATE_RECHECK def _check_update(self): """Interroge le serveur, sans bloquer ni déranger si rien n'est trouvé.""" if self._update_worker is not None or self._update_dismissed: return self._update_worker = ValueWorker(updater.check, self._cfg, app_version, parent=self) self._update_worker.done.connect(self._on_update_checked) self._update_worker.finished.connect(self._update_worker.deleteLater) self._update_worker.start() def _on_update_checked(self, found): self._update_worker = None self._update = found self._update_checked_at = datetime.now(timezone.utc) if found is None or self._update_dismissed: return taille = f"{found.size / 1048576:.0f} Mo" if found.size else "" self._update_label.setText( f"⬆️ Version {found.version} disponible" + (f" ({taille})" if taille else "")) self._update_banner.setVisible(True) self._sync_window_height() self._add_log(f"Version {found.version} disponible sur le serveur", "info") def _dismiss_update(self): """Écarte la proposition pour cette session, sans la refuser à jamais. Rien n'est écrit : au prochain démarrage la question se repose, et une version ignorée par mégarde n'est pas perdue pour toujours. """ self._update_dismissed = True self._update_banner.setVisible(False) self._sync_window_height() def _install_update(self): """Télécharge puis installe, après confirmation explicite.""" if self._update is None or self._update_worker is not None: return reponse = QMessageBox.question( self, "Mettre à jour WGSecure", f"Installer la version {self._update.version} ?\n\n" f"Le fichier sera téléchargé depuis le serveur, puis vérifié " f"contre l'empreinte qu'il annonce avant d'être installé.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No) if reponse != QMessageBox.StandardButton.Yes: return self._btn_update.setEnabled(False) self._btn_update.setText("…") self._update_progress.setRange(0, 0) # va-et-vient tant que rien self._update_progress.setVisible(True) # n'est encore reçu self._sync_window_height() self._update_worker = ValueWorker(self._download_and_apply, self._update, parent=self) self._update_worker.done.connect(self._on_update_installed) self._update_worker.finished.connect(self._update_worker.deleteLater) self._update_worker.start() def _download_and_apply(self, info): """Exécuté hors du thread graphique : 125 Mo, puis remplacement. Rend le message à montrer, ou le relaie préfixé — `ValueWorker` avale les exceptions en rendant `None`, ce qui ne dirait pas ce qui a échoué. """ try: if is_windows(): # Le service fait tout : il télécharge et vérifie lui-même, # pour qu'aucun compte du poste ne puisse lui faire installer # un fichier de son choix avec les privilèges du système. return updater.apply_via_helper(self._cfg) cible = updater.current_binary() archive = updater.download( self._cfg, info, progress=lambda recu, total: self.update_progress.emit(recu, total), dest_dir=os.path.dirname(cible) if cible else "") return updater.apply_update(archive) except Exception as exc: # noqa: BLE001 — remonté à l'UI return f"ERREUR: {exc}" def _on_update_installed(self, message): self._update_worker = None self._btn_update.setEnabled(True) self._btn_update.setText("MAJ") self._update_progress.setVisible(False) self._sync_window_height() texte = message or "Échec inattendu de la mise à jour." if texte.startswith("ERREUR: "): texte = texte[8:] self._add_log(f"Mise à jour : {texte}", "error") QMessageBox.warning(self, "Mise à jour impossible", texte) return self._add_log(texte, "success") self._update_banner.setVisible(False) self._sync_window_height() QMessageBox.information(self, "Mise à jour", texte) # Sous Windows, l'installeur tourne déjà et attend que les fichiers se # libèrent : rester ouvert le ferait patienter jusqu'à son délai, puis # échouer. On rend donc la main tout de suite — `prepare_quit` démonte # le tunnel et restaure le DNS avant de fermer, ce qui laisse aussi le # service du tunnel arrêtable par l'installeur. if is_windows(): self._add_log("Fermeture pour laisser l'installeur remplacer les " "fichiers", "warning") self.prepare_quit() QApplication.quit() return # Sous Linux, le binaire est déjà remplacé et l'application continue de # tourner sur l'ancien, que le noyau garde en vie. Rien n'oblige à # fermer — mais le dire sans rien proposer laissait l'utilisateur # devant une version périmée sans savoir quoi faire. cible = updater.current_binary() if not cible: return if QMessageBox.question( self, "Redémarrer WGSecure", "La nouvelle version est en place. Redémarrer maintenant ?\n\n" "Le tunnel sera coupé le temps du redémarrage.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.Yes ) != QMessageBox.StandardButton.Yes: return try: # Lancé détaché : sans cela le nouveau processus mourrait avec # celui-ci, qui s'apprête justement à disparaître. subprocess.Popen([cible], start_new_session=True, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except OSError as exc: QMessageBox.warning(self, "Redémarrage impossible", str(exc)) return self.prepare_quit() QApplication.quit() # ── Section repliable : tout ce qui n'est pas statut/action ────────── def _build_details(self) -> QWidget: self._details = QWidget() d = QVBoxLayout(self._details) d.setContentsMargins(0, 0, 0, 0) d.setSpacing(0) d.addWidget(self._hsep()) d.addWidget(self._build_dynamic_details()) d.addWidget(self._hsep()) d.addWidget(self._build_config_summary()) d.addWidget(self._hsep()) # ── Bande passante ─────────────────────────────────────────────── d.addWidget(self._section_header("📊 Bande passante")) self._bw_graph = BandwidthGraph() self._bw_graph.setFixedHeight(75) d.addWidget(self._bw_graph) d.addWidget(self._hsep()) # ── Journal ─────────────────────────────────────────────────────── d.addWidget(self._section_header("📋 Journal", btn_text="Effacer", btn_slot=self._clear_log)) self._log_list = QListWidget() self._log_list.setFixedHeight(78) self._log_list.setStyleSheet( f"QListWidget {{ background: {theme.BG_SUNKEN}; color: {theme.ACCENT_LIGHT};" " border: none; font-size: 10px; font-family: Courier; }" "QListWidget::item { padding: 1px 6px; }" ) self._log_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded) d.addWidget(self._log_list) d.addWidget(self._hsep()) return self._details # ── En-tête : identité + profil ───────────────────────────────────── def _build_header(self) -> QWidget: header = QWidget() header.setFixedHeight(64) header.setStyleSheet(f"background: {theme.BG_SUNKEN};") h = QHBoxLayout(header) h.setContentsMargins(14, 8, 14, 8) self._icon_label = QLabel() self._icon_label.setFixedSize(36, 36) self._icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter) h.addWidget(self._icon_label) title_lbl = QLabel("WGSecure") title_lbl.setStyleSheet( f"color: {theme.TEXT}; font-size: 16px; font-weight: bold; background: transparent;" ) h.addWidget(title_lbl) h.addStretch() profile_col = QVBoxLayout() profile_col.setSpacing(1) profile_lbl = QLabel("Profil actif") profile_lbl.setAlignment(Qt.AlignmentFlag.AlignRight) profile_lbl.setStyleSheet( f"color: {theme.TEXT_FAINT}; font-size: 9px; background: transparent;" ) profile_col.addWidget(profile_lbl) self._profile_combo = QComboBox() self._profile_combo.setFixedHeight(22) self._profile_combo.setMinimumWidth(150) self._profile_combo.setStyleSheet( f"QComboBox {{ background: {theme.BG_RAISED}; color: {theme.ACCENT_LIGHT};" " border: none; border-radius: 3px; font-size: 10px; padding: 0 6px; }" "QComboBox::drop-down { border: none; }" f"QComboBox QAbstractItemView {{ background: {theme.BG_RAISED}; color: {theme.TEXT}; }}" ) self._profile_combo.currentIndexChanged.connect(self._on_profile_changed) profile_col.addWidget(self._profile_combo) h.addLayout(profile_col) return header # ── Statut + action principale ─────────────────────────────────────── def _build_status_action(self) -> QWidget: status_area = QWidget() status_area.setStyleSheet(f"background: {theme.BG_RAISED};") s = QVBoxLayout(status_area) s.setContentsMargins(14, 12, 14, 12) s.setSpacing(8) top_row = QHBoxLayout() top_row.setSpacing(8) status_col = QVBoxLayout() status_col.setSpacing(1) self._status_badge = QLabel("● Déconnecté") self._status_badge.setStyleSheet( f"color: {theme.FAIL_TEXT}; font-size: 15px; font-weight: bold; background: transparent;" ) status_col.addWidget(self._status_badge) self._duration_label = QLabel("") self._duration_label.setStyleSheet( f"color: {theme.TEXT_FAINT}; font-size: 11px; background: transparent;" ) status_col.addWidget(self._duration_label) top_row.addLayout(status_col) top_row.addStretch() # Badge de ping : santé de cette connexion, donc juste à côté de son # statut plutôt que dans l'en-tête d'identité. self._ping_badge = QLabel("— ms") self._ping_badge.setFixedSize(66, 24) self._ping_badge.setAlignment(Qt.AlignmentFlag.AlignCenter) self._ping_badge.setStyleSheet( f"border-radius: 12px; font-size: 10px; font-weight: bold;" f" {theme.PING_CSS['offline']}" ) top_row.addWidget(self._ping_badge, 0, Qt.AlignmentFlag.AlignVCenter) s.addLayout(top_row) self._btn_connect = QPushButton("Se connecter") self._btn_connect.setFixedHeight(44) self._btn_connect.setStyleSheet( f"QPushButton {{ background: {theme.OK_SOLID}; color: {theme.TEXT}; font-size: 14px;" " font-weight: bold; border-radius: 6px; border: none; }" "QPushButton:hover { background: #27ae60; }" f"QPushButton:disabled {{ background: {theme.BG_RAISED}; color: {theme.TEXT_FAINT}; }}" ) self._btn_connect.clicked.connect(self._on_connect) s.addWidget(self._btn_connect) # Saisie du code d'accès, atteignable à tout moment. Le tunnel monte # en quarantaine et l'autorisation obtenue expire d'elle-même : sans # ce bouton, un code refusé, une reconnexion automatique ou une # autorisation échue ne laissaient d'autre issue que se déconnecter # puis se reconnecter. Masqué tant qu'il n'y a rien à saisir, pour ne # rien ajouter aux installations sans authentification serveur. self._btn_access = QPushButton("🔐 Saisir le code d'accès") self._btn_access.setFixedHeight(32) self._btn_access.clicked.connect(self.request_access_code) self._btn_access.setVisible(False) s.addWidget(self._btn_access) self._style_access_button(False) # Repli/dépli du détail (transfert, config, bande passante, journal) : # rattaché à l'action qu'il détaille, pas à l'en-tête. self._btn_toggle_details = QPushButton("▾ Afficher le détail") self._btn_toggle_details.setFixedHeight(20) self._btn_toggle_details.setStyleSheet( f"QPushButton {{ background: transparent; color: {theme.TEXT_FAINT};" " font-size: 10px; border: none; }" f"QPushButton:hover {{ color: {theme.TEXT}; }}" ) self._btn_toggle_details.clicked.connect(self._toggle_details) s.addWidget(self._btn_toggle_details) return status_area # ── Détails dynamiques : transfert + handshake ─────────────────────── def _build_dynamic_details(self) -> QWidget: frame = QWidget() frame.setStyleSheet(f"background: {theme.BG};") dl = QHBoxLayout(frame) dl.setContentsMargins(14, 8, 14, 8) dl.setSpacing(16) self._info_labels: dict[str, QLabel] = {} for key, icon_txt, label in [ ("rx_tx", "↕️", "Transfert"), ("handshake", "🤝", "Handshake"), ]: col = QVBoxLayout() col.setSpacing(2) hdr = QLabel(f"{icon_txt} {label}") hdr.setStyleSheet( f"color: {theme.TEXT_FAINT}; font-size: 10px; background: transparent;" ) col.addWidget(hdr) val = self._lbl("—", color=theme.ACCENT_LIGHT, size=13) val.setWordWrap(True) self._info_labels[key] = val col.addWidget(val) dl.addLayout(col, 1) return frame # ── Config statique (repliée visuellement) ─────────────────────────── def _build_config_summary(self) -> QWidget: frame = QFrame() frame.setStyleSheet(f"background: {theme.BG};") il = QVBoxLayout(frame) il.setContentsMargins(14, 8, 14, 8) il.setSpacing(4) for key, icon_txt, label in [ ("server", "🌐", "Serveur"), ("iface", "🔌", "Interface"), ("addr", "📍", "Adresse IP"), ("access", "🔐", "Accès distant"), ]: rw = QWidget() rw.setStyleSheet("background: transparent;") rl = QHBoxLayout(rw) rl.setContentsMargins(0, 0, 0, 0) rl.setSpacing(5) rl.addWidget(self._lbl(icon_txt, fixed=18, color=theme.TEXT_FAINT, size=10)) rl.addWidget(self._lbl(f"{label} :", fixed=90, color=theme.TEXT_FAINT, size=11)) val = self._lbl("—", color=theme.TEXT_MUTED, size=11) val.setWordWrap(True) self._info_labels[key] = val rl.addWidget(val, 1) il.addWidget(rw) return frame # ── Pied de page : accès secondaires ───────────────────────────────── def _build_footer(self) -> QWidget: btn_area = QWidget() btn_area.setStyleSheet(f"background: {theme.BG_SUNKEN};") bl = QHBoxLayout(btn_area) bl.setContentsMargins(14, 10, 14, 12) bl.setSpacing(6) btn_admin = QPushButton("⚙️ Administrateur") btn_admin.setFixedHeight(30) btn_admin.setStyleSheet( f"QPushButton {{ background: {theme.BG_RAISED}; color: {theme.TEXT}; border-radius: 6px;" " font-size: 11px; border: none; }" f"QPushButton:hover {{ background: {theme.ACCENT_HOVER}; }}" ) btn_admin.clicked.connect(self._open_admin) bl.addWidget(btn_admin) btn_history = QPushButton("📊 Historique") btn_history.setFixedHeight(30) btn_history.setStyleSheet( f"QPushButton {{ background: {theme.BG_RAISED}; color: {theme.TEXT}; border-radius: 6px;" " font-size: 11px; border: none; }" f"QPushButton:hover {{ background: {theme.ACCENT_HOVER}; }}" ) btn_history.clicked.connect(self._open_history) bl.addWidget(btn_history) return btn_area # ── Repli/dépli du détail ───────────────────────────────────────────── def _toggle_details(self): # `QWidget.isVisible()` reflète la visibilité effective à l'écran, # qui dépend de toute la chaîne de parents : avant que la fenêtre # elle-même ne soit affichée, elle vaut toujours False. Un drapeau # explicite évite ce piège. self._details_expanded = not self._details_expanded self._details.setVisible(self._details_expanded) self._btn_toggle_details.setText( "▴ Masquer le détail" if self._details_expanded else "▾ Afficher le détail" ) self._sync_window_height() def _style_access_button(self, expected: bool): """Met le bouton en avant seulement quand un code est réellement dû.""" bg, hover = ((theme.WARN_SOLID, "#d68910") if expected else (theme.BG_RAISED, theme.ACCENT_HOVER)) weight = "bold" if expected else "normal" self._btn_access.setStyleSheet( f"QPushButton {{ background: {bg}; color: {theme.TEXT};" f" font-size: 12px; font-weight: {weight}; border-radius: 6px;" " border: none; }" f"QPushButton:hover {{ background: {hover}; }}" ) def _sync_window_height(self): """Réajuste la hauteur fixe de la fenêtre à son contenu visible. La fenêtre reste de taille fixe (pas de redimensionnement manuel), mais cette taille change avec le repli/dépli du détail et avec l'apparition du bouton de code d'accès — il faut donc lever la contrainte précédente avant de laisser le layout recalculer sa hauteur naturelle, puis la reverrouiller. Le rendre visible ne suffit pas à mettre les layouts à jour : Qt poste un `LayoutRequest` traité au tour de boucle suivant, et mesurer avant ce traitement rend une hauteur en retard d'un cycle. La fenêtre gardait alors sa taille précédente et rognait le widget qui venait d'apparaître — le bouton « Saisir le code d'accès » existait, mais sous le bord inférieur. D'où la remise à plat explicite des layouts avant toute mesure. """ self.setMinimumHeight(0) self.setMaximumHeight(16777215) central = self.centralWidget() QApplication.sendPostedEvents(None, QEvent.Type.LayoutRequest) central.layout().activate() central.adjustSize() self.adjustSize() self.setFixedHeight(self.sizeHint().height()) # ── Helpers UI ─────────────────────────────────────────────────────── def _lbl(self, text: str, fixed: int = 0, color: str = theme.TEXT, size: int = 12) -> QLabel: lbl = QLabel(text) lbl.setStyleSheet( f"color: {color}; font-size: {size}px; background: transparent;" ) if fixed: lbl.setFixedWidth(fixed) return lbl def _hsep(self) -> QFrame: f = QFrame() f.setFrameShape(QFrame.Shape.HLine) f.setFixedHeight(1) f.setStyleSheet(f"background: {theme.BG_SUNKEN};") return f def _section_header(self, title: str, btn_text: str = "", btn_slot=None) -> QWidget: w = QWidget() w.setFixedHeight(24) w.setStyleSheet(f"background: {theme.BG_SUNKEN};") hl = QHBoxLayout(w) hl.setContentsMargins(12, 0, 12, 0) t = QLabel(title) t.setStyleSheet(f"color: {theme.TEXT_FAINT}; font-size: 10px; font-weight: bold;" " background: transparent;") hl.addWidget(t) hl.addStretch() if btn_text and btn_slot: b = QPushButton(btn_text) b.setFixedHeight(18) b.setStyleSheet( f"QPushButton {{ background: transparent; color: {theme.TEXT_FAINT};" " font-size: 9px; border: none; padding: 0 4px; }" f"QPushButton:hover {{ color: {theme.TEXT}; }}" ) b.clicked.connect(btn_slot) hl.addWidget(b) return w # ------------------------------------------------------------------ # # Profils # ------------------------------------------------------------------ # def _reload_profiles(self): self._profile_combo.blockSignals(True) self._profile_combo.clear() profiles = self._cfg.list_profiles() active = self._cfg.active_profile if active not in profiles: profiles = [active] + profiles self._profile_combo.addItems(profiles) idx = self._profile_combo.findText(active) if idx >= 0: self._profile_combo.setCurrentIndex(idx) self._profile_combo.blockSignals(False) def _on_profile_changed(self, index: int): name = self._profile_combo.itemText(index) if not name or name == self._cfg.active_profile: return if self._cfg.load_profile(name): self._add_log(f"Profil chargé : {name}", "info") self._reload_profiles() self._refresh_status() # ------------------------------------------------------------------ # # Journal # ------------------------------------------------------------------ # def _reload_log(self): self._log_list.clear() for ev in conn_log.get_events(30): self._add_log_item(ev["ts"], ev["msg"], ev["kind"]) self._log_list.scrollToBottom() def _clear_log(self): conn_log.clear() self._log_list.clear() def _add_log(self, msg: str, kind: str = "info"): conn_log.log_event(msg, kind) ts = QDateTime.currentDateTime().toString("yyyy-MM-dd hh:mm:ss") self._add_log_item(ts, msg, kind) def _add_log_item(self, ts: str, msg: str, kind: str): color = theme.LOG_COLORS.get(kind, theme.ACCENT_LIGHT) item = QListWidgetItem(f"[{ts}] {msg}") item.setForeground(QColor(color)) # Ligne trop longue pour tenir dans le panneau : barre de défilement # horizontale (ci-dessus) + tooltip pour lire le message complet # sans avoir à défiler. item.setToolTip(msg) self._log_list.addItem(item) self._log_list.scrollToBottom() # ------------------------------------------------------------------ # # Timers # ------------------------------------------------------------------ # def _tick_duration(self): if self._connected_since is None: return secs = self._connected_since.secsTo(QDateTime.currentDateTime()) h, rem = divmod(secs, 3600) m, s = divmod(rem, 60) self._duration_label.setText(f"Connecté depuis {h:02d}h {m:02d}m {s:02d}s") def _update_bw(self): iface = self._cfg.wg.get("interface_name", "wgs0") result = wg_core.get_interface_bytes(iface) if result is None or self._prev_rx is None: self._prev_rx, self._prev_tx = (result or (0, 0)) return rx_now, tx_now = result dt = 3.0 rx_bps = max(0, rx_now - self._prev_rx) / dt tx_bps = max(0, tx_now - self._prev_tx) / dt self._prev_rx, self._prev_tx = rx_now, tx_now # `_is_connected` est rafraîchi par _refresh_status : inutile de # relancer un `wg show` (un sous-processus) toutes les 3 secondes. self._bw_graph.push(rx_bps if self._is_connected else 0, tx_bps if self._is_connected else 0) def _update_ping(self): host = self._cfg.wg.get("server_endpoint", "") if not host or not self._is_connected: self._set_ping_badge(None, "— ms", "offline") return if self._ping_worker is not None and self._ping_worker.isRunning(): return # `ping` bloque jusqu'à 4 s : exécuté dans le thread Qt, il faisait # saccader toute l'interface toutes les 10 secondes. self._ping_worker = ValueWorker(wg_core.ping_or_handshake, self._cfg, 2, parent=self) self._ping_worker.done.connect(self._on_ping_done) # Détruire le QThread depuis `finished` et non depuis `done` : `done` # est émis avant la sortie de run(), un deleteLater() à ce moment # provoque « QThread: Destroyed while thread is still running ». self._ping_worker.finished.connect(self._ping_worker.deleteLater) self._ping_worker.start() def _on_ping_done(self, result): self._ping_worker = None ms, source = result if result else (None, "down") if not self._is_connected: self._set_ping_badge(None, "— ms", "offline") elif source == "down": self._set_ping_badge(None, "hors ligne", "bad") elif source == "stale": self._set_ping_badge(None, "handshake ancien", "bad") elif source == "handshake": # ICMP bloqué par le serveur/pare-feu mais handshake WireGuard # récent : le tunnel fonctionne, la latence réelle est juste # invisible depuis ce test. self._set_ping_badge(None, f"actif ({ms}s)", "medium") elif ms < 50: self._set_ping_badge(ms, f"{ms} ms", "good") elif ms < 200: self._set_ping_badge(ms, f"{ms} ms", "medium") else: self._set_ping_badge(ms, f"{ms} ms", "bad") def _set_ping_badge(self, _ms, text: str, css_key: str): self._ping_badge.setText(text) self._ping_badge.setStyleSheet( "border-radius: 12px; font-size: 10px; font-weight: bold; " + theme.PING_CSS[css_key] ) def _check_reconnect(self): if not self._cfg.get("ui", "auto_reconnect") or self._quitting: return if self._worker is not None: return # une (dé)connexion est déjà en cours if not self._cfg.configured or self._is_connected: self._reconnect_failures = 0 return if self._reconnect_failures >= 5: # Arrêt du harcèlement : chaque tentative demande des privilèges. if self._reconnect_failures == 5: self._add_log("Auto-reconnexion suspendue après 5 échecs", "error") self._reconnect_failures += 1 return self._add_log("Auto-reconnexion…", "warning") self._start_connect(silent=True) # ------------------------------------------------------------------ # # Statut WireGuard # ------------------------------------------------------------------ # def _access_valid(self) -> bool: """Vrai tant que l'autorisation obtenue court encore. L'échéance vient du serveur et de lui seul : le client la relit, il ne la calcule pas — sa durée est un réglage serveur, modifiable sans que le poste en sache rien. Un serveur qui n'annonce aucune échéance laisse l'autorisation valable jusqu'à la chute du tunnel, faute de terme connu ; en inventer un fermerait un accès encore ouvert. """ if not self._access_open: return False if self._access_until is None: return True return datetime.now(timezone.utc) < self._access_until def _access_stale(self) -> bool: """L'état de l'accès mérite-t-il d'être redemandé au serveur ? L'échéance annoncée n'est pas la seule façon dont un accès se referme : le serveur peut révoquer avant terme — redémarrage, purge de sa table de sessions, décision d'un administrateur — sans que rien n'en parvienne au poste. Le client ne sondait qu'au retour du tunnel : il affichait donc « Ouvert ✓ » et masquait le bouton devant un réseau déjà muet, jusqu'à la chute du tunnel. Sans échéance annoncée, cet état pouvait durer indéfiniment. """ if self._access_probed_at is None: return True return (datetime.now(timezone.utc) - self._access_probed_at) >= _ACCESS_RECHECK def _remaining_text(self) -> str: """Temps restant avant l'échéance, tel qu'on le lit d'un coup d'œil. Rafraîchi au rythme du sondage de statut (3 s) : la minute affichée est donc juste, ce qui suffit pour une fenêtre qui se compte en heures. Aucun minuteur supplémentaire pour l'entretenir. """ seconds = int((self._access_until - datetime.now(timezone.utc)).total_seconds()) # Arrondi à la minute, et non tronqué : une fenêtre de 8 h qui vient # de s'ouvrir doit se lire « 8 h 00 min », pas « 7 h 59 min ». minutes = (seconds + 30) // 60 if minutes >= 60: return f"{minutes // 60} h {minutes % 60:02d} min" if minutes >= 1: return f"{minutes} min" return f"{max(seconds, 0)} s" def _access_state(self) -> str: """État de l'autorisation réseau accordée par le serveur. Distinct de l'état du tunnel : monté, celui-ci ne mène nulle part tant que le serveur n'a pas levé la quarantaine. Sans cette ligne, un utilisateur ayant refusé la saisie du code verrait « Connecté » et un réseau inexplicablement muet. Sans identifiant VPN, l'état est « non vérifié » et non « non géré » : le client n'en sait rien. Il l'affirmait, et se trompait — le serveur peut parfaitement tenir ce pair en quarantaine pendant que la configuration locale, elle, ne mentionne aucun serveur. """ if not self._is_connected: return "—" if self._access_valid(): if self._access_until is None: return "Ouvert ✓" fin = self._access_until.astimezone() # La date n'est rappelée que si l'échéance n'est pas aujourd'hui : # la durée est réglable jusqu'à 168 h côté serveur, « jusqu'à # 02:30 » serait alors ambigu. quand = (f"{fin:%H:%M}" if fin.date() == datetime.now().date() else f"{fin:%d/%m à %H:%M}") return f"Ouvert ✓ jusqu'à {quand} (reste {self._remaining_text()})" if self._access_open: # L'autorisation a couru puis expiré : le dire, plutôt que de # laisser croire qu'aucun code n'a jamais été saisi. return "Autorisation expirée (code requis)" if self._access_enrolled is False: # Le serveur affirme ne connaître aucun pair à cette adresse : # aucun code n'ouvrira quoi que ce soit ici. return "Non géré par le serveur" if self._access_enrolled is None: # Rien demandé, ou serveur muet : le client n'en sait rien et ne # prétend donc rien. Il l'affirmait autrefois d'après sa seule # configuration locale, et se trompait. return "Non vérifié" return "Quarantaine (code requis)" def _refresh_status(self): wg = self._cfg.wg srv = wg.get("server_endpoint", "") port = wg.get("server_port", "") self._info_labels["server"].setText(f"{srv}:{port}" if srv else "—") self._info_labels["iface"].setText(wg.get("interface_name", "wgs0")) self._info_labels["addr"].setText(wg.get("client_address", "—")) # Auto-reconnect timer if self._cfg.get("ui", "auto_reconnect"): interval = int(self._cfg.get("ui", "reconnect_interval") or 15) * 1000 if not self._reconnect_timer.isActive(): self._reconnect_timer.start(interval) else: self._reconnect_timer.stop() try: info = wg_core.get_status_info(self._cfg) connected = info["connected"] except Exception: connected = False info = {} was_connected = self._is_connected self._is_connected = connected if connected: self._reconnect_failures = 0 if not was_connected: # Tunnel monté à l'instant — démarrage de l'application sur un # tunnel déjà actif, reconnexion automatique, ou montage fait # hors de l'application. Demander au serveur avant de conclure # quoi que ce soit : lui seul sait si l'accès est encore ouvert. self._probe_access(prompt_if_closed=False) # Le serveur de mise à jour n'est joignable que par le # tunnel : la question ne se pose qu'une fois celui-ci monté. QTimer.singleShot(2000, self._check_update) elif self._update_due(): self._check_update() elif self._access_open and self._access_stale(): # Accès cru ouvert : le reconfirmer périodiquement, une # révocation côté serveur n'étant annoncée par rien. self._probe_access(prompt_if_closed=False) else: # Tunnel tombé : le client oublie ce qu'il croyait savoir. Le # serveur, lui, n'oublie rien — l'autorisation porte sur l'adresse # du pair, qui ne change pas d'une reconnexion à l'autre, et elle # court jusqu'à son échéance. D'où le sondage au retour du tunnel # plutôt qu'une remise à zéro tenue pour définitive : c'est cette # confusion qui faisait réclamer un code déjà validé. self._access_open = False self._access_until = None self._access_probed_at = None self._access_denials = 0 self._access_message = "" self._probe_retried = False self._access_enrolled = None self._info_labels["access"].setText(self._access_state()) # Le bouton n'apparaît que lorsqu'il y a réellement un code à saisir ; # sa hauteur entre et sort du calcul de la fenêtre, d'où le # réajustement. `isHidden()` et non `isVisible()` : cette dernière est # fausse dès que la fenêtre est repliée dans le systray, ce qui # relancerait le calcul de hauteur à chaque cycle de rafraîchissement. needed = self.access_code_needed() if needed == self._btn_access.isHidden(): self._btn_access.setVisible(needed) self._sync_window_height() if needed: # Un code est attendu de façon certaine quand le serveur a # reconnu ce pair ; sinon le bouton reste disponible mais # discret, pour ne pas réclamer une action à une installation # sans serveur d'authentification. self._style_access_button(self._access_enrolled is True) if connected: if self._connected_since is None: self._connected_since = QDateTime.currentDateTime() self._status_badge.setText("● Connecté") self._status_badge.setStyleSheet( f"color: {theme.OK_TEXT}; font-size: 15px; font-weight: bold; background: transparent;" ) self._btn_connect.setText("Se déconnecter") self._btn_connect.setStyleSheet( f"QPushButton {{ background: {theme.FAIL_SOLID}; color: {theme.TEXT}; font-size: 14px;" " font-weight: bold; border-radius: 6px; border: none; }" "QPushButton:hover { background: #c0392b; }" ) self._info_labels["rx_tx"].setText( f"↓ {info.get('rx_bytes','—')} ↑ {info.get('tx_bytes','—')}" ) self._info_labels["handshake"].setText(info.get("last_handshake", "—")) self._duration_label.setStyleSheet( f"color: {theme.OK_TEXT}; font-size: 11px; background: transparent;" ) self.setWindowIcon(icons.icon_connected()) else: if self._connected_since is not None: self._connected_since = None self._duration_label.setText("") self._bw_graph.reset() self._prev_rx = self._prev_tx = None self._status_badge.setText("● Déconnecté") self._status_badge.setStyleSheet( f"color: {theme.FAIL_TEXT}; font-size: 15px; font-weight: bold; background: transparent;" ) self._btn_connect.setText("Se connecter") self._btn_connect.setStyleSheet( f"QPushButton {{ background: {theme.OK_SOLID}; color: {theme.TEXT}; font-size: 14px;" " font-weight: bold; border-radius: 6px; border: none; }" "QPushButton:hover { background: #27ae60; }" ) self._info_labels["rx_tx"].setText("—") self._info_labels["handshake"].setText("—") self._duration_label.setStyleSheet( f"color: {theme.TEXT_FAINT}; font-size: 11px; background: transparent;" ) self.setWindowIcon(icons.icon_disconnected()) pix = (icons.icon_connected() if connected else icons.icon_disconnected()).pixmap(36, 36) self._icon_label.setPixmap(pix) # ------------------------------------------------------------------ # # Actions # ------------------------------------------------------------------ # def _on_connect(self): """Bascule connexion/déconnexion. Le travail réel part dans un thread.""" if self._worker is not None: return # opération déjà en cours if self._is_connected or wg_core.is_connected(self._cfg): self._start_disconnect() else: self._start_connect(silent=False) # ── Connexion ──────────────────────────────────────────────────────── def _start_connect(self, silent: bool = False): if self._worker is not None: return if not self._cfg.configured: if not silent: QMessageBox.information( self, "Configuration manquante", "WireGuard n'est pas encore configuré.\n" "Ouvrez le panneau Administrateur." ) return self._begin_task("Connexion…") shutdown_guard.arm() # le tunnel devra être rendu à la sortie self._worker = TaskWorker(wg_core.connect, self._cfg, tag="connect" if not silent else "connect-silent", parent=self) self._worker.done.connect(self._on_task_done) self._worker.finished.connect(self._worker.deleteLater) self._worker.start() # ── Déconnexion ────────────────────────────────────────────────────── def _start_disconnect(self): if self._worker is not None: return self._begin_task("Déconnexion…") self._worker = TaskWorker(wg_core.disconnect, self._cfg, tag="disconnect", parent=self) self._worker.done.connect(self._on_task_done) self._worker.finished.connect(self._worker.deleteLater) self._worker.start() def access_code_needed(self) -> bool: """Vrai quand une saisie de code aurait un sens ici et maintenant. Deux conditions seulement : un tunnel monté — l'API du serveur n'est joignable que par l'intérieur — et aucune autorisation en cours. « En cours » et non « obtenue » : une autorisation a un terme, fixé par le serveur. Passé celui-ci le pair retourne en quarantaine sans que rien ne bouge côté poste, et le bouton doit reparaître — sans quoi l'utilisateur retrouve un réseau muet et un bouton masqué, exactement la situation que ce bouton existe pour éviter. L'identifiant VPN n'en fait volontairement pas partie. Il en faisait partie, et c'était un piège : un poste bel et bien tenu en quarantaine par le serveur, mais dont ce champ n'avait jamais été renseigné, n'affichait aucun moyen de saisir son code — précisément dans la situation où il en avait besoin. La fenêtre de saisie demande l'identifiant quand il manque. """ return self._is_connected and not self._access_valid() def request_access_code(self): """Ouvre la saisie du code, à la demande de l'utilisateur. Distincte de l'enchaînement automatique après connexion : le tunnel monte en quarantaine et l'autorisation expire d'elle-même, il faut donc pouvoir revenir saisir un code à tout moment — après un refus, après une reconnexion automatique, ou une fois l'autorisation échue. """ if not self._is_connected: QMessageBox.information( self, "Tunnel inactif", "Le serveur d'authentification n'est joignable que depuis " "l'intérieur du tunnel.\nConnectez-vous d'abord.") return self._open_access_dialog() def _authenticate_with_server(self, silent: bool): """Invite spontanée à saisir le code, juste après une connexion. Sans identifiant VPN configuré, l'invite est tue : le serveur peut très bien tourner en mode permissif, où un compte non enrôlé conserve son accès, et une installation sans serveur d'authentification n'a aucune raison de voir surgir cette fenêtre à chaque connexion. Cela ne vaut que pour l'invite : la saisie reste atteignable à la demande, par le bouton et par le systray, quelle que soit la configuration. """ # L'invite ne part qu'après la réponse du serveur : le tunnel qui # remonte ne révoque pas l'autorisation en cours, et réclamer un code # sans avoir demandé revenait à en redemander un déjà validé. # # Une reconnexion automatique ne fait jamais surgir de fenêtre modale # — personne n'est devant l'écran, l'application resterait figée ; le # bouton prend le relais. # # Le cas « pas de serveur d'authentification » n'est plus tranché ici : # il l'était d'après la présence d'un identifiant VPN en configuration, # ce que le poste ne pouvait pas savoir. C'est `_on_probe_reply` qui # renonce à l'invite quand le serveur répond ne connaître aucun pair à # cette adresse — avec sa réponse en main, plutôt qu'avant de l'avoir. self._probe_access(prompt_if_closed=not silent) def _probe_access(self, prompt_if_closed: bool): """Interroge le serveur sur l'état de l'accès, sans bloquer l'écran. Un seul sondage à la fois : les deux déclencheurs — front montant du tunnel et fin d'une connexion — se produisent souvent coup sur coup. L'intention d'inviter à saisir un code survit à cette fusion, sans quoi une connexion manuelle pourrait n'ouvrir aucune fenêtre parce qu'un sondage silencieux serait parti une fraction de seconde plus tôt. """ self._probe_prompt = self._probe_prompt or prompt_if_closed if self._probe_worker is not None: return self._probe_worker = ValueWorker(vpn_session.session_status, self._cfg, parent=self) self._probe_worker.done.connect(self._on_probe_reply) self._probe_worker.finished.connect(self._probe_worker.deleteLater) self._probe_worker.start() def _retry_probe(self): """Second sondage, après un silence du serveur (cf. `_on_probe_reply`). Le différé peut arriver dans un monde qui a changé : tunnel retombé, application en train de se fermer. Dans les deux cas il n'y a plus rien à demander, et surtout plus personne à qui montrer une fenêtre. """ if self._quitting or not self._is_connected: self._probe_prompt = False return self._probe_access(prompt_if_closed=True) def _on_probe_reply(self, result): """Adopte l'état renvoyé par le serveur, ou réclame un code.""" self._probe_worker = None prompt = self._probe_prompt self._probe_prompt = False was_open = self._access_open self._access_probed_at = datetime.now(timezone.utc) # Seule une réponse du serveur renseigne l'enrôlement ; son silence # laisse le constat précédent en place plutôt que de le contredire. if result is not None and result.authoritative: self._access_enrolled = result.enrolled if result is not None and result.ok: self._access_open = True self._access_until = result.access_until self._access_denials = 0 self._probe_retried = False # Journalisé au changement seulement : la re-vérification repasse # ici chaque minute, et répéter la même ligne noierait le journal. if not was_open or result.message != self._access_message: self._add_log(result.message, "success") self._access_message = result.message self._refresh_status() return # Le serveur a-t-il seulement parlé ? Un worker en échec rend None, et # `authoritative` distingue le refus du silence. spoke = result is not None and result.authoritative if was_open: # Re-vérification d'un accès qu'on croyait ouvert. if spoke: # Le serveur affirme que l'accès ne l'est plus : inutile # d'attendre confirmation, il est seul juge. self._access_open = False self._access_until = None self._access_message = "" self._access_denials = 0 self._add_log(result.message or "Accès distant refermé par le " "serveur — code à ressaisir", "warning") else: # Silence : n'en rien conclure tout de suite, sinon l'état # clignoterait au moindre hoquet. Après plusieurs minutes sans # réponse, mieux vaut rendre le bouton que d'afficher un # « Ouvert ✓ » que plus rien n'étaye. self._access_denials += 1 if self._access_denials >= _ACCESS_UNKNOWN_BEFORE_CLOSE: self._access_open = False self._access_until = None self._access_message = "" self._add_log("Serveur d'authentification sans réponse — " "accès distant à reconfirmer", "warning") self._refresh_status() return # Accès jamais ouvert de cette session. Une invite était-elle attendue ? if prompt and not spoke and not self._probe_retried: # Le serveur n'a rien dit : faire surgir la fenêtre maintenant # ferait saisir un code peut-être inutile — l'accès est peut-être # déjà ouvert, et le tunnel vient de monter. Un second essai avant # de déranger l'utilisateur. self._probe_retried = True self._probe_prompt = True QTimer.singleShot(_PROBE_RETRY_MS, self._retry_probe) self._refresh_status() return self._probe_retried = False if prompt and self._access_enrolled is False: # Le serveur ne connaît aucun pair à cette adresse : lui réclamer # un code serait absurde, aucun ne l'ouvrira. Le bouton reste # atteignable — discret — car le poste peut être enrôlé plus tard # sans que rien ici ne l'apprenne. self._add_log("Ce poste n'est pas enrôlé sur le serveur " "d'authentification", "warning") self._refresh_status() return if prompt: self._open_access_dialog() return self._add_log("Code d'accès requis — bouton « Saisir le code " "d'accès »", "warning") self._refresh_status() def _open_access_dialog(self): """Ouvre la fenêtre de saisie et journalise la réponse du serveur. La fenêtre est lue puis détruite dans la foulée : parentée à celle-ci, elle survivait autrement à chaque saisie — minuteur compris — et `_access_dialog` écrasait la précédente, dont le thread n'était alors plus attendu à l'arrêt de l'application. """ dlg = AccessCodeDialog(self._cfg, self) self._access_dialog = dlg try: accepted = dlg.exec() == AccessCodeDialog.DialogCode.Accepted # Lus avant destruction : ils ne survivent pas à `dispose()`. message = dlg.message() until = dlg.access_until() finally: self._access_dialog = None dlg.dispose() if accepted: self._access_open = True self._access_until = until self._access_message = message # Le serveur vient de répondre : cela vaut sondage, et remet à # zéro le décompte des silences qui finit par refermer l'accès. self._access_probed_at = datetime.now(timezone.utc) self._access_denials = 0 self._probe_retried = False self._add_log(message, "success") self._announce_access_granted() else: # Abandon : l'état d'accès n'est pas touché. Il l'était, et à tort # — un sondage abouti pendant que la fenêtre était ouverte (les # minuteries tournent dans la boucle imbriquée d'`exec()`) pouvait # établir que l'accès était ouvert, que « Plus tard » remettait # aussitôt à fermé. Renoncer à saisir un code ne referme rien côté # serveur ; ça ne doit rien refermer ici non plus. # # La fenêtre affiche déjà le refus du serveur et laisse réessayer : # y superposer une boîte d'alerte n'apprendrait rien de plus. self._add_log(message or "Authentification abandonnée — " "accès distant fermé", "warning") self._refresh_status() def _announce_access_granted(self): """Confirme la validation, en disant jusqu'à quand l'accès tient. Le message du serveur partait jusqu'ici dans le seul journal, pendant que la fenêtre se refermait : rien ne distinguait à l'écran un code accepté d'un code ignoré, et la durée obtenue restait invisible alors que c'est précisément ce que l'utilisateur vient de demander. """ if self._access_until is not None: fin = self._access_until.astimezone() texte = (f"Code accepté.\n\nAccès au réseau distant ouvert " f"jusqu'au {fin:%d/%m/%Y à %H:%M} " f"(soit {self._remaining_text()}).") else: texte = ("Code accepté.\n\nAccès au réseau distant ouvert. Le " "serveur n'a pas annoncé d'échéance.") QMessageBox.information(self, "Accès ouvert", texte) # Le serveur peut avoir enregistré l'autorisation sans que le trafic # passe pour autant — règle de pare-feu non appliquée, par exemple. # Le constater vaut mieux que laisser l'utilisateur découvrir seul un # réseau resté muet après un code pourtant accepté. self._verify_access_effective() def _verify_access_effective(self): """Vérifie en fond que le trafic atteint réellement le réseau distant.""" if self._reach_worker is not None: return self._reach_worker = ValueWorker(vpn_session.remote_network_reachable, self._cfg, parent=self) self._reach_worker.done.connect(self._on_reach_reply) self._reach_worker.finished.connect(self._reach_worker.deleteLater) self._reach_worker.start() def _on_reach_reply(self, joignable: bool): self._reach_worker = None cible = (self._cfg.wg.get("split_dns_server") or "").strip() if not cible: return # rien à sonder, donc rien à conclure if joignable: self._add_log(f"Trafic vers le réseau distant confirmé ({cible})", "success") return self._add_log( f"Code accepté, mais {cible} reste injoignable : le serveur a " "enregistré l'autorisation sans que le trafic passe. Prévenez " "l'administrateur.", "warning") def _begin_task(self, label: str): self._connecting = True self._btn_connect.setEnabled(False) self._btn_connect.setText(label) def _on_task_done(self, ok: bool, msg: str, tag: str): self._worker = None self._connecting = False self._btn_connect.setEnabled(True) silent = tag.endswith("-silent") if tag.startswith("connect"): if ok: self._reconnect_failures = 0 self._connected_since = QDateTime.currentDateTime() endpoint = self._cfg.wg.get("server_endpoint", "") self._session_id = hist.start_session(endpoint, self._cfg.active_profile) self._add_log(f"Connexion établie → {endpoint}", "success") # `connect()` ajoute une ligne d'avertissement au message de # succès quand le split-DNS n'a pas pu être appliqué — sans # quoi cet échec restait invisible. base, _, extra = msg.partition("\n") if extra: self._add_log(extra, "warning") # Le tunnel est monté mais ne mène encore nulle part : le # serveur garde le pair en quarantaine tant qu'un code n'a # pas été validé. C'est seulement maintenant que l'API est # joignable — elle ne l'est que depuis l'intérieur du tunnel. self._authenticate_with_server(silent) QTimer.singleShot(5000, self._update_ping) else: self._reconnect_failures += 1 self._add_log(f"Erreur : {msg}", "error") if not silent: QMessageBox.warning(self, "Erreur de connexion", msg) else: # disconnect if ok: if self._session_id is not None: hist.end_session(self._session_id, self._prev_rx or 0, self._prev_tx or 0) self._session_id = None self._add_log("Tunnel WireGuard désactivé", "warning") else: self._add_log(f"Erreur déconnexion : {msg}", "error") if not silent: QMessageBox.warning(self, "Erreur", msg) self._check_dns_health() self._warn_foreign_interfaces() self._refresh_status() def _warn_foreign_interfaces(self): """Signale un tunnel WireGuard monté hors de WGSecure. Il impose son propre DNS au système et WGSecure ne le démontera pas : c'est une cause fréquente de « plus d'accès internet » après coup. """ try: foreign = wg_core.foreign_wg_interfaces(self._cfg) except Exception: return for iface in foreign: if iface in self._warned_foreign: continue self._warned_foreign.add(iface) self._add_log( f"Tunnel « {iface} » actif hors de WGSecure " f"(interface configurée : {self._cfg.wg.get('interface_name','wgs0')})", "warning", ) def _check_dns_health(self): """Signale — et répare — une résolution DNS restée détournée.""" iface = self._cfg.wg.get("interface_name", "wgs0") repaired, msg = dns_util.repair_if_broken(iface, self._cfg.wg.get("dns", "")) if msg: self._add_log(msg, "warning" if repaired else "error") def _open_history(self): HistoryDialog(self).exec() def _open_admin(self): if self._cfg.has_admin_password(): pw, ok = QInputDialog.getText( self, "Accès Administrateur", "Mot de passe administrateur :", QLineEdit.EchoMode.Password, ) if not ok: return if not self._cfg.check_admin_password(pw): QMessageBox.warning(self, "Refusé", "Mot de passe incorrect.") return dlg = AdminWindow(self._cfg, self) dlg.exec() self._reload_profiles() self._refresh_status() # ------------------------------------------------------------------ # # Fermeture — restitution de l'état réseau # ------------------------------------------------------------------ # def prepare_quit(self) -> None: """Arrête minuteries et threads, puis rend le réseau à son état initial. Appelé par toutes les voies de sortie (croix de la fenêtre, « Quitter » du systray, signal système). Synchrone : la restauration doit être terminée avant que le processus ne disparaisse, sinon la configuration DNS du tunnel survit à l'application et la machine perd la résolution de noms. """ if self._quitting: return self._quitting = True for timer in (self._status_timer, self._clock_timer, self._bw_timer, self._ping_timer, self._reconnect_timer): timer.stop() # Laisser les threads en cours se terminer avant de démonter le tunnel : # une connexion et une déconnexion simultanées laisseraient un état mixte. for worker in (self._worker, self._ping_worker, self._probe_worker, self._reach_worker, self._update_worker): try: if worker is not None and worker.isRunning(): worker.wait(15000) except RuntimeError: pass # objet Qt déjà détruit if self._access_dialog is not None: # Une validation peut être partie puis la fenêtre fermée aussitôt : # son thread survit à la fermeture et ne doit pas être détruit avec # la fenêtre principale. self._access_dialog.wait_for_worker(15000) self._access_dialog = None self._worker = None self._ping_worker = None self._probe_worker = None self._reach_worker = None self._update_worker = None if self._session_id is not None: hist.end_session(self._session_id, self._prev_rx or 0, self._prev_tx or 0) self._session_id = None shutdown_guard.restore_network("fermeture de l'application") def closeEvent(self, event: QCloseEvent): if self._cfg.get("ui", "minimize_to_tray") and not self._quitting: # Simple masquage : ce n'est pas une fermeture, le tunnel reste actif. event.ignore() self.hide() return self.prepare_quit() event.accept()