snoopy/pyssh-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
23f5aaccd11d7b4d67d3c12ff333084845aeb3fa
pyssh-client/app/ui/tabs.py
Florian 47005b1e26 refactor: decouper main_window.py en controllers/services/ui 
main_window.py concentrait la fenetre principale, l'arbre des sessions, les 4
classes d'onglets, la detection de commandes ssh et le multi-send : 983 lignes
et une seule classe qui pilotait toute l'application.

- app/services/ssh_target.py : detection d'une cible ssh dans un terminal (pur,
  sans dependance Qt)
- app/ui/tabs.py : BaseTerminalTab / SessionTabWidget / SFTPTabWidget /
  LocalTerminalTabWidget (deplaces tels quels)
- app/controllers/session_controller.py : panneau lateral (arbre sessions +
  terminaux locaux) et CRUD des sessions
- app/controllers/connection_controller.py : cycle de vie des onglets
  (ouverture, connexion rapide, renommage, fermeture)
- app/controllers/broadcast_manager.py : bouton et logique du mode Multi-Send

main_window.py ne fait plus que construire ces composants et les relier par
signaux (983 -> ~200 lignes). Comportement inchange ; MainWindow.tabs et
detect_ssh_target restent importables depuis app.main_window pour la
compatibilite avec les tests existants.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-05 01:42:31 +02:00

451 lines
19 KiB
Python
Raw Blame History

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