tuxgyverandClaude Sonnet 5
6a3e8700c3
Ajoute un fichier log texte rotatif (wgsecure.log) dans le dossier applicatif, en complément du journal JSON interne, pour diagnostiquer un échec (ex. split-DNS) sans dépendre de l'UI. Corrige le panneau Journal qui coupait les messages longs sans moyen de les lire en entier (défilement horizontal réactivé + infobulle). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
835 lines
36 KiB
Python
835 lines
36 KiB
Python
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 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 _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")
|
|
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()
|