"""Widgets d'onglet : terminal SSH, terminal local, SFTP autonome. Chaque onglet encapsule une connexion (SSH ou locale) qui expose la meme interface de signaux (connected / data_received / disconnected / error / send / resize), ce qui permet a `BaseTerminalTab` de rester independante du type de connexion sous-jacent. """ from __future__ import annotations import re from datetime import datetime from pathlib import Path from PySide6.QtCore import Qt, Signal, QTimer from PySide6.QtGui import QKeySequence, QShortcut from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMessageBox, QSplitter, QFrame, QLineEdit, QFileDialog, ) from app.session_manager import Session, AUTH_KEY, AUTH_PASSWORD from app.ssh_connection import SSHConnection from app.local_terminal import LocalTerminalConnection, SHELL_POWERSHELL, SHELL_LABELS from app.terminal_widget import TerminalWidget from app.sftp_panel import SFTPPanel from app.dialogs.tunnel_dialog import TunnelDialog from app.tunnels import TunnelManager from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change from app.services.ssh_target import detect_ssh_target from app.services.reconnect import compute_backoff_delay # sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles ANSI_ESCAPE_RE = re.compile( r"\x1b\[[0-9;?]*[a-zA-Z]" # CSI : couleurs, deplacements du curseur r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC : titre de fenetre r"|\x1b[()][A-Z0-9]" # jeux de caracteres r"|\x1b[=>]" # mode clavier r"|[\x00-\x08\x0b\x0c\x0e-\x1f]" # autres caracteres de controle ) class BaseTerminalTab(QWidget): """Base commune a un onglet terminal : barre de statut + terminal, autour d'une connexion (SSH ou locale) qui expose la meme interface de signaux (connected / data_received / disconnected / error / send / resize).""" def __init__(self, initial_status: str, parent=None): super().__init__(parent) layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self.bar = QHBoxLayout() self.bar.setContentsMargins(10, 6, 10, 6) self.status_label = QLabel(initial_status) self.status_label.setObjectName("statusLabel") self.bar.addWidget(self.status_label, 1) # Apparait quand la connexion tombe : les serveurs de test redemarrent souvent, # inutile de fermer l'onglet et de tout ressaisir. self.reconnect_btn = QToolButton() self.reconnect_btn.setObjectName("primary") self.reconnect_btn.setText("⟳ Reconnecter") self.reconnect_btn.hide() self.reconnect_btn.clicked.connect(self.reconnect) self.bar.addWidget(self.reconnect_btn) self.log_btn = QToolButton() self.log_btn.setText("⏺ Journal") self.log_btn.setToolTip("Enregistrer la session dans un fichier") self.log_btn.setCheckable(True) self.log_btn.clicked.connect(self.toggle_logging) self.bar.addWidget(self.log_btn) bar_frame = QFrame() bar_frame.setObjectName("tabBar") bar_frame.setLayout(self.bar) layout.addWidget(bar_frame) self.splitter = QSplitter(Qt.Horizontal) self.terminal = TerminalWidget() self.splitter.addWidget(self.terminal) layout.addWidget(self.splitter, 1) # barre de recherche, masquee tant qu'on n'a pas fait Ctrl+F self.search_bar = QFrame() self.search_bar.setObjectName("tabBar") search_layout = QHBoxLayout(self.search_bar) search_layout.setContentsMargins(10, 4, 10, 4) self.search_input = QLineEdit() self.search_input.setPlaceholderText("Rechercher dans l'historique... (Entree = precedent, Echap = fermer)") self.search_status = QLabel("") self.search_status.setObjectName("statusLabel") close_search = QToolButton() close_search.setText("✕") search_layout.addWidget(self.search_input, 1) search_layout.addWidget(self.search_status) search_layout.addWidget(close_search) self.search_bar.hide() layout.addWidget(self.search_bar) self.terminal.search_requested.connect(self.open_search) self.search_input.returnPressed.connect(self._run_search) close_search.clicked.connect(self.close_search) QShortcut(QKeySequence(Qt.Key_Escape), self.search_input, activated=self.close_search) self.connection = None # assigne par les sous-classes self._log_file = None self._connection_generation = 0 self._closing = False # ---------- journalisation ---------- def toggle_logging(self) -> None: if self._log_file: self.stop_logging() return default = Path.home() / f"session-{datetime.now():%Y%m%d-%H%M%S}.log" path, _ = QFileDialog.getSaveFileName(self, "Journaliser la session", str(default), "Journal (*.log)") if not path: self.log_btn.setChecked(False) return try: self._log_file = open(path, "a", encoding="utf-8", errors="replace") except OSError as exc: QMessageBox.critical(self, "Erreur", f"Impossible d'ecrire le journal : {exc}") self.log_btn.setChecked(False) return self._log_file.write(f"\n===== Session du {datetime.now():%d/%m/%Y %H:%M:%S} =====\n") self.log_btn.setChecked(True) self.log_btn.setToolTip(f"Journalisation active : {path}") def stop_logging(self) -> None: if self._log_file: try: self._log_file.close() except OSError: pass self._log_file = None self.log_btn.setChecked(False) self.log_btn.setToolTip("Enregistrer la session dans un fichier") def _log_output(self, data: bytes) -> None: if not self._log_file: return # on retire les sequences d'echappement : un journal doit rester lisible text = ANSI_ESCAPE_RE.sub("", data.decode("utf-8", errors="replace")) try: self._log_file.write(text) self._log_file.flush() except OSError: self.stop_logging() def _wire_connection(self) -> None: # Un signal deconnexion/erreur/connexion emis par une connexion remplacee # entre-temps (ex. reconnexion en cours) ne doit pas ecraser l'etat de la # nouvelle : chaque connexion est associee a la generation en cours au # moment ou elle est cablee, et un signal d'une generation perimee est ignore. generation = self._connection_generation self.connection.connected.connect(lambda: self._on_current_generation(generation, self._on_connected)) self.connection.data_received.connect(self.terminal.feed) self.connection.data_received.connect(self._log_output) self.connection.disconnected.connect( lambda reason: self._on_current_generation(generation, self._on_disconnected, reason) ) self.connection.error.connect( lambda message: self._on_current_generation(generation, self._on_error, message) ) self.terminal.data_to_send.connect(self.connection.send) self.terminal.resized.connect(self.connection.resize) def _on_current_generation(self, generation: int, fn, *args) -> None: if generation == self._connection_generation: fn(*args) def _create_connection(self): """Fabrique la connexion. Redefinie par chaque sous-classe : c'est ce qui permet de la recreer a l'identique lors d'une reconnexion.""" raise NotImplementedError def start_connection(self) -> None: self._connection_generation += 1 self.connection = self._create_connection() self._wire_connection() self.connection.start() def reconnect(self) -> None: if self.connection: self.connection.close_connection() self.connection.wait(1500) self.terminal.reset_screen() self.reconnect_btn.hide() self.status_label.setText("○ Reconnexion...") self.start_connection() def _on_connected(self) -> None: self.reconnect_btn.hide() self.terminal.setFocus() def _on_disconnected(self, reason: str) -> None: self.status_label.setText(f"● Deconnecte ({reason})") self.reconnect_btn.show() def _on_error(self, message: str) -> None: self.status_label.setText(f"● Erreur : {message}") self.reconnect_btn.show() QMessageBox.critical(self, "Erreur", message) # ---------- recherche dans l'historique ---------- def open_search(self) -> None: self.search_bar.show() self.search_input.setFocus() self.search_input.selectAll() def close_search(self) -> None: self.search_bar.hide() self.search_status.clear() self.terminal.setFocus() def _run_search(self) -> None: needle = self.search_input.text() if not needle: return found = self.terminal.find_text(needle) self.search_status.setText("" if found else "aucun resultat") def close_connection(self) -> None: self._closing = True # ne jamais reconnecter automatiquement une fermeture voulue self.stop_logging() if self.connection: self.connection.close_connection() self.connection.wait(2000) class SessionTabWidget(BaseTerminalTab): """Un onglet = une connexion SSH avec son terminal et un panneau SFTP optionnel.""" def __init__( self, session: Session, secret: str, bastion_target=None, bastion_password: str = "", parent=None, ): super().__init__(f"○ Connexion a {session.username}@{session.host}...", parent) self.session = session self._bastion_target = bastion_target self._bastion_password = bastion_password self.sftp_panel: SFTPPanel | None = None self.sftp_toggle = QToolButton() self.sftp_toggle.setText("📁 Fichiers (SFTP)") self.sftp_toggle.setToolTip("Afficher / masquer le navigateur de fichiers SFTP") self.sftp_toggle.setCheckable(True) self.sftp_toggle.setChecked(True) self.sftp_toggle.setEnabled(False) self.bar.addWidget(self.sftp_toggle) self.tunnel_btn = QToolButton() self.tunnel_btn.setText("🔀 Tunnels") self.tunnel_btn.setToolTip("Redirections de ports SSH (-L / -R)") self.tunnel_btn.setEnabled(False) self.tunnel_btn.clicked.connect(self.open_tunnels) self.bar.addWidget(self.tunnel_btn) self.tunnel_manager: TunnelManager | None = None self._secret = secret self.sftp_toggle.toggled.connect(self._toggle_sftp) # Reconnexion automatique : un timer a declenchement unique pour la # reconnexion elle-meme, et un second pour rafraichir le compte a # rebours affiche a l'utilisateur pendant l'attente. self._reconnect_attempt = 0 self._reconnect_seconds_left = 0 self._reconnect_last_reason = "" self._reconnect_timer = QTimer(self) self._reconnect_timer.setSingleShot(True) self._reconnect_timer.timeout.connect(self._do_reconnect) self._reconnect_countdown_timer = QTimer(self) self._reconnect_countdown_timer.timeout.connect(self._tick_reconnect_countdown) self.start_connection() def _create_connection(self): session = self.session connection = SSHConnection( host=session.host, port=session.port, username=session.username, auth_type=session.auth_type, key_path=session.key_path, password=self._secret if session.auth_type == AUTH_PASSWORD else "", passphrase=self._secret if session.auth_type == AUTH_KEY else "", # comme MobaXterm : le navigateur de fichiers est la des l'ouverture, # pas cache derriere une action supplementaire open_sftp_on_connect=True, bastion_target=self._bastion_target, bastion_password=self._bastion_password, keepalive_enabled=session.keepalive_enabled, keepalive_interval=session.keepalive_interval, keepalive_count_max=session.keepalive_count_max, ) connection.sftp_ready.connect(self._on_sftp_ready) connection.password_prompt.connect(self._on_password_prompt) connection.host_key_prompt.connect(self._on_host_key_prompt) connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt) connection.passphrase_prompt.connect(self._on_passphrase_prompt) return connection def open_tunnels(self) -> None: if not self.tunnel_manager: QMessageBox.information(self, "Tunnels", "La session doit etre connectee.") return TunnelDialog(self.tunnel_manager, parent=self).exec() actifs = len(self.tunnel_manager.tunnels) self.tunnel_btn.setText(f"🔀 Tunnels ({actifs})" if actifs else "🔀 Tunnels") def _on_connected(self) -> None: self._reconnect_attempt = 0 # une connexion reussie remet le compteur a zero self.status_label.setText(f"● Connecte : {self.session.username}@{self.session.host}:{self.session.port}") transport = self.connection.client.get_transport() if self.connection.client else None if transport: self.tunnel_manager = TunnelManager(transport) self.tunnel_btn.setEnabled(True) # le bouton n'est active qu'une fois le canal SFTP reellement ouvert # (_on_sftp_ready) : certains serveurs n'exposent pas le sous-systeme SFTP super()._on_connected() def _on_disconnected(self, reason: str) -> None: super()._on_disconnected(reason) self.sftp_toggle.setEnabled(False) self._maybe_schedule_reconnect(reason) def reconnect(self) -> None: self._cancel_reconnect_timers() super().reconnect() # ---------- reconnexion automatique ---------- def _cancel_reconnect_timers(self) -> None: self._reconnect_timer.stop() self._reconnect_countdown_timer.stop() def _maybe_schedule_reconnect(self, reason: str) -> None: if self._closing or not self.session.reconnect_enabled: return max_attempts = self.session.reconnect_max_attempts self._reconnect_attempt += 1 if max_attempts and self._reconnect_attempt > max_attempts: self.status_label.setText( f"● Deconnecte ({reason}). Abandon apres {max_attempts} tentative(s) de reconnexion." ) self.reconnect_btn.show() return self._reconnect_last_reason = reason self._reconnect_seconds_left = compute_backoff_delay( self._reconnect_attempt, self.session.reconnect_initial_delay, ) self._update_reconnect_status() self._reconnect_countdown_timer.start(1000) self._reconnect_timer.start(self._reconnect_seconds_left * 1000) def _tick_reconnect_countdown(self) -> None: self._reconnect_seconds_left -= 1 if self._reconnect_seconds_left <= 0: self._reconnect_countdown_timer.stop() return self._update_reconnect_status() def _update_reconnect_status(self) -> None: max_attempts = self.session.reconnect_max_attempts total = f"/{max_attempts}" if max_attempts else "" self.status_label.setText( f"● Connexion perdue ({self._reconnect_last_reason}). " f"Reconnexion dans {self._reconnect_seconds_left} seconde(s)... " f"(Tentative {self._reconnect_attempt}{total})" ) def _do_reconnect(self) -> None: self._reconnect_countdown_timer.stop() if self._closing: return self.reconnect() def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None: accepted = confirm_host_key(self, hostname, key_type, fingerprint) self.connection.confirm_host_key(accepted) def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None: accepted = confirm_host_key_change(self, hostname, key_type, old_fp, new_fp) self.connection.confirm_host_key_change(accepted) def _on_passphrase_prompt(self, key_path: str) -> None: text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}") self.connection.provide_passphrase(text if ok else None) def _on_password_prompt(self, target: str) -> None: text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :") self.connection.provide_password(text if ok else None) def _on_sftp_ready(self, sftp) -> None: """Le canal SFTP a ete ouvert en arriere-plan : on affiche le navigateur de fichiers a cote du terminal, sans action de l'utilisateur.""" if self.sftp_panel is not None: return self.sftp_panel = SFTPPanel(sftp, transfer_sftp_factory=self._open_transfer_channel) self.splitter.insertWidget(0, self.sftp_panel) self.splitter.setSizes([460, 540]) self.sftp_toggle.setEnabled(True) self.sftp_toggle.setChecked(True) self.terminal.setFocus() def _open_transfer_channel(self): """Ouvre un second canal SFTP, dedie aux transferts, pour ne jamais faire attendre la navigation derriere un gros envoi/reception (voir SFTPPanel).""" import paramiko return paramiko.SFTPClient.from_transport(self.connection.client.get_transport()) def close_connection(self) -> None: self._cancel_reconnect_timers() if self.sftp_panel: self.sftp_panel.shutdown() if self.tunnel_manager: self.tunnel_manager.stop_all() self.tunnel_manager = None super().close_connection() def _toggle_sftp(self, checked: bool) -> None: if self.sftp_panel is not None: self.sftp_panel.setVisible(checked) class SFTPTabWidget(QWidget): """Onglet SFTP autonome (facon WinSCP) : double panneau de fichiers, sans terminal.""" def __init__( self, session: Session, secret: str, bastion_target=None, bastion_password: str = "", parent=None, ): super().__init__(parent) self.session = session self.sftp_panel: SFTPPanel | None = None layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) bar = QHBoxLayout() bar.setContentsMargins(10, 6, 10, 6) self.status_label = QLabel(f"○ Connexion SFTP a {session.username}@{session.host}...") self.status_label.setObjectName("statusLabel") bar.addWidget(self.status_label, 1) bar_frame = QFrame() bar_frame.setObjectName("tabBar") bar_frame.setLayout(bar) layout.addWidget(bar_frame) self.container = QVBoxLayout() holder = QWidget() holder.setLayout(self.container) layout.addWidget(holder, 1) self.connection = SSHConnection( host=session.host, port=session.port, username=session.username, auth_type=session.auth_type, key_path=session.key_path, password=secret if session.auth_type == AUTH_PASSWORD else "", passphrase=secret if session.auth_type == AUTH_KEY else "", open_shell=False, open_sftp_on_connect=True, bastion_target=bastion_target, bastion_password=bastion_password, keepalive_enabled=session.keepalive_enabled, keepalive_interval=session.keepalive_interval, keepalive_count_max=session.keepalive_count_max, ) self.connection.sftp_ready.connect(self._on_sftp_ready) self.connection.password_prompt.connect(self._on_password_prompt) self.connection.error.connect(self._on_error) self.connection.disconnected.connect(self._on_disconnected) self.connection.host_key_prompt.connect(self._on_host_key_prompt) self.connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt) self.connection.passphrase_prompt.connect(self._on_passphrase_prompt) self.connection.start() def _on_sftp_ready(self, sftp) -> None: self.sftp_panel = SFTPPanel(sftp, transfer_sftp_factory=self._open_transfer_channel) self.container.addWidget(self.sftp_panel) self.status_label.setText( f"● SFTP connecte : {self.session.username}@{self.session.host}:{self.session.port}" ) def _open_transfer_channel(self): import paramiko return paramiko.SFTPClient.from_transport(self.connection.client.get_transport()) def _on_disconnected(self, reason: str) -> None: self.status_label.setText(f"● Deconnecte ({reason})") def _on_error(self, message: str) -> None: self.status_label.setText(f"● Erreur : {message}") QMessageBox.critical(self, "Erreur SFTP", message) def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None: self.connection.confirm_host_key(confirm_host_key(self, hostname, key_type, fingerprint)) def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None: self.connection.confirm_host_key_change( confirm_host_key_change(self, hostname, key_type, old_fp, new_fp) ) def _on_passphrase_prompt(self, key_path: str) -> None: text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}") self.connection.provide_passphrase(text if ok else None) def _on_password_prompt(self, target: str) -> None: text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :") self.connection.provide_password(text if ok else None) def close_connection(self) -> None: if self.sftp_panel: self.sftp_panel.shutdown() self.connection.close_connection() self.connection.wait(2000) class LocalTerminalTabWidget(BaseTerminalTab): """Un onglet = un terminal Windows local (PowerShell / cmd), sans session enregistree.""" sftp_requested = Signal(object) # cible ssh detectee (ou None) def __init__(self, shell: str = SHELL_POWERSHELL, parent=None): super().__init__(f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...", parent) self.shell = shell # Meme dans un terminal local, on tape souvent "ssh user@ip" a la main : # ce bouton ouvre le SFTP sur cet hote, sans passer par un favori. self.sftp_btn = QToolButton() self.sftp_btn.setText("📁 SFTP sur cette connexion") self.sftp_btn.setToolTip( "Ouvre un gestionnaire de fichiers SFTP sur l'hote de la commande ssh tapee ici" ) self.sftp_btn.clicked.connect( lambda: self.sftp_requested.emit(detect_ssh_target(self.terminal.buffer_text())) ) self.bar.addWidget(self.sftp_btn) self.start_connection() def _create_connection(self): return LocalTerminalConnection(shell=self.shell) def _on_connected(self) -> None: self.status_label.setText(f"● Terminal local — {SHELL_LABELS.get(self.shell, self.shell)}") super()._on_connected()