"""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, QTimer, Signal from PySide6.QtGui import QKeySequence, QShortcut from PySide6.QtWidgets import ( QFileDialog, QFrame, QHBoxLayout, QInputDialog, QLabel, QLineEdit, QMenu, QMessageBox, QSplitter, QToolButton, QVBoxLayout, QWidget, ) from app.ai.danger import check_dangerous from app.ai.redaction import redact from app.ai.service import KIND_DIAGNOSE, KIND_EXPLAIN, KIND_GENERATE, KIND_SOLUTION, AIRequestWorker from app.dialogs.ai_command_dialog import AICommandDialog from app.dialogs.ai_prompts import CANCEL, MASK, confirm_sensitive_content, ensure_ai_notice_acknowledged from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change from app.dialogs.snippet_dialog import SnippetManagerDialog from app.dialogs.tunnel_dialog import TunnelDialog from app.local_terminal import SHELL_LABELS, SHELL_POWERSHELL, LocalTerminalConnection from app.services.reconnect import compute_backoff_delay from app.services.ssh_target import detect_ssh_target from app.session_manager import AUTH_KEY, AUTH_PASSWORD, Session from app.sftp_panel import SFTPPanel from app.snippets import Snippet, SnippetStore from app.ssh_connection import SSHConnection from app.terminal_widget import TerminalWidget from app.tunnels import TunnelManager from app.ui.ai_panel import AIPanel # 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, snippet_store: SnippetStore | None = None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None, ): super().__init__(parent) self.snippet_store = snippet_store self.broadcast = broadcast self.ai_provider_factory = ai_provider_factory self.credential_store = credential_store self.ai_panel: AIPanel | None = None self._ai_command_worker: AIRequestWorker | None = None 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) # Clic droit = coller (comme un terminal Unix) : les snippets passent donc # par ce bouton plutot que par un menu contextuel, qui entrerait en conflit. self.snippets_btn = QToolButton() self.snippets_btn.setText("📋 Snippets") self.snippets_btn.setToolTip("Inserer ou executer une commande enregistree") self.snippets_btn.setPopupMode(QToolButton.InstantPopup) self.snippets_btn.clicked.connect(self._show_snippets_menu) self.bar.addWidget(self.snippets_btn) self.ai_btn = QToolButton() self.ai_btn.setText("🤖 Assistant IA") self.ai_btn.setToolTip("Analyser la sortie selectionnee ou generer une commande (optionnel)") self.ai_btn.setPopupMode(QToolButton.InstantPopup) self.ai_btn.clicked.connect(self._show_ai_menu) self.bar.addWidget(self.ai_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) # ---------- snippets ---------- def _show_snippets_menu(self) -> None: menu = self._build_snippets_menu() menu.exec(self.snippets_btn.mapToGlobal(self.snippets_btn.rect().bottomLeft())) def _build_snippets_menu(self) -> QMenu: menu = QMenu(self) if self.snippet_store is None or not self.snippet_store.snippets: hint = menu.addAction("(aucun snippet)") hint.setEnabled(False) else: for category, snippets in self.snippet_store.by_category().items(): cat_menu = menu.addMenu(category) for snippet in snippets: snippet_menu = cat_menu.addMenu(snippet.name) if snippet.description: snippet_menu.setToolTip(snippet.description) insert_action = snippet_menu.addAction("Inserer dans le terminal") insert_action.triggered.connect(lambda checked=False, s=snippet: self.insert_snippet(s)) exec_action = snippet_menu.addAction("Executer") exec_action.triggered.connect(lambda checked=False, s=snippet: self.execute_snippet(s)) if self.broadcast is not None: multi_action = snippet_menu.addAction("Executer avec Multi-Send") multi_action.triggered.connect( lambda checked=False, s=snippet: self.execute_snippet(s, multi=True) ) menu.addSeparator() manage_action = menu.addAction("Gerer les snippets...") manage_action.triggered.connect(self._manage_snippets) return menu def _manage_snippets(self) -> None: if self.snippet_store is None: return SnippetManagerDialog(self.snippet_store, parent=self).exec() def insert_snippet(self, snippet: Snippet) -> None: """N'envoie jamais Entree : le texte est ecrit, a l'utilisateur de valider.""" if self.connection: self.connection.send(snippet.command.encode("utf-8")) def execute_snippet(self, snippet: Snippet, multi: bool = False) -> None: data = snippet.command.encode("utf-8") + b"\r" if multi and self.broadcast is not None: self.broadcast.broadcast_now(data) elif self.connection: self.connection.send(data) # ---------- assistant IA (optionnel) ---------- def _show_ai_menu(self) -> None: menu = self._build_ai_menu() menu.exec(self.ai_btn.mapToGlobal(self.ai_btn.rect().bottomLeft())) def _ai_source_text(self) -> str: """Texte selectionne s'il y en a un (pour cibler precisement une partie de la sortie), sinon tout l'ecran actuellement visible - pour que "Expliquer cette sortie" porte par defaut sur ce que l'utilisateur a reellement sous les yeux, sans obliger a selectionner au prealable.""" selection = self.terminal.selected_text() return selection if selection else self.terminal.visible_text() def _build_ai_menu(self) -> QMenu: menu = QMenu(self) has_content = bool(self._ai_source_text().strip()) tooltip = ( "Utilise le texte selectionne, ou tout l'ecran visible si rien n'est selectionne." if has_content else "Rien n'est encore affiche dans ce terminal." ) explain_action = menu.addAction("Expliquer cette sortie") explain_action.setEnabled(has_content) explain_action.setToolTip(tooltip) explain_action.triggered.connect(self._ai_explain) diagnose_action = menu.addAction("Diagnostiquer cette erreur") diagnose_action.setEnabled(has_content) diagnose_action.setToolTip(tooltip) diagnose_action.triggered.connect(self._ai_diagnose) solution_action = menu.addAction("Proposer une solution") solution_action.setEnabled(has_content) solution_action.setToolTip(tooltip) solution_action.triggered.connect(self._ai_solution) ask_action = menu.addAction("Poser une question...") ask_action.setToolTip( "La selection (ou l'ecran visible a defaut) est jointe comme contexte a la question." ) ask_action.triggered.connect(self._ai_ask) menu.addSeparator() generate_action = menu.addAction("Generer une commande...") generate_action.triggered.connect(self._ai_generate_command) menu.addSeparator() panel_visible = self.ai_panel is not None and self.ai_panel.isVisible() toggle_action = menu.addAction("Masquer le panneau IA" if panel_visible else "Afficher le panneau IA") toggle_action.triggered.connect(self._toggle_ai_panel) menu.addSeparator() settings_action = menu.addAction("Parametres de l'assistant IA...") settings_action.triggered.connect(self._open_ai_settings) return menu def _ensure_ai_panel(self) -> AIPanel: if self.ai_panel is None: self.ai_panel = AIPanel(self.ai_provider_factory) self.splitter.addWidget(self.ai_panel) # cree masque : sans cela, Qt affiche automatiquement un widget # fraichement ajoute a un splitter deja visible, ce qui romprait # l'etat "ferme par defaut" attendu du panneau. self.ai_panel.hide() return self.ai_panel def _toggle_ai_panel(self) -> None: panel = self._ensure_ai_panel() panel.setVisible(not panel.isVisible()) def _open_ai_settings(self) -> None: from app.dialogs.ai_settings_dialog import AISettingsDialog AISettingsDialog(credential_store=self.credential_store, parent=self).exec() def _ai_explain(self) -> None: self._run_ai_selection_action(KIND_EXPLAIN, "Expliquer cette sortie") def _ai_diagnose(self) -> None: self._run_ai_selection_action(KIND_DIAGNOSE, "Diagnostiquer cette erreur") def _ai_solution(self) -> None: self._run_ai_selection_action(KIND_SOLUTION, "Proposer une solution") def _run_ai_selection_action(self, kind: str, label: str) -> None: text = self._ai_source_text() if not text.strip(): return panel = self._ensure_ai_panel() panel.run_request(kind, text, label=label) def _ai_ask(self) -> None: panel = self._ensure_ai_panel() panel.prompt_question(context=self._ai_source_text()) def _ai_generate_command(self) -> None: if self._ai_command_worker is not None and self._ai_command_worker.isRunning(): return provider = self.ai_provider_factory() if self.ai_provider_factory else None if provider is None: QMessageBox.information( self, "Assistant IA", "L'assistant IA n'est pas configure (menu Assistant IA > Parametres de l'assistant IA...).", ) return request, ok = QInputDialog.getMultiLineText( self, "Generer une commande", "Decrivez ce que vous voulez accomplir :", ) if not ok or not request.strip(): return request = request.strip() if not ensure_ai_notice_acknowledged(self): return decision = confirm_sensitive_content(self, request) if decision == CANCEL: return if decision == MASK: request = redact(request) self.ai_btn.setEnabled(False) self.ai_btn.setText("🤖 Generation...") self._ai_command_worker = AIRequestWorker(provider, KIND_GENERATE, request) self._ai_command_worker.finished_ok.connect(self._on_ai_command_ok) self._ai_command_worker.finished_error.connect(self._on_ai_command_error) self._ai_command_worker.start() def _reset_ai_generate_button(self) -> None: self.ai_btn.setEnabled(True) self.ai_btn.setText("🤖 Assistant IA") def _on_ai_command_ok(self, response: str) -> None: self._reset_ai_generate_button() dangerous = [m.label for m in check_dangerous(response)] dialog = AICommandDialog(response, dangerous_labels=dangerous, parent=self) if dialog.exec() and dialog.insert_requested and dialog.command: self.insert_snippet_text(dialog.command) def _on_ai_command_error(self, message: str) -> None: self._reset_ai_generate_button() QMessageBox.critical(self, "Assistant IA", message) def insert_snippet_text(self, text: str) -> None: """N'envoie jamais Entree : le texte est ecrit, a l'utilisateur de valider.""" if self.connection: self.connection.send(text.encode("utf-8")) # ---------- 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.ai_panel is not None: self.ai_panel.shutdown() if self._ai_command_worker is not None and self._ai_command_worker.isRunning(): self._ai_command_worker.wait(2000) 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 = "", snippet_store=None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None, ): super().__init__( f"○ Connexion a {session.username}@{session.host}...", snippet_store=snippet_store, broadcast=broadcast, ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=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, snippet_store=None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None, ): super().__init__( f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...", snippet_store=snippet_store, broadcast=broadcast, ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=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()