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: