from __future__ import annotations from PyQt6.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QFrame, QMessageBox, QInputDialog, QLineEdit, QListWidget, QListWidgetItem, QComboBox, ) from PyQt6.QtCore import Qt, QTimer, QDateTime from PyQt6.QtGui import QFont, QCloseEvent, QColor from app.core.config import Config from app.core import vpn_session 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.ui.worker import TaskWorker, ValueWorker from app.ui.mfa_dialog import MFADialog 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 class MainWindow(QMainWindow): 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._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_status_action()) root.addWidget(self._build_details()) # ── Pied de page ───────────────────────────────────────────────── root.addWidget(self._build_footer()) # ── 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) # 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"), ("mfa", "🔐", "MFA"), ]: 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 _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 — il faut donc lever la contrainte précédente avant de laisser le layout recalculer sa hauteur naturelle, puis la reverrouiller. """ self.setMinimumHeight(0) self.setMaximumHeight(16777215) self.centralWidget().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 self._cfg.mfa_enabled: # Rouvrir une fenêtre MFA modale toutes les 30 s rendrait # l'application inutilisable : la reconnexion doit rester manuelle. return 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 _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", "—")) self._info_labels["mfa"].setText( "Activé ✓" if self._cfg.mfa_enabled else "Désactivé" ) # 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 = {} self._is_connected = connected if connected: self._reconnect_failures = 0 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 if self._cfg.mfa_enabled: if not self._cfg.mfa_secret: if not silent: QMessageBox.warning(self, "MFA non configuré", "MFA activé mais aucun secret défini.") return if silent: return # jamais de fenêtre MFA sur minuterie dlg = MFADialog(self._cfg.mfa_secret, self) if dlg.exec() != MFADialog.DialogCode.Accepted or not dlg.is_verified(): self._add_log("MFA annulé ou échoué", "warning") return self._add_log("Authentification MFA réussie ✓", "success") 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 _authenticate_with_server(self, silent: bool): """Demande le code à 6 chiffres et le fait valider par le serveur. Sans identifiant VPN configuré, l'étape est ignorée : le serveur peut très bien tourner en mode permissif, où un compte non enrôlé conserve son accès. Imposer la saisie dans ce cas bloquerait des installations qui fonctionnent parfaitement. """ if not (self._cfg.wg.get("vpn_username") or "").strip(): return if silent: # Reconnexion automatique : ouvrir une fenêtre modale sans # personne devant l'écran laisserait l'application figée. self._add_log("Code d'accès requis — reconnectez-vous manuellement", "warning") return code, accepted = QInputDialog.getText( self, "Code d'accès", "Le tunnel est monté, mais l'accès aux ressources distantes\n" "reste fermé tant que votre code n'est pas validé.\n\n" "Code à 6 chiffres :", ) if not accepted or not code.strip(): self._add_log("Authentification annulée — accès distant fermé", "warning") return ok, message = vpn_session.authenticate(self._cfg, code.strip()) self._add_log(message, "success" if ok else "error") if not ok: QMessageBox.warning(self, "Accès refusé", message) 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): try: if worker is not None and worker.isRunning(): worker.wait(15000) except RuntimeError: pass # objet Qt déjà détruit self._worker = None self._ping_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()