tuxgyverandClaude Opus 5
996bbe240d
La fenêtre de saisie ne s'ouvrait qu'une fois, juste après une connexion manuelle réussie. Un refus, une fenêtre fermée, une reconnexion automatique — qui n'ouvre volontairement aucune modale — ou l'expiration de l'autorisation au bout de 12 h laissaient un tunnel monté, un réseau muet et aucun élément d'interface pour ressaisir un code. Un bouton sous l'action de connexion et une entrée de systray apparaissent désormais dès qu'une saisie a un sens : identifiant VPN configuré, tunnel monté, accès non ouvert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
928 lines
40 KiB
Python
928 lines
40 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.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 — 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 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()
|