From 47005b1e269aaaa48817e6c64f6f52978e8efd02 Mon Sep 17 00:00:00 2001 From: Florian Date: Sat, 5 Sep 2026 01:42:31 +0200 Subject: [PATCH] refactor: decouper main_window.py en controllers/services/ui main_window.py concentrait la fenetre principale, l'arbre des sessions, les 4 classes d'onglets, la detection de commandes ssh et le multi-send : 983 lignes et une seule classe qui pilotait toute l'application. - app/services/ssh_target.py : detection d'une cible ssh dans un terminal (pur, sans dependance Qt) - app/ui/tabs.py : BaseTerminalTab / SessionTabWidget / SFTPTabWidget / LocalTerminalTabWidget (deplaces tels quels) - app/controllers/session_controller.py : panneau lateral (arbre sessions + terminaux locaux) et CRUD des sessions - app/controllers/connection_controller.py : cycle de vie des onglets (ouverture, connexion rapide, renommage, fermeture) - app/controllers/broadcast_manager.py : bouton et logique du mode Multi-Send main_window.py ne fait plus que construire ces composants et les relier par signaux (983 -> ~200 lignes). Comportement inchange ; MainWindow.tabs et detect_ssh_target restent importables depuis app.main_window pour la compatibilite avec les tests existants. Co-Authored-By: Claude Sonnet 5 --- app/controllers/__init__.py | 0 app/controllers/broadcast_manager.py | 47 ++ app/controllers/connection_controller.py | 183 +++++ app/controllers/session_controller.py | 232 ++++++ app/main_window.py | 892 ++--------------------- app/services/__init__.py | 0 app/services/ssh_target.py | 55 ++ app/ui/__init__.py | 0 app/ui/tabs.py | 450 ++++++++++++ 9 files changed, 1022 insertions(+), 837 deletions(-) create mode 100644 app/controllers/__init__.py create mode 100644 app/controllers/broadcast_manager.py create mode 100644 app/controllers/connection_controller.py create mode 100644 app/controllers/session_controller.py create mode 100644 app/services/__init__.py create mode 100644 app/services/ssh_target.py create mode 100644 app/ui/__init__.py create mode 100644 app/ui/tabs.py diff --git a/app/controllers/__init__.py b/app/controllers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/controllers/broadcast_manager.py b/app/controllers/broadcast_manager.py new file mode 100644 index 0000000..9173f52 --- /dev/null +++ b/app/controllers/broadcast_manager.py @@ -0,0 +1,47 @@ +"""Envoi simultane (ยซ Multi-Send ") : une frappe faite dans un onglet terminal part +aussi vers tous les autres onglets terminal ouverts, quand le mode est actif.""" +from __future__ import annotations + +from typing import Callable, Iterable + +from PySide6.QtCore import QObject +from PySide6.QtWidgets import QToolButton, QWidget + + +class BroadcastManager(QObject): + """Ne connait pas le QTabWidget : `tabs_provider` lui fournit les onglets courants, + ce qui la garde independante de ConnectionController.""" + + def __init__(self, tabs_provider: Callable[[], Iterable[QWidget]], parent=None): + super().__init__(parent) + self._tabs_provider = tabs_provider + + self.button = QToolButton() + self.button.setText("โ‡‰ Multi") + self.button.setCheckable(True) + self.button.setToolTip( + "Envoi simultane : ce que vous tapez part vers tous les onglets terminal ouverts" + ) + + @property + def enabled(self) -> bool: + return self.button.isChecked() + + def register(self, tab) -> None: + """A appeler pour chaque nouvel onglet terminal (SSH ou local) susceptible + d'emettre et de recevoir des donnees diffusees.""" + tab.terminal.data_to_send.connect(lambda data, source=tab: self._broadcast(data, source)) + + def _broadcast(self, data: bytes, source) -> None: + if not self.enabled: + return + # import tardif : evite un cycle (app.ui.tabs importe des dialogues, pas ce module) + from app.ui.tabs import BaseTerminalTab + + for tab in self._tabs_provider(): + if tab is source or not isinstance(tab, BaseTerminalTab): + continue + if tab.connection: + # on ecrit directement dans la connexion : passer par le terminal + # relancerait la diffusion et boucherait a l'infini + tab.connection.send(data) diff --git a/app/controllers/connection_controller.py b/app/controllers/connection_controller.py new file mode 100644 index 0000000..7e8f528 --- /dev/null +++ b/app/controllers/connection_controller.py @@ -0,0 +1,183 @@ +"""Cycle de vie des onglets (terminal SSH, terminal local, SFTP autonome) : +ouverture, connexion rapide, renommage, fermeture.""" +from __future__ import annotations + +from PySide6.QtCore import Qt, Signal, QObject +from PySide6.QtWidgets import QTabWidget, QWidget, QMenu, QInputDialog + +from app.session_manager import SessionStore, Session, AUTH_PASSWORD +from app.key_manager import KeyManager +from app.local_terminal import DEFAULT_SHELL, SHELL_LABELS +from app.dialogs.session_dialog import SessionDialog +from app.dialogs.connect_prompt import ask_secret +from app.services.ssh_target import detect_ssh_target +from app.controllers.broadcast_manager import BroadcastManager +from app.ui.tabs import BaseTerminalTab, SessionTabWidget, SFTPTabWidget, LocalTerminalTabWidget + + +class ConnectionController(QObject): + """Possede le QTabWidget central. `session_saved` est emis quand une connexion + rapide est aussi enregistree comme session, pour que le panneau lateral se + rafraichisse sans que ce controleur ait besoin de le connaitre.""" + + session_saved = Signal() + + def __init__( + self, + session_store: SessionStore, + key_manager: KeyManager, + broadcast: BroadcastManager, + parent_widget: QWidget | None = None, + ): + super().__init__(parent_widget) + self.session_store = session_store + self.key_manager = key_manager + self.broadcast = broadcast + self._parent_widget = parent_widget + + self.tabs = QTabWidget(parent_widget) + self.tabs.setTabsClosable(True) + self.tabs.setMovable(True) + self.tabs.tabBarDoubleClicked.connect(self.rename_tab) + self.tabs.tabBar().setContextMenuPolicy(Qt.CustomContextMenu) + self.tabs.tabBar().customContextMenuRequested.connect(self._on_tab_context_menu) + self.tabs.tabCloseRequested.connect(self._on_tab_close) + + # ---------- ouverture ---------- + def open_session(self, session: Session) -> None: + secret = "" + if session.auth_type == AUTH_PASSWORD: + text, ok = ask_secret( + self._parent_widget, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :" + ) + if not ok: + return + secret = text + + tab = SessionTabWidget(session, secret) + self.broadcast.register(tab) + index = self.tabs.addTab(tab, f"๐Ÿ–ฅ {session.name}") + self.tabs.setCurrentIndex(index) + + def open_sftp_session(self, session: Session) -> None: + """Ouvre un onglet SFTP autonome (double panneau) pour cette session.""" + secret = "" + if session.auth_type == AUTH_PASSWORD: + text, ok = ask_secret( + self._parent_widget, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :" + ) + if not ok: + return + secret = text + + tab = SFTPTabWidget(session, secret) + index = self.tabs.addTab(tab, f"๐Ÿ“ {session.name}") + self.tabs.setCurrentIndex(index) + + def open_local_terminal(self, shell: str = DEFAULT_SHELL) -> None: + tab = LocalTerminalTabWidget(shell) + tab.sftp_requested.connect(self.quick_sftp) + self.broadcast.register(tab) + label = SHELL_LABELS.get(shell, shell) + index = self.tabs.addTab(tab, f"๐Ÿ’ป {label}") + self.tabs.setCurrentIndex(index) + + def quick_sftp(self, target: dict | None = None) -> None: + """Ouvre un SFTP en saisissant directement les infos de connexion, sans favori. + + Si l'onglet courant est un terminal ou une commande ssh a ete tapee, l'hote, + l'utilisateur et le port sont pre-remplis automatiquement.""" + if target is None: + target = self._current_tab_ssh_target() + self._quick_connect(sftp=True, target=target) + + def quick_ssh(self) -> None: + self._quick_connect(sftp=False, target=self._current_tab_ssh_target()) + + def _current_tab_ssh_target(self) -> dict | None: + widget = self.tabs.currentWidget() + if isinstance(widget, SessionTabWidget): + return { + "user": widget.session.username, + "host": widget.session.host, + "port": widget.session.port, + } + if isinstance(widget, BaseTerminalTab): + return detect_ssh_target(widget.terminal.buffer_text()) + return None + + def _quick_connect(self, sftp: bool, target: dict | None = None) -> None: + prefill = None + if target: + prefill = Session( + name=target.get("host", ""), + host=target.get("host", ""), + username=target.get("user", ""), + port=target.get("port", 22), + ) + dialog = SessionDialog( + self.key_manager, + session=prefill, + parent=self._parent_widget, + connect_label="Se connecter", + title="Connexion SFTP" if sftp else "Connexion SSH", + ) + if not dialog.exec(): + return + session = dialog.result_session() + if dialog.should_save(): + self.session_store.add(session) + self.session_saved.emit() + if sftp: + self.open_sftp_session(session) + else: + self.open_session(session) + + # ---------- gestion des onglets ---------- + def rename_tab(self, index: int) -> None: + if index < 0: + return + actuel = self.tabs.tabText(index) + nouveau, ok = QInputDialog.getText( + self._parent_widget, "Renommer l'onglet", "Nom de l'onglet :", text=actuel + ) + if ok and nouveau.strip(): + self.tabs.setTabText(index, nouveau.strip()) + + def _on_tab_context_menu(self, pos) -> None: + bar = self.tabs.tabBar() + index = bar.tabAt(pos) + if index < 0: + return + menu = QMenu(self._parent_widget) + act_rename = menu.addAction("Renommer... (double-clic)") + menu.addSeparator() + act_close = menu.addAction("Fermer cet onglet") + act_close_others = menu.addAction("Fermer les autres onglets") + act_close_others.setEnabled(self.tabs.count() > 1) + + action = menu.exec(bar.mapToGlobal(pos)) + if action == act_rename: + self.rename_tab(index) + elif action == act_close: + self._on_tab_close(index) + elif action == act_close_others: + # on ferme par la fin : retirer un onglet decale les suivants + cible = self.tabs.widget(index) + for other in range(self.tabs.count() - 1, -1, -1): + if self.tabs.widget(other) is not cible: + self._on_tab_close(other) + + def _on_tab_close(self, index: int) -> None: + widget = self.tabs.widget(index) + if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): + widget.close_connection() + self.tabs.removeTab(index) + + def close_all(self) -> None: + """A appeler a la fermeture de l'application : ferme proprement toutes les + connexions (SSH, locales, SFTP) avant que les widgets ne soient detruits.""" + for i in range(self.tabs.count()): + widget = self.tabs.widget(i) + if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): + widget.close_connection() diff --git a/app/controllers/session_controller.py b/app/controllers/session_controller.py new file mode 100644 index 0000000..7736530 --- /dev/null +++ b/app/controllers/session_controller.py @@ -0,0 +1,232 @@ +"""Panneau lateral : arbre des sessions SSH enregistrees + terminaux locaux, et +leur CRUD (creer / modifier / dupliquer / supprimer). N'ouvre aucune connexion +elle-meme : elle emet des signaux que ConnectionController ecoute.""" +from __future__ import annotations + +import copy +import uuid + +from PySide6.QtCore import Qt, Signal, QObject +from PySide6.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QTreeWidget, QTreeWidgetItem, + QMenu, QMessageBox, +) + +from app.session_manager import SessionStore, Session +from app.key_manager import KeyManager +from app.local_terminal import SHELL_LABELS +from app.dialogs.session_dialog import SessionDialog + +SESSION_ROLE = Qt.UserRole + 10 +LOCAL_SHELL_ROLE = Qt.UserRole + 11 + + +class SessionController(QObject): + open_session_requested = Signal(object) # Session + open_sftp_requested = Signal(object) # Session + open_local_shell_requested = Signal(str) # identifiant de shell + quick_sftp_requested = Signal() + manage_keys_requested = Signal() + manage_known_hosts_requested = Signal() + + def __init__( + self, + session_store: SessionStore, + key_manager: KeyManager, + broadcast_button: QToolButton, + parent_widget: QWidget | None = None, + ): + super().__init__(parent_widget) + self.session_store = session_store + self.key_manager = key_manager + self._parent_widget = parent_widget + + self.widget = QWidget(parent_widget) + layout = QVBoxLayout(self.widget) + layout.setContentsMargins(6, 6, 6, 6) + layout.setSpacing(6) + + toolbar = QHBoxLayout() + new_btn = QToolButton() + new_btn.setObjectName("primary") + new_btn.setText("+ Session") + new_btn.clicked.connect(self.new_session) + toolbar.addWidget(new_btn) + + term_btn = QToolButton() + term_btn.setText("๐Ÿ’ป Terminal") + term_btn.setPopupMode(QToolButton.InstantPopup) + term_menu = QMenu(term_btn) + for shell_id, label in SHELL_LABELS.items(): + term_menu.addAction(label, lambda s=shell_id: self.open_local_shell_requested.emit(s)) + term_btn.setMenu(term_menu) + toolbar.addWidget(term_btn) + + toolbar.addWidget(broadcast_button) + + sftp_btn = QToolButton() + sftp_btn.setText("๐Ÿ“ SFTP") + sftp_btn.setToolTip("Ouvrir un gestionnaire de fichiers SFTP (sans session enregistree)") + sftp_btn.clicked.connect(self.quick_sftp_requested.emit) + toolbar.addWidget(sftp_btn) + + toolbar.addStretch() + + manage_btn = QToolButton() + manage_btn.setText("โš™") + manage_btn.setToolTip("Cles SSH et hotes connus") + manage_btn.setPopupMode(QToolButton.InstantPopup) + manage_menu = QMenu(manage_btn) + manage_menu.addAction("Gestionnaire de cles SSH...", self.manage_keys_requested.emit) + manage_menu.addAction("Gestionnaire des hotes connus...", self.manage_known_hosts_requested.emit) + manage_btn.setMenu(manage_menu) + toolbar.addWidget(manage_btn) + + layout.addLayout(toolbar) + + self.tree = QTreeWidget() + self.tree.setHeaderHidden(True) + self.tree.itemDoubleClicked.connect(self._on_tree_double_click) + self.tree.setContextMenuPolicy(Qt.CustomContextMenu) + self.tree.customContextMenuRequested.connect(self._on_tree_context_menu) + layout.addWidget(self.tree) + + # Sans minimum explicite, la largeur plancher est imposee par le contenu + # (les noms de sessions), ce qui empeche de retrecir vraiment le panneau. + self.widget.setMinimumWidth(120) + self.tree.setMinimumWidth(0) + + self.reload() + + # ---------- arbre ---------- + def reload(self) -> None: + self.tree.clear() + + locals_header = QTreeWidgetItem(["๐Ÿ’ป Terminaux locaux"]) + locals_header.setFlags(locals_header.flags() & ~Qt.ItemIsSelectable) + font = locals_header.font(0) + font.setBold(True) + locals_header.setFont(0, font) + self.tree.addTopLevelItem(locals_header) + for shell, label in SHELL_LABELS.items(): + child = QTreeWidgetItem([f" {label}"]) + child.setData(0, LOCAL_SHELL_ROLE, shell) + locals_header.addChild(child) + + sessions_header = QTreeWidgetItem(["๐Ÿ” Sessions SSH"]) + sessions_header.setFlags(sessions_header.flags() & ~Qt.ItemIsSelectable) + sessions_header.setFont(0, font) + self.tree.addTopLevelItem(sessions_header) + + folder_items: dict[str, QTreeWidgetItem] = {} + + def get_folder(path: str) -> QTreeWidgetItem: + if not path: + return sessions_header + if path in folder_items: + return folder_items[path] + parent_path, _, name = path.rpartition("/") + parent_item = get_folder(parent_path) + item = QTreeWidgetItem([f"๐Ÿ“ {name}"]) + item.setFlags(item.flags() & ~Qt.ItemIsSelectable) + parent_item.addChild(item) + folder_items[path] = item + return item + + for session in sorted(self.session_store.sessions, key=lambda s: s.name.lower()): + parent = get_folder(session.group) + item = QTreeWidgetItem([f"๐Ÿ–ฅ {session.name} ({session.host})"]) + item.setData(0, SESSION_ROLE, session.id) + parent.addChild(item) + + if not self.session_store.sessions: + hint = QTreeWidgetItem([" (aucune session โ€” bouton ยซ + Session ยป ou ยซ ๐Ÿ“ SFTP ยป)"]) + hint.setFlags(hint.flags() & ~Qt.ItemIsSelectable) + hint.setForeground(0, Qt.gray) + sessions_header.addChild(hint) + + self.tree.expandAll() + + def _session_from_item(self, item: QTreeWidgetItem) -> Session | None: + if not item: + return None + session_id = item.data(0, SESSION_ROLE) + if not session_id: + return None + return next((s for s in self.session_store.sessions if s.id == session_id), None) + + def _shell_from_item(self, item: QTreeWidgetItem) -> str | None: + if not item: + return None + return item.data(0, LOCAL_SHELL_ROLE) + + def _on_tree_double_click(self, item: QTreeWidgetItem, _col: int) -> None: + session = self._session_from_item(item) + if session: + self.open_session_requested.emit(session) + return + shell = self._shell_from_item(item) + if shell: + self.open_local_shell_requested.emit(shell) + + def _on_tree_context_menu(self, pos) -> None: + item = self.tree.itemAt(pos) + menu = QMenu(self.widget) + session = self._session_from_item(item) if item else None + shell = self._shell_from_item(item) if item else None + + if session: + connect_action = menu.addAction("๐Ÿ–ฅ Ouvrir un terminal SSH") + sftp_action = menu.addAction("๐Ÿ“ Ouvrir en SFTP (gestionnaire de fichiers)") + menu.addSeparator() + edit_action = menu.addAction("Modifier...") + duplicate_action = menu.addAction("Dupliquer") + delete_action = menu.addAction("Supprimer") + action = menu.exec(self.tree.viewport().mapToGlobal(pos)) + if action == connect_action: + self.open_session_requested.emit(session) + elif action == sftp_action: + self.open_sftp_requested.emit(session) + elif action == edit_action: + self.edit_session(session) + elif action == duplicate_action: + self.duplicate_session(session) + elif action == delete_action: + self.delete_session(session) + elif shell: + open_action = menu.addAction("Ouvrir") + action = menu.exec(self.tree.viewport().mapToGlobal(pos)) + if action == open_action: + self.open_local_shell_requested.emit(shell) + else: + new_action = menu.addAction("Nouvelle session...") + action = menu.exec(self.tree.viewport().mapToGlobal(pos)) + if action == new_action: + self.new_session() + + # ---------- CRUD ---------- + def new_session(self) -> None: + dialog = SessionDialog(self.key_manager, parent=self._parent_widget) + if dialog.exec(): + self.session_store.add(dialog.result_session()) + self.reload() + + def edit_session(self, session: Session) -> None: + dialog = SessionDialog(self.key_manager, session=session, parent=self._parent_widget) + if dialog.exec(): + self.session_store.update(dialog.result_session()) + self.reload() + + def duplicate_session(self, session: Session) -> None: + new_session = copy.deepcopy(session) + new_session.id = str(uuid.uuid4()) + new_session.name = f"{session.name} (copie)" + self.session_store.add(new_session) + self.reload() + + def delete_session(self, session: Session) -> None: + if QMessageBox.question( + self._parent_widget, "Supprimer", f"Supprimer la session '{session.name}' ?" + ) == QMessageBox.Yes: + self.session_store.delete(session.id) + self.reload() diff --git a/app/main_window.py b/app/main_window.py index ae0d0fa..6c36aa1 100644 --- a/app/main_window.py +++ b/app/main_window.py @@ -1,492 +1,30 @@ -"""Fenetre principale : arborescence des sessions, onglets terminal/SFTP.""" +"""Fenetre principale : assemble le panneau des sessions, les onglets, les menus. + +La logique metier vit dans des controleurs dedies (app.controllers) et les widgets +d'onglet dans app.ui.tabs ; cette classe se contente de les construire et de les +relier. +""" from __future__ import annotations -import copy -import re -import uuid -from datetime import datetime -from pathlib import Path - -from PySide6.QtCore import Qt, Signal -from PySide6.QtGui import QAction, QActionGroup, QKeySequence, QShortcut +from PySide6.QtCore import Qt +from PySide6.QtGui import QAction, QActionGroup, QKeySequence from PySide6.QtWidgets import ( - QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget, - QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter, - QStatusBar, QFrame, QApplication, QLineEdit, QFileDialog, QInputDialog, + QMainWindow, QDockWidget, QMessageBox, QStatusBar, QApplication, ) -from app.session_manager import SessionStore, Session, AUTH_KEY, AUTH_PASSWORD +from app.session_manager import SessionStore from app.key_manager import KeyManager -from app.ssh_connection import SSHConnection -from app.local_terminal import ( - LocalTerminalConnection, SHELL_POWERSHELL, SHELL_CMD, SHELL_LABELS, DEFAULT_SHELL, -) +from app.local_terminal import SHELL_LABELS, DEFAULT_SHELL from app.terminal_widget import TerminalWidget -from app.sftp_panel import SFTPPanel -from app.dialogs.session_dialog import SessionDialog from app.dialogs.key_dialog import KeyManagerDialog from app.dialogs.known_hosts_dialog import KnownHostsDialog -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.theme import THEME_LABELS, apply_theme, current_theme, save_theme +from app.controllers.broadcast_manager import BroadcastManager +from app.controllers.session_controller import SessionController +from app.controllers.connection_controller import ConnectionController -SESSION_ROLE = Qt.UserRole + 10 -LOCAL_SHELL_ROLE = Qt.UserRole + 11 - -# Repere une commande ssh tapee a la main dans un terminal, pour proposer le SFTP -# sur cet hote sans avoir a creer de favori. -SSH_COMMAND_RE = re.compile(r"\bssh\s+(?P[^\r\n|;&]+)") -# options ssh qui consomment la valeur suivante (a ne pas confondre avec l'hote) -SSH_VALUE_OPTIONS = { - "-p", "-i", "-l", "-o", "-F", "-J", "-b", "-c", "-D", "-L", "-R", "-W", "-e", "-m", "-Q", -} -HOST_RE = re.compile(r"^[A-Za-z0-9][\w.\-]*$") -# 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 -) - - -def _parse_ssh_args(args: str) -> dict | None: - tokens = args.split() - user, host, port = "", "", 22 - index = 0 - while index < len(tokens): - token = tokens[index] - if token.startswith("-"): - if token in SSH_VALUE_OPTIONS: - value = tokens[index + 1] if index + 1 < len(tokens) else "" - if token == "-p" and value.isdigit(): - port = int(value) - if token == "-l" and value: - user = value - index += 2 - continue - if token.startswith("-p") and token[2:].isdigit(): # forme collee : -p2222 - port = int(token[2:]) - index += 1 - continue - if not host: - candidate = token - if "@" in candidate: - user, _, candidate = candidate.partition("@") - if HOST_RE.match(candidate): - host = candidate - index += 1 - return {"user": user, "host": host, "port": port} if host else None - - -def detect_ssh_target(text: str) -> dict | None: - """Renvoie la derniere cible ssh trouvee dans le texte d'un terminal.""" - result = None - for match in SSH_COMMAND_RE.finditer(text): - parsed = _parse_ssh_args(match.group("args")) - if parsed: - result = parsed - return result - - -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 - - # ---------- 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: - self.connection.connected.connect(self._on_connected) - self.connection.data_received.connect(self.terminal.feed) - self.connection.data_received.connect(self._log_output) - self.connection.disconnected.connect(self._on_disconnected) - self.connection.error.connect(self._on_error) - self.terminal.data_to_send.connect(self.connection.send) - self.terminal.resized.connect(self.connection.resize) - - 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 = 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.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, parent=None): - super().__init__(f"โ—‹ Connexion a {session.username}@{session.host}...", parent) - self.session = session - 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) - 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, - ) - 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.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) - - 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) - 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 close_connection(self) -> None: - 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, 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, - ) - 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) - self.container.addWidget(self.sftp_panel) - self.status_label.setText( - f"โ— SFTP connecte : {self.session.username}@{self.session.host}:{self.session.port}" - ) - - 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: - 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() +# re-exporte pour compatibilite (utilise par tests/test_detect.py et du code existant) +from app.services.ssh_target import detect_ssh_target # noqa: F401 class MainWindow(QMainWindow): @@ -498,34 +36,61 @@ class MainWindow(QMainWindow): self.session_store = SessionStore() self.key_manager = KeyManager() + self.broadcast = BroadcastManager(self._iter_tabs, self) + self.connection_controller = ConnectionController( + self.session_store, self.key_manager, self.broadcast, parent_widget=self, + ) + self.session_controller = SessionController( + self.session_store, self.key_manager, self.broadcast.button, parent_widget=self, + ) + self._wire_controllers() + self._build_menu() self._build_sidebar() - self._build_tabs() + self.setCentralWidget(self.connection_controller.tabs) self.setStatusBar(QStatusBar()) - self.reload_sessions() - # demarre directement dans un terminal utilisable, sans avoir a en choisir un - self.open_local_terminal(DEFAULT_SHELL) + self.connection_controller.open_local_terminal(DEFAULT_SHELL) + + # ---------- acces retro-compatible (tests, code externe) ---------- + @property + def tabs(self): + return self.connection_controller.tabs + + def _iter_tabs(self): + tabs = self.connection_controller.tabs + return [tabs.widget(i) for i in range(tabs.count())] + + # ---------- cablage entre controleurs ---------- + def _wire_controllers(self) -> None: + sc, cc = self.session_controller, self.connection_controller + sc.open_session_requested.connect(cc.open_session) + sc.open_sftp_requested.connect(cc.open_sftp_session) + sc.open_local_shell_requested.connect(cc.open_local_terminal) + sc.quick_sftp_requested.connect(cc.quick_sftp) + sc.manage_keys_requested.connect(self.open_key_manager) + sc.manage_known_hosts_requested.connect(self.open_known_hosts_manager) + cc.session_saved.connect(sc.reload) # ---------- construction UI ---------- def _build_menu(self) -> None: file_menu = self.menuBar().addMenu("&Fichier") new_session_action = QAction("Nouvelle session...", self) new_session_action.setShortcut(QKeySequence("Ctrl+N")) - new_session_action.triggered.connect(self.new_session) + new_session_action.triggered.connect(self.session_controller.new_session) file_menu.addAction(new_session_action) file_menu.addSeparator() quick_ssh_action = QAction("Connexion SSH rapide...", self) quick_ssh_action.setShortcut(QKeySequence("Ctrl+Shift+S")) - quick_ssh_action.triggered.connect(self.quick_ssh) + quick_ssh_action.triggered.connect(self.connection_controller.quick_ssh) file_menu.addAction(quick_ssh_action) quick_sftp_action = QAction("Connexion SFTP (gestionnaire de fichiers)...", self) quick_sftp_action.setShortcut(QKeySequence("Ctrl+Shift+F")) - quick_sftp_action.triggered.connect(self.quick_sftp) + quick_sftp_action.triggered.connect(self.connection_controller.quick_sftp) file_menu.addAction(quick_sftp_action) file_menu.addSeparator() @@ -534,12 +99,12 @@ class MainWindow(QMainWindow): f"Nouveau terminal local ({SHELL_LABELS.get(DEFAULT_SHELL, DEFAULT_SHELL)})", self ) default_action.setShortcut(QKeySequence("Ctrl+Shift+T")) - default_action.triggered.connect(lambda: self.open_local_terminal(DEFAULT_SHELL)) + default_action.triggered.connect(lambda: self.connection_controller.open_local_terminal(DEFAULT_SHELL)) file_menu.addAction(default_action) others_menu = file_menu.addMenu("Autre terminal local") for shell_id, label in SHELL_LABELS.items(): - others_menu.addAction(label, lambda s=shell_id: self.open_local_terminal(s)) + others_menu.addAction(label, lambda s=shell_id: self.connection_controller.open_local_terminal(s)) file_menu.addSeparator() quit_action = QAction("Quitter", self) @@ -591,357 +156,13 @@ class MainWindow(QMainWindow): ) self.sessions_dock = dock - container = QWidget() - layout = QVBoxLayout(container) - layout.setContentsMargins(6, 6, 6, 6) - layout.setSpacing(6) - - toolbar = QHBoxLayout() - new_btn = QToolButton() - new_btn.setObjectName("primary") - new_btn.setText("+ Session") - new_btn.clicked.connect(self.new_session) - toolbar.addWidget(new_btn) - - term_btn = QToolButton() - term_btn.setText("๐Ÿ’ป Terminal") - term_btn.setPopupMode(QToolButton.InstantPopup) - term_menu = QMenu(term_btn) - for shell_id, label in SHELL_LABELS.items(): - term_menu.addAction(label, lambda s=shell_id: self.open_local_terminal(s)) - term_btn.setMenu(term_menu) - toolbar.addWidget(term_btn) - - self.broadcast_btn = QToolButton() - self.broadcast_btn.setText("โ‡‰ Multi") - self.broadcast_btn.setCheckable(True) - self.broadcast_btn.setToolTip( - "Envoi simultane : ce que vous tapez part vers tous les onglets terminal ouverts" - ) - toolbar.addWidget(self.broadcast_btn) - - sftp_btn = QToolButton() - sftp_btn.setText("๐Ÿ“ SFTP") - sftp_btn.setToolTip("Ouvrir un gestionnaire de fichiers SFTP (sans session enregistree)") - sftp_btn.clicked.connect(self.quick_sftp) - toolbar.addWidget(sftp_btn) - - toolbar.addStretch() - - manage_btn = QToolButton() - manage_btn.setText("โš™") - manage_btn.setToolTip("Cles SSH et hotes connus") - manage_btn.setPopupMode(QToolButton.InstantPopup) - manage_menu = QMenu(manage_btn) - manage_menu.addAction("Gestionnaire de cles SSH...", self.open_key_manager) - manage_menu.addAction("Gestionnaire des hotes connus...", self.open_known_hosts_manager) - manage_btn.setMenu(manage_menu) - toolbar.addWidget(manage_btn) - - layout.addLayout(toolbar) - - self.tree = QTreeWidget() - self.tree.setHeaderHidden(True) - self.tree.itemDoubleClicked.connect(self._on_tree_double_click) - self.tree.setContextMenuPolicy(Qt.CustomContextMenu) - self.tree.customContextMenuRequested.connect(self._on_tree_context_menu) - layout.addWidget(self.tree) - - # Sans minimum explicite, la largeur plancher est imposee par le contenu - # (les noms de sessions), ce qui empeche de retrecir vraiment le panneau. - container.setMinimumWidth(120) - self.tree.setMinimumWidth(0) - dock.setWidget(container) + dock.setWidget(self.session_controller.widget) self.addDockWidget(Qt.LeftDockWidgetArea, dock) self.resizeDocks([dock], [300], Qt.Horizontal) # garde la case du menu Affichage synchronisee si on ferme le panneau par sa croix dock.visibilityChanged.connect(self.toggle_sidebar_action.setChecked) - def _build_tabs(self) -> None: - self.tabs = QTabWidget() - self.tabs.setTabsClosable(True) - self.tabs.setMovable(True) - # double-clic sur un onglet pour le renommer, clic droit pour le menu - self.tabs.tabBarDoubleClicked.connect(self.rename_tab) - self.tabs.tabBar().setContextMenuPolicy(Qt.CustomContextMenu) - self.tabs.tabBar().customContextMenuRequested.connect(self._on_tab_context_menu) - self.tabs.tabCloseRequested.connect(self._on_tab_close) - self.setCentralWidget(self.tabs) - - # ---------- sessions / arbre ---------- - def reload_sessions(self) -> None: - self.tree.clear() - - locals_header = QTreeWidgetItem(["๐Ÿ’ป Terminaux locaux"]) - locals_header.setFlags(locals_header.flags() & ~Qt.ItemIsSelectable) - font = locals_header.font(0) - font.setBold(True) - locals_header.setFont(0, font) - self.tree.addTopLevelItem(locals_header) - for shell, label in SHELL_LABELS.items(): - child = QTreeWidgetItem([f" {label}"]) - child.setData(0, LOCAL_SHELL_ROLE, shell) - locals_header.addChild(child) - - sessions_header = QTreeWidgetItem(["๐Ÿ” Sessions SSH"]) - sessions_header.setFlags(sessions_header.flags() & ~Qt.ItemIsSelectable) - sessions_header.setFont(0, font) - self.tree.addTopLevelItem(sessions_header) - - folder_items: dict[str, QTreeWidgetItem] = {} - - def get_folder(path: str) -> QTreeWidgetItem: - if not path: - return sessions_header - if path in folder_items: - return folder_items[path] - parent_path, _, name = path.rpartition("/") - parent_item = get_folder(parent_path) - item = QTreeWidgetItem([f"๐Ÿ“ {name}"]) - item.setFlags(item.flags() & ~Qt.ItemIsSelectable) - parent_item.addChild(item) - folder_items[path] = item - return item - - for session in sorted(self.session_store.sessions, key=lambda s: s.name.lower()): - parent = get_folder(session.group) - item = QTreeWidgetItem([f"๐Ÿ–ฅ {session.name} ({session.host})"]) - item.setData(0, SESSION_ROLE, session.id) - parent.addChild(item) - - if not self.session_store.sessions: - hint = QTreeWidgetItem([" (aucune session โ€” bouton ยซ + Session ยป ou ยซ ๐Ÿ“ SFTP ยป)"]) - hint.setFlags(hint.flags() & ~Qt.ItemIsSelectable) - hint.setForeground(0, Qt.gray) - sessions_header.addChild(hint) - - self.tree.expandAll() - - def _session_from_item(self, item: QTreeWidgetItem) -> Session | None: - if not item: - return None - session_id = item.data(0, SESSION_ROLE) - if not session_id: - return None - return next((s for s in self.session_store.sessions if s.id == session_id), None) - - def _shell_from_item(self, item: QTreeWidgetItem) -> str | None: - if not item: - return None - return item.data(0, LOCAL_SHELL_ROLE) - - def _on_tree_double_click(self, item: QTreeWidgetItem, _col: int) -> None: - session = self._session_from_item(item) - if session: - self.open_session(session) - return - shell = self._shell_from_item(item) - if shell: - self.open_local_terminal(shell) - - def _on_tree_context_menu(self, pos) -> None: - item = self.tree.itemAt(pos) - menu = QMenu(self) - session = self._session_from_item(item) if item else None - shell = self._shell_from_item(item) if item else None - - if session: - connect_action = menu.addAction("๐Ÿ–ฅ Ouvrir un terminal SSH") - sftp_action = menu.addAction("๐Ÿ“ Ouvrir en SFTP (gestionnaire de fichiers)") - menu.addSeparator() - edit_action = menu.addAction("Modifier...") - duplicate_action = menu.addAction("Dupliquer") - delete_action = menu.addAction("Supprimer") - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - if action == connect_action: - self.open_session(session) - elif action == sftp_action: - self.open_sftp_session(session) - elif action == edit_action: - self.edit_session(session) - elif action == duplicate_action: - self.duplicate_session(session) - elif action == delete_action: - self.delete_session(session) - elif shell: - open_action = menu.addAction("Ouvrir") - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - if action == open_action: - self.open_local_terminal(shell) - else: - new_action = menu.addAction("Nouvelle session...") - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - if action == new_action: - self.new_session() - - def new_session(self) -> None: - dialog = SessionDialog(self.key_manager, parent=self) - if dialog.exec(): - self.session_store.add(dialog.result_session()) - self.reload_sessions() - - def edit_session(self, session: Session) -> None: - dialog = SessionDialog(self.key_manager, session=session, parent=self) - if dialog.exec(): - self.session_store.update(dialog.result_session()) - self.reload_sessions() - - def duplicate_session(self, session: Session) -> None: - new_session = copy.deepcopy(session) - new_session.id = str(uuid.uuid4()) - new_session.name = f"{session.name} (copie)" - self.session_store.add(new_session) - self.reload_sessions() - - def delete_session(self, session: Session) -> None: - if QMessageBox.question(self, "Supprimer", f"Supprimer la session '{session.name}' ?") == QMessageBox.Yes: - self.session_store.delete(session.id) - self.reload_sessions() - - # ---------- connexion / onglets ---------- - def open_session(self, session: Session) -> None: - secret = "" - if session.auth_type == AUTH_PASSWORD: - text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :") - if not ok: - return - secret = text - - tab = SessionTabWidget(session, secret) - self._register_broadcast(tab) - index = self.tabs.addTab(tab, f"๐Ÿ–ฅ {session.name}") - self.tabs.setCurrentIndex(index) - - def quick_sftp(self, target: dict | None = None) -> None: - """Ouvre un SFTP en saisissant directement les infos de connexion, sans favori. - - Si l'onglet courant est un terminal ou une commande ssh a ete tapee, l'hote, - l'utilisateur et le port sont pre-remplis automatiquement.""" - if target is None: - target = self._current_tab_ssh_target() - self._quick_connect(sftp=True, target=target) - - def quick_ssh(self) -> None: - self._quick_connect(sftp=False, target=self._current_tab_ssh_target()) - - def _current_tab_ssh_target(self) -> dict | None: - widget = self.tabs.currentWidget() - if isinstance(widget, SessionTabWidget): - return { - "user": widget.session.username, - "host": widget.session.host, - "port": widget.session.port, - } - if isinstance(widget, BaseTerminalTab): - return detect_ssh_target(widget.terminal.buffer_text()) - return None - - def _quick_connect(self, sftp: bool, target: dict | None = None) -> None: - prefill = None - if target: - prefill = Session( - name=target.get("host", ""), - host=target.get("host", ""), - username=target.get("user", ""), - port=target.get("port", 22), - ) - dialog = SessionDialog( - self.key_manager, - session=prefill, - parent=self, - connect_label="Se connecter", - title="Connexion SFTP" if sftp else "Connexion SSH", - ) - if not dialog.exec(): - return - session = dialog.result_session() - if dialog.should_save(): - self.session_store.add(session) - self.reload_sessions() - if sftp: - self.open_sftp_session(session) - else: - self.open_session(session) - - def open_sftp_session(self, session: Session) -> None: - """Ouvre un onglet SFTP autonome (double panneau) pour cette session.""" - secret = "" - if session.auth_type == AUTH_PASSWORD: - text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :") - if not ok: - return - secret = text - - tab = SFTPTabWidget(session, secret) - index = self.tabs.addTab(tab, f"๐Ÿ“ {session.name}") - self.tabs.setCurrentIndex(index) - - def open_local_terminal(self, shell: str = DEFAULT_SHELL) -> None: - tab = LocalTerminalTabWidget(shell) - tab.sftp_requested.connect(self.quick_sftp) - self._register_broadcast(tab) - label = SHELL_LABELS.get(shell, shell) - index = self.tabs.addTab(tab, f"๐Ÿ’ป {label}") - self.tabs.setCurrentIndex(index) - - def _broadcast(self, data: bytes, source) -> None: - """Envoie la meme frappe aux autres onglets terminal quand le mode Multi est actif.""" - if not self.broadcast_btn.isChecked(): - return - for index in range(self.tabs.count()): - tab = self.tabs.widget(index) - if tab is source or not isinstance(tab, BaseTerminalTab): - continue - if tab.connection: - # on ecrit directement dans la connexion : passer par le terminal - # relancerait la diffusion et boucherait a l'infini - tab.connection.send(data) - - def _register_broadcast(self, tab) -> None: - tab.terminal.data_to_send.connect(lambda data, t=tab: self._broadcast(data, t)) - - # ---------- renommage des onglets ---------- - def rename_tab(self, index: int) -> None: - if index < 0: - return - actuel = self.tabs.tabText(index) - nouveau, ok = QInputDialog.getText( - self, "Renommer l'onglet", "Nom de l'onglet :", text=actuel - ) - if ok and nouveau.strip(): - self.tabs.setTabText(index, nouveau.strip()) - - def _on_tab_context_menu(self, pos) -> None: - bar = self.tabs.tabBar() - index = bar.tabAt(pos) - if index < 0: - return - menu = QMenu(self) - act_rename = menu.addAction("Renommer... (double-clic)") - menu.addSeparator() - act_close = menu.addAction("Fermer cet onglet") - act_close_others = menu.addAction("Fermer les autres onglets") - act_close_others.setEnabled(self.tabs.count() > 1) - - action = menu.exec(bar.mapToGlobal(pos)) - if action == act_rename: - self.rename_tab(index) - elif action == act_close: - self._on_tab_close(index) - elif action == act_close_others: - # on ferme par la fin : retirer un onglet decale les suivants - cible = self.tabs.widget(index) - for other in range(self.tabs.count() - 1, -1, -1): - if self.tabs.widget(other) is not cible: - self._on_tab_close(other) - - def _on_tab_close(self, index: int) -> None: - widget = self.tabs.widget(index) - if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): - widget.close_connection() - self.tabs.removeTab(index) - - # ---------- cles / a propos ---------- + # ---------- cles / affichage / a propos ---------- def toggle_sidebar(self, visible: bool | None = None) -> None: """Affiche ou masque le panneau des sessions (F9).""" if visible is None: @@ -976,8 +197,5 @@ class MainWindow(QMainWindow): ) def closeEvent(self, event) -> None: - for i in range(self.tabs.count()): - widget = self.tabs.widget(i) - if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): - widget.close_connection() + self.connection_controller.close_all() super().closeEvent(event) diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/ssh_target.py b/app/services/ssh_target.py new file mode 100644 index 0000000..84229d8 --- /dev/null +++ b/app/services/ssh_target.py @@ -0,0 +1,55 @@ +"""Detection d'une cible ssh (user/host/port) tapee a la main dans un terminal. + +Sert a proposer le SFTP sur l'hote d'une commande `ssh ...` sans avoir a creer +un favori au prealable. +""" +from __future__ import annotations + +import re + +# Repere une commande ssh tapee a la main dans un terminal. +SSH_COMMAND_RE = re.compile(r"\bssh\s+(?P[^\r\n|;&]+)") +# options ssh qui consomment la valeur suivante (a ne pas confondre avec l'hote) +SSH_VALUE_OPTIONS = { + "-p", "-i", "-l", "-o", "-F", "-J", "-b", "-c", "-D", "-L", "-R", "-W", "-e", "-m", "-Q", +} +HOST_RE = re.compile(r"^[A-Za-z0-9][\w.\-]*$") + + +def _parse_ssh_args(args: str) -> dict | None: + tokens = args.split() + user, host, port = "", "", 22 + index = 0 + while index < len(tokens): + token = tokens[index] + if token.startswith("-"): + if token in SSH_VALUE_OPTIONS: + value = tokens[index + 1] if index + 1 < len(tokens) else "" + if token == "-p" and value.isdigit(): + port = int(value) + if token == "-l" and value: + user = value + index += 2 + continue + if token.startswith("-p") and token[2:].isdigit(): # forme collee : -p2222 + port = int(token[2:]) + index += 1 + continue + if not host: + candidate = token + if "@" in candidate: + user, _, candidate = candidate.partition("@") + if HOST_RE.match(candidate): + host = candidate + index += 1 + return {"user": user, "host": host, "port": port} if host else None + + +def detect_ssh_target(text: str) -> dict | None: + """Renvoie la derniere cible ssh trouvee dans le texte d'un terminal.""" + result = None + for match in SSH_COMMAND_RE.finditer(text): + parsed = _parse_ssh_args(match.group("args")) + if parsed: + result = parsed + return result diff --git a/app/ui/__init__.py b/app/ui/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/ui/tabs.py b/app/ui/tabs.py new file mode 100644 index 0000000..f7c0f30 --- /dev/null +++ b/app/ui/tabs.py @@ -0,0 +1,450 @@ +"""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 +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 + +# 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 + + # ---------- 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: + self.connection.connected.connect(self._on_connected) + self.connection.data_received.connect(self.terminal.feed) + self.connection.data_received.connect(self._log_output) + self.connection.disconnected.connect(self._on_disconnected) + self.connection.error.connect(self._on_error) + self.terminal.data_to_send.connect(self.connection.send) + self.terminal.resized.connect(self.connection.resize) + + 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 = 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.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, parent=None): + super().__init__(f"โ—‹ Connexion a {session.username}@{session.host}...", parent) + self.session = session + 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) + 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, + ) + 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.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) + + 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: + 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, 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, + ) + 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()