from __future__ import annotations from PyQt6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QFrame, QMessageBox, QInputDialog, QLineEdit, QListWidget, QListWidgetItem, QComboBox, ) from PyQt6.QtCore import Qt, QTimer, QDateTime, QEvent 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.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 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._access_open = 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) # 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.setStyleSheet( f"QPushButton {{ background: {theme.WARN_SOLID}; color: {theme.TEXT};" " font-size: 12px; font-weight: bold; border-radius: 6px; border: none; }" "QPushButton:hover { background: #d68910; }" ) self._btn_access.clicked.connect(self.request_access_code) self._btn_access.setVisible(False) s.addWidget(self._btn_access) # 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 _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_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. """ if not (self._cfg.wg.get("vpn_username") or "").strip(): return "Non géré par le serveur" if not self._is_connected: return "—" return "Ouvert ✓" if self._access_open else "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 = {} self._is_connected = connected if connected: self._reconnect_failures = 0 else: # Tunnel tombé : l'autorisation obtenue auprès du serveur ne vaut # que pour l'adresse du pair, elle ne survit pas à la déconnexion. self._access_open = False 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 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. Trois conditions : un identifiant VPN configuré (sans quoi le serveur ne gère pas ce poste), un tunnel monté (l'API n'est joignable que par l'intérieur) et aucune autorisation en cours. """ return (bool((self._cfg.wg.get("vpn_username") or "").strip()) and self._is_connected and not self._access_open) 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 if not (self._cfg.wg.get("vpn_username") or "").strip(): QMessageBox.information( self, "Aucun identifiant VPN", "Renseignez l'identifiant VPN connu du serveur dans le " "panneau Administrateur.") return self._authenticate_with_server(silent=False) 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. Le # bouton « Saisir le code d'accès » prend le relais. self._add_log("Code d'accès requis — bouton « Saisir le code " "d'accès »", "warning") return dlg = AccessCodeDialog(self._cfg, self) accepted = dlg.exec() == AccessCodeDialog.DialogCode.Accepted self._access_open = accepted if accepted: self._add_log(dlg.message(), "success") else: # 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(dlg.message() or "Authentification abandonnée — " "accès distant fermé", "warning") self._refresh_status() 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()