From 6de0aad79cf409019038ead1889fafd3117be921 Mon Sep 17 00:00:00 2001 From: snoopy Date: Fri, 4 Sep 2026 11:35:39 +0200 Subject: [PATCH] Edition de fichier distant en place, et journalisation de session Edition SFTP facon WinSCP : clic droit sur un fichier distant > "Editer". Le fichier est telecharge dans un temporaire, ouvert avec l'application associee du systeme, et renvoye automatiquement a chaque enregistrement. Beaucoup d'editeurs enregistrent en remplacant le fichier plutot qu'en le reecrivant : le chemin surveille disparait alors du QFileSystemWatcher, il est donc reinscrit apres chaque renvoi pour que les enregistrements suivants fonctionnent aussi. Journalisation : un bouton dans la barre de l'onglet enregistre toute la session dans un fichier. Les sequences d'echappement ANSI sont retirees pour que le journal reste lisible dans un editeur de texte. Co-Authored-By: Claude Opus 5 --- app/main_window.py | 1651 +++++++++++++++++++++++--------------------- app/sftp_panel.py | 63 +- 2 files changed, 917 insertions(+), 797 deletions(-) diff --git a/app/main_window.py b/app/main_window.py index 81e3ee4..3812597 100644 --- a/app/main_window.py +++ b/app/main_window.py @@ -1,795 +1,856 @@ -"""Fenetre principale : arborescence des sessions, onglets terminal/SFTP.""" -from __future__ import annotations - -import copy -import re -import uuid - -from PySide6.QtCore import Qt, Signal -from PySide6.QtGui import QAction, QActionGroup, QKeySequence, QShortcut -from PySide6.QtWidgets import ( - QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget, - QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter, - QStatusBar, QFrame, QApplication, QLineEdit, -) - -from app.session_manager import SessionStore, Session, AUTH_KEY, AUTH_PASSWORD -from app.key_manager import KeyManager -from app.ssh_connection import SSHConnection -from app.local_terminal import LocalTerminalConnection, SHELL_POWERSHELL, SHELL_CMD, SHELL_LABELS -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.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change -from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme - -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.\-]*$") - - -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) - 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 - - def _wire_connection(self) -> None: - self.connection.connected.connect(self._on_connected) - self.connection.data_received.connect(self.terminal.feed) - 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: - 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._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 _on_connected(self) -> None: - self.status_label.setText(f"● Connecte : {self.session.username}@{self.session.host}:{self.session.port}") - # 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 _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() - - -class MainWindow(QMainWindow): - def __init__(self): - super().__init__() - self.setWindowTitle("PySSH Client") - self.resize(1280, 820) - - self.session_store = SessionStore() - self.key_manager = KeyManager() - - self._build_menu() - self._build_sidebar() - self._build_tabs() - self.setStatusBar(QStatusBar()) - - self.reload_sessions() - - # demarre directement dans un terminal utilisable, sans avoir a en choisir un - self.open_local_terminal(SHELL_POWERSHELL) - - # ---------- 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) - 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) - 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) - file_menu.addAction(quick_sftp_action) - - file_menu.addSeparator() - - new_ps_action = QAction("Nouveau terminal local (PowerShell)", self) - new_ps_action.setShortcut(QKeySequence("Ctrl+Shift+T")) - new_ps_action.triggered.connect(lambda: self.open_local_terminal(SHELL_POWERSHELL)) - file_menu.addAction(new_ps_action) - - new_cmd_action = QAction("Nouveau terminal local (Invite de commandes)", self) - new_cmd_action.triggered.connect(lambda: self.open_local_terminal(SHELL_CMD)) - file_menu.addAction(new_cmd_action) - - file_menu.addSeparator() - quit_action = QAction("Quitter", self) - quit_action.triggered.connect(self.close) - file_menu.addAction(quit_action) - - keys_menu = self.menuBar().addMenu("&Cles SSH") - manage_keys_action = QAction("Gestionnaire de cles...", self) - manage_keys_action.triggered.connect(self.open_key_manager) - keys_menu.addAction(manage_keys_action) - - view_menu = self.menuBar().addMenu("&Affichage") - theme_menu = view_menu.addMenu("Theme") - self.theme_group = QActionGroup(self) - self.theme_group.setExclusive(True) - for theme_id, label in THEME_LABELS.items(): - action = QAction(label, self, checkable=True) - action.setChecked(theme_id == current_theme()) - action.triggered.connect(lambda _checked=False, t=theme_id: self.change_theme(t)) - self.theme_group.addAction(action) - theme_menu.addAction(action) - - hosts_menu = self.menuBar().addMenu("&Hotes connus") - manage_hosts_action = QAction("Gestionnaire des hotes connus...", self) - manage_hosts_action.triggered.connect(self.open_known_hosts_manager) - hosts_menu.addAction(manage_hosts_action) - - help_menu = self.menuBar().addMenu("&Aide") - about_action = QAction("A propos", self) - about_action.triggered.connect(self.show_about) - help_menu.addAction(about_action) - - def _build_sidebar(self) -> None: - dock = QDockWidget("Sessions", self) - dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) - - 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) - term_menu.addAction("PowerShell", lambda: self.open_local_terminal(SHELL_POWERSHELL)) - term_menu.addAction("Invite de commandes (cmd)", lambda: self.open_local_terminal(SHELL_CMD)) - term_btn.setMenu(term_menu) - toolbar.addWidget(term_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) - - dock.setWidget(container) - self.addDockWidget(Qt.LeftDockWidgetArea, dock) - - def _build_tabs(self) -> None: - self.tabs = QTabWidget() - self.tabs.setTabsClosable(True) - self.tabs.setMovable(True) - 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) - 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 = SHELL_POWERSHELL) -> None: - tab = LocalTerminalTabWidget(shell) - tab.sftp_requested.connect(self.quick_sftp) - label = "PowerShell" if shell == SHELL_POWERSHELL else "cmd" - index = self.tabs.addTab(tab, f"💻 {label}") - self.tabs.setCurrentIndex(index) - - 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 ---------- - def change_theme(self, theme: str) -> None: - """Applique un theme a chaud : feuille de style de l'interface + couleurs des - terminaux deja ouverts (qui ont leur propre palette, hors QSS).""" - app = QApplication.instance() - if app is None: - return - apply_theme(app, theme) - save_theme(theme) - for widget in self.findChildren(TerminalWidget): - widget.refresh_theme() - - def open_key_manager(self) -> None: - dialog = KeyManagerDialog(self.key_manager, parent=self) - dialog.exec() - - def open_known_hosts_manager(self) -> None: - dialog = KnownHostsDialog(parent=self) - dialog.exec() - - def show_about(self) -> None: - QMessageBox.information( - self, "A propos", - "PySSH Client\n\nClient SSH/SFTP local avec gestion de sessions, de cles, " - "et terminaux locaux integres.\nBase sur PySide6, paramiko, pywinpty et pyte.", - ) - - 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() - super().closeEvent(event) +"""Fenetre principale : arborescence des sessions, onglets terminal/SFTP.""" +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.QtWidgets import ( + QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget, + QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter, + QStatusBar, QFrame, QApplication, QLineEdit, QFileDialog, +) + +from app.session_manager import SessionStore, Session, AUTH_KEY, AUTH_PASSWORD +from app.key_manager import KeyManager +from app.ssh_connection import SSHConnection +from app.local_terminal import LocalTerminalConnection, SHELL_POWERSHELL, SHELL_CMD, SHELL_LABELS +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.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change +from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme + +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._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 _on_connected(self) -> None: + self.status_label.setText(f"● Connecte : {self.session.username}@{self.session.host}:{self.session.port}") + # 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 _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() + + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("PySSH Client") + self.resize(1280, 820) + + self.session_store = SessionStore() + self.key_manager = KeyManager() + + self._build_menu() + self._build_sidebar() + self._build_tabs() + self.setStatusBar(QStatusBar()) + + self.reload_sessions() + + # demarre directement dans un terminal utilisable, sans avoir a en choisir un + self.open_local_terminal(SHELL_POWERSHELL) + + # ---------- 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) + 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) + 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) + file_menu.addAction(quick_sftp_action) + + file_menu.addSeparator() + + new_ps_action = QAction("Nouveau terminal local (PowerShell)", self) + new_ps_action.setShortcut(QKeySequence("Ctrl+Shift+T")) + new_ps_action.triggered.connect(lambda: self.open_local_terminal(SHELL_POWERSHELL)) + file_menu.addAction(new_ps_action) + + new_cmd_action = QAction("Nouveau terminal local (Invite de commandes)", self) + new_cmd_action.triggered.connect(lambda: self.open_local_terminal(SHELL_CMD)) + file_menu.addAction(new_cmd_action) + + file_menu.addSeparator() + quit_action = QAction("Quitter", self) + quit_action.triggered.connect(self.close) + file_menu.addAction(quit_action) + + keys_menu = self.menuBar().addMenu("&Cles SSH") + manage_keys_action = QAction("Gestionnaire de cles...", self) + manage_keys_action.triggered.connect(self.open_key_manager) + keys_menu.addAction(manage_keys_action) + + view_menu = self.menuBar().addMenu("&Affichage") + theme_menu = view_menu.addMenu("Theme") + self.theme_group = QActionGroup(self) + self.theme_group.setExclusive(True) + for theme_id, label in THEME_LABELS.items(): + action = QAction(label, self, checkable=True) + action.setChecked(theme_id == current_theme()) + action.triggered.connect(lambda _checked=False, t=theme_id: self.change_theme(t)) + self.theme_group.addAction(action) + theme_menu.addAction(action) + + hosts_menu = self.menuBar().addMenu("&Hotes connus") + manage_hosts_action = QAction("Gestionnaire des hotes connus...", self) + manage_hosts_action.triggered.connect(self.open_known_hosts_manager) + hosts_menu.addAction(manage_hosts_action) + + help_menu = self.menuBar().addMenu("&Aide") + about_action = QAction("A propos", self) + about_action.triggered.connect(self.show_about) + help_menu.addAction(about_action) + + def _build_sidebar(self) -> None: + dock = QDockWidget("Sessions", self) + dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) + + 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) + term_menu.addAction("PowerShell", lambda: self.open_local_terminal(SHELL_POWERSHELL)) + term_menu.addAction("Invite de commandes (cmd)", lambda: self.open_local_terminal(SHELL_CMD)) + term_btn.setMenu(term_menu) + toolbar.addWidget(term_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) + + dock.setWidget(container) + self.addDockWidget(Qt.LeftDockWidgetArea, dock) + + def _build_tabs(self) -> None: + self.tabs = QTabWidget() + self.tabs.setTabsClosable(True) + self.tabs.setMovable(True) + 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) + 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 = SHELL_POWERSHELL) -> None: + tab = LocalTerminalTabWidget(shell) + tab.sftp_requested.connect(self.quick_sftp) + label = "PowerShell" if shell == SHELL_POWERSHELL else "cmd" + index = self.tabs.addTab(tab, f"💻 {label}") + self.tabs.setCurrentIndex(index) + + 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 ---------- + def change_theme(self, theme: str) -> None: + """Applique un theme a chaud : feuille de style de l'interface + couleurs des + terminaux deja ouverts (qui ont leur propre palette, hors QSS).""" + app = QApplication.instance() + if app is None: + return + apply_theme(app, theme) + save_theme(theme) + for widget in self.findChildren(TerminalWidget): + widget.refresh_theme() + + def open_key_manager(self) -> None: + dialog = KeyManagerDialog(self.key_manager, parent=self) + dialog.exec() + + def open_known_hosts_manager(self) -> None: + dialog = KnownHostsDialog(parent=self) + dialog.exec() + + def show_about(self) -> None: + QMessageBox.information( + self, "A propos", + "PySSH Client\n\nClient SSH/SFTP local avec gestion de sessions, de cles, " + "et terminaux locaux integres.\nBase sur PySide6, paramiko, pywinpty et pyte.", + ) + + 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() + super().closeEvent(event) diff --git a/app/sftp_panel.py b/app/sftp_panel.py index e8c0309..474cf1e 100644 --- a/app/sftp_panel.py +++ b/app/sftp_panel.py @@ -5,12 +5,13 @@ from __future__ import annotations import json import os import posixpath +import tempfile import stat from datetime import datetime from pathlib import Path import paramiko -from PySide6.QtCore import Qt, QThread, Signal, QMimeData, QPoint +from PySide6.QtCore import Qt, QThread, Signal, QMimeData, QPoint, QFileSystemWatcher from PySide6.QtGui import QDrag, QKeySequence, QShortcut from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLineEdit, QToolButton, QTreeWidget, @@ -324,9 +325,11 @@ class FilePanel(QWidget): act_rename.setEnabled(has_selection) act_delete = menu.addAction("Supprimer (Suppr)") act_delete.setEnabled(has_selection) - act_perms = None + act_perms = act_edit = None if self.side == "remote": menu.addSeparator() + act_edit = menu.addAction("Editer (renvoi automatique a l'enregistrement)") + act_edit.setEnabled(has_selection) act_perms = menu.addAction("Permissions...") act_perms.setEnabled(has_selection) menu.addSeparator() @@ -340,6 +343,8 @@ class FilePanel(QWidget): self.rename_selected() elif action == act_delete: self.delete_selected() + elif act_edit is not None and action == act_edit: + self.edit_selected() elif act_perms is not None and action == act_perms: self.change_permissions() elif action == act_mkdir: @@ -361,6 +366,7 @@ class FilePanel(QWidget): def rename_selected(self) -> None: ... def delete_selected(self) -> None: ... def change_permissions(self) -> None: ... + def edit_selected(self) -> None: ... class LocalFilePanel(FilePanel): @@ -453,6 +459,10 @@ class RemoteFilePanel(FilePanel): def __init__(self, sftp: paramiko.SFTPClient, parent=None): super().__init__("🌐 Distant", "remote", parent) self.sftp = sftp + # fichiers ouverts pour edition : chemin local temporaire -> chemin distant + self._edited: dict[str, str] = {} + self._watcher = QFileSystemWatcher(self) + self._watcher.fileChanged.connect(self._on_edited_file_changed) try: home = self.sftp.normalize(".") except Exception: @@ -546,6 +556,55 @@ class RemoteFilePanel(FilePanel): self.sftp.remove(sub) self.sftp.rmdir(path) + # ---------- edition en place ---------- + def edit_selected(self) -> None: + """Telecharge le fichier, l'ouvre dans l'editeur par defaut du systeme et le + renvoie automatiquement a chaque enregistrement (facon WinSCP).""" + entries = [(n, d) for n, d in self.selected_entries() if not d] + if not entries: + QMessageBox.information(self, "Edition", "Selectionnez un fichier (pas un dossier).") + return + name = entries[0][0] + remote_path = self.join(self.current_path, name) + + local_dir = Path(tempfile.gettempdir()) / "pyssh-edit" + local_dir.mkdir(parents=True, exist_ok=True) + local_path = local_dir / name + try: + self.sftp.get(remote_path, str(local_path)) + except (IOError, OSError) as exc: + QMessageBox.critical(self, "Erreur", f"Telechargement impossible : {exc}") + return + + self._edited[str(local_path)] = remote_path + self._watcher.addPath(str(local_path)) + self.status.setText(f"Edition de {name} — enregistrez pour renvoyer sur le serveur") + try: + os.startfile(str(local_path)) # ouvre avec l'application associee + except OSError as exc: + QMessageBox.warning( + self, "Ouverture impossible", + f"Le fichier a ete telecharge dans :\n{local_path}\n\n" + f"Ouvrez-le manuellement ; il sera renvoye a chaque enregistrement.\n({exc})", + ) + + def _on_edited_file_changed(self, local_path: str) -> None: + remote_path = self._edited.get(local_path) + if not remote_path: + return + try: + self.sftp.put(local_path, remote_path) + except (IOError, OSError) as exc: + self.status.setText(f"Echec du renvoi : {exc}") + return + # Beaucoup d'editeurs enregistrent en remplacant le fichier : le chemin surveille + # disparait alors du watcher, il faut le reinscrire pour les enregistrements suivants. + if local_path not in self._watcher.files() and Path(local_path).exists(): + self._watcher.addPath(local_path) + self.refresh() + # apres refresh() : sinon le message serait ecrase par le compteur d'elements + self.status.setText(f"↑ {posixpath.basename(remote_path)} renvoye sur le serveur") + def change_permissions(self) -> None: entries = self.selected_entries() if not entries: