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>
This commit is contained in:
Florian
0
app/controllers/__init__.py
Normal file
0
app/controllers/__init__.py
Normal file
47
app/controllers/broadcast_manager.py
Normal file
47
app/controllers/broadcast_manager.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""Envoi simultane (« Multi-Send ") : une frappe faite dans un onglet terminal part
|
||||
aussi vers tous les autres onglets terminal ouverts, quand le mode est actif."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable, Iterable
|
||||
|
||||
from PySide6.QtCore import QObject
|
||||
from PySide6.QtWidgets import QToolButton, QWidget
|
||||
|
||||
|
||||
class BroadcastManager(QObject):
|
||||
"""Ne connait pas le QTabWidget : `tabs_provider` lui fournit les onglets courants,
|
||||
ce qui la garde independante de ConnectionController."""
|
||||
|
||||
def __init__(self, tabs_provider: Callable[[], Iterable[QWidget]], parent=None):
|
||||
super().__init__(parent)
|
||||
self._tabs_provider = tabs_provider
|
||||
|
||||
self.button = QToolButton()
|
||||
self.button.setText("⇉ Multi")
|
||||
self.button.setCheckable(True)
|
||||
self.button.setToolTip(
|
||||
"Envoi simultane : ce que vous tapez part vers tous les onglets terminal ouverts"
|
||||
)
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
return self.button.isChecked()
|
||||
|
||||
def register(self, tab) -> None:
|
||||
"""A appeler pour chaque nouvel onglet terminal (SSH ou local) susceptible
|
||||
d'emettre et de recevoir des donnees diffusees."""
|
||||
tab.terminal.data_to_send.connect(lambda data, source=tab: self._broadcast(data, source))
|
||||
|
||||
def _broadcast(self, data: bytes, source) -> None:
|
||||
if not self.enabled:
|
||||
return
|
||||
# import tardif : evite un cycle (app.ui.tabs importe des dialogues, pas ce module)
|
||||
from app.ui.tabs import BaseTerminalTab
|
||||
|
||||
for tab in self._tabs_provider():
|
||||
if tab is source or not isinstance(tab, BaseTerminalTab):
|
||||
continue
|
||||
if tab.connection:
|
||||
# on ecrit directement dans la connexion : passer par le terminal
|
||||
# relancerait la diffusion et boucherait a l'infini
|
||||
tab.connection.send(data)
|
||||
183
app/controllers/connection_controller.py
Normal file
183
app/controllers/connection_controller.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""Cycle de vie des onglets (terminal SSH, terminal local, SFTP autonome) :
|
||||
ouverture, connexion rapide, renommage, fermeture."""
|
||||
from __future__ import annotations
|
||||
|
||||
from PySide6.QtCore import Qt, Signal, QObject
|
||||
from PySide6.QtWidgets import QTabWidget, QWidget, QMenu, QInputDialog
|
||||
|
||||
from app.session_manager import SessionStore, Session, AUTH_PASSWORD
|
||||
from app.key_manager import KeyManager
|
||||
from app.local_terminal import DEFAULT_SHELL, SHELL_LABELS
|
||||
from app.dialogs.session_dialog import SessionDialog
|
||||
from app.dialogs.connect_prompt import ask_secret
|
||||
from app.services.ssh_target import detect_ssh_target
|
||||
from app.controllers.broadcast_manager import BroadcastManager
|
||||
from app.ui.tabs import BaseTerminalTab, SessionTabWidget, SFTPTabWidget, LocalTerminalTabWidget
|
||||
|
||||
|
||||
class ConnectionController(QObject):
|
||||
"""Possede le QTabWidget central. `session_saved` est emis quand une connexion
|
||||
rapide est aussi enregistree comme session, pour que le panneau lateral se
|
||||
rafraichisse sans que ce controleur ait besoin de le connaitre."""
|
||||
|
||||
session_saved = Signal()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
session_store: SessionStore,
|
||||
key_manager: KeyManager,
|
||||
broadcast: BroadcastManager,
|
||||
parent_widget: QWidget | None = None,
|
||||
):
|
||||
super().__init__(parent_widget)
|
||||
self.session_store = session_store
|
||||
self.key_manager = key_manager
|
||||
self.broadcast = broadcast
|
||||
self._parent_widget = parent_widget
|
||||
|
||||
self.tabs = QTabWidget(parent_widget)
|
||||
self.tabs.setTabsClosable(True)
|
||||
self.tabs.setMovable(True)
|
||||
self.tabs.tabBarDoubleClicked.connect(self.rename_tab)
|
||||
self.tabs.tabBar().setContextMenuPolicy(Qt.CustomContextMenu)
|
||||
self.tabs.tabBar().customContextMenuRequested.connect(self._on_tab_context_menu)
|
||||
self.tabs.tabCloseRequested.connect(self._on_tab_close)
|
||||
|
||||
# ---------- ouverture ----------
|
||||
def open_session(self, session: Session) -> None:
|
||||
secret = ""
|
||||
if session.auth_type == AUTH_PASSWORD:
|
||||
text, ok = ask_secret(
|
||||
self._parent_widget, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :"
|
||||
)
|
||||
if not ok:
|
||||
return
|
||||
secret = text
|
||||
|
||||
tab = SessionTabWidget(session, secret)
|
||||
self.broadcast.register(tab)
|
||||
index = self.tabs.addTab(tab, f"🖥 {session.name}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def open_sftp_session(self, session: Session) -> None:
|
||||
"""Ouvre un onglet SFTP autonome (double panneau) pour cette session."""
|
||||
secret = ""
|
||||
if session.auth_type == AUTH_PASSWORD:
|
||||
text, ok = ask_secret(
|
||||
self._parent_widget, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :"
|
||||
)
|
||||
if not ok:
|
||||
return
|
||||
secret = text
|
||||
|
||||
tab = SFTPTabWidget(session, secret)
|
||||
index = self.tabs.addTab(tab, f"📁 {session.name}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def open_local_terminal(self, shell: str = DEFAULT_SHELL) -> None:
|
||||
tab = LocalTerminalTabWidget(shell)
|
||||
tab.sftp_requested.connect(self.quick_sftp)
|
||||
self.broadcast.register(tab)
|
||||
label = SHELL_LABELS.get(shell, shell)
|
||||
index = self.tabs.addTab(tab, f"💻 {label}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def quick_sftp(self, target: dict | None = None) -> None:
|
||||
"""Ouvre un SFTP en saisissant directement les infos de connexion, sans favori.
|
||||
|
||||
Si l'onglet courant est un terminal ou une commande ssh a ete tapee, l'hote,
|
||||
l'utilisateur et le port sont pre-remplis automatiquement."""
|
||||
if target is None:
|
||||
target = self._current_tab_ssh_target()
|
||||
self._quick_connect(sftp=True, target=target)
|
||||
|
||||
def quick_ssh(self) -> None:
|
||||
self._quick_connect(sftp=False, target=self._current_tab_ssh_target())
|
||||
|
||||
def _current_tab_ssh_target(self) -> dict | None:
|
||||
widget = self.tabs.currentWidget()
|
||||
if isinstance(widget, SessionTabWidget):
|
||||
return {
|
||||
"user": widget.session.username,
|
||||
"host": widget.session.host,
|
||||
"port": widget.session.port,
|
||||
}
|
||||
if isinstance(widget, BaseTerminalTab):
|
||||
return detect_ssh_target(widget.terminal.buffer_text())
|
||||
return None
|
||||
|
||||
def _quick_connect(self, sftp: bool, target: dict | None = None) -> None:
|
||||
prefill = None
|
||||
if target:
|
||||
prefill = Session(
|
||||
name=target.get("host", ""),
|
||||
host=target.get("host", ""),
|
||||
username=target.get("user", ""),
|
||||
port=target.get("port", 22),
|
||||
)
|
||||
dialog = SessionDialog(
|
||||
self.key_manager,
|
||||
session=prefill,
|
||||
parent=self._parent_widget,
|
||||
connect_label="Se connecter",
|
||||
title="Connexion SFTP" if sftp else "Connexion SSH",
|
||||
)
|
||||
if not dialog.exec():
|
||||
return
|
||||
session = dialog.result_session()
|
||||
if dialog.should_save():
|
||||
self.session_store.add(session)
|
||||
self.session_saved.emit()
|
||||
if sftp:
|
||||
self.open_sftp_session(session)
|
||||
else:
|
||||
self.open_session(session)
|
||||
|
||||
# ---------- gestion des onglets ----------
|
||||
def rename_tab(self, index: int) -> None:
|
||||
if index < 0:
|
||||
return
|
||||
actuel = self.tabs.tabText(index)
|
||||
nouveau, ok = QInputDialog.getText(
|
||||
self._parent_widget, "Renommer l'onglet", "Nom de l'onglet :", text=actuel
|
||||
)
|
||||
if ok and nouveau.strip():
|
||||
self.tabs.setTabText(index, nouveau.strip())
|
||||
|
||||
def _on_tab_context_menu(self, pos) -> None:
|
||||
bar = self.tabs.tabBar()
|
||||
index = bar.tabAt(pos)
|
||||
if index < 0:
|
||||
return
|
||||
menu = QMenu(self._parent_widget)
|
||||
act_rename = menu.addAction("Renommer... (double-clic)")
|
||||
menu.addSeparator()
|
||||
act_close = menu.addAction("Fermer cet onglet")
|
||||
act_close_others = menu.addAction("Fermer les autres onglets")
|
||||
act_close_others.setEnabled(self.tabs.count() > 1)
|
||||
|
||||
action = menu.exec(bar.mapToGlobal(pos))
|
||||
if action == act_rename:
|
||||
self.rename_tab(index)
|
||||
elif action == act_close:
|
||||
self._on_tab_close(index)
|
||||
elif action == act_close_others:
|
||||
# on ferme par la fin : retirer un onglet decale les suivants
|
||||
cible = self.tabs.widget(index)
|
||||
for other in range(self.tabs.count() - 1, -1, -1):
|
||||
if self.tabs.widget(other) is not cible:
|
||||
self._on_tab_close(other)
|
||||
|
||||
def _on_tab_close(self, index: int) -> None:
|
||||
widget = self.tabs.widget(index)
|
||||
if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)):
|
||||
widget.close_connection()
|
||||
self.tabs.removeTab(index)
|
||||
|
||||
def close_all(self) -> None:
|
||||
"""A appeler a la fermeture de l'application : ferme proprement toutes les
|
||||
connexions (SSH, locales, SFTP) avant que les widgets ne soient detruits."""
|
||||
for i in range(self.tabs.count()):
|
||||
widget = self.tabs.widget(i)
|
||||
if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)):
|
||||
widget.close_connection()
|
||||
232
app/controllers/session_controller.py
Normal file
232
app/controllers/session_controller.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""Panneau lateral : arbre des sessions SSH enregistrees + terminaux locaux, et
|
||||
leur CRUD (creer / modifier / dupliquer / supprimer). N'ouvre aucune connexion
|
||||
elle-meme : elle emet des signaux que ConnectionController ecoute."""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import uuid
|
||||
|
||||
from PySide6.QtCore import Qt, Signal, QObject
|
||||
from PySide6.QtWidgets import (
|
||||
QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QTreeWidget, QTreeWidgetItem,
|
||||
QMenu, QMessageBox,
|
||||
)
|
||||
|
||||
from app.session_manager import SessionStore, Session
|
||||
from app.key_manager import KeyManager
|
||||
from app.local_terminal import SHELL_LABELS
|
||||
from app.dialogs.session_dialog import SessionDialog
|
||||
|
||||
SESSION_ROLE = Qt.UserRole + 10
|
||||
LOCAL_SHELL_ROLE = Qt.UserRole + 11
|
||||
|
||||
|
||||
class SessionController(QObject):
|
||||
open_session_requested = Signal(object) # Session
|
||||
open_sftp_requested = Signal(object) # Session
|
||||
open_local_shell_requested = Signal(str) # identifiant de shell
|
||||
quick_sftp_requested = Signal()
|
||||
manage_keys_requested = Signal()
|
||||
manage_known_hosts_requested = Signal()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
session_store: SessionStore,
|
||||
key_manager: KeyManager,
|
||||
broadcast_button: QToolButton,
|
||||
parent_widget: QWidget | None = None,
|
||||
):
|
||||
super().__init__(parent_widget)
|
||||
self.session_store = session_store
|
||||
self.key_manager = key_manager
|
||||
self._parent_widget = parent_widget
|
||||
|
||||
self.widget = QWidget(parent_widget)
|
||||
layout = QVBoxLayout(self.widget)
|
||||
layout.setContentsMargins(6, 6, 6, 6)
|
||||
layout.setSpacing(6)
|
||||
|
||||
toolbar = QHBoxLayout()
|
||||
new_btn = QToolButton()
|
||||
new_btn.setObjectName("primary")
|
||||
new_btn.setText("+ Session")
|
||||
new_btn.clicked.connect(self.new_session)
|
||||
toolbar.addWidget(new_btn)
|
||||
|
||||
term_btn = QToolButton()
|
||||
term_btn.setText("💻 Terminal")
|
||||
term_btn.setPopupMode(QToolButton.InstantPopup)
|
||||
term_menu = QMenu(term_btn)
|
||||
for shell_id, label in SHELL_LABELS.items():
|
||||
term_menu.addAction(label, lambda s=shell_id: self.open_local_shell_requested.emit(s))
|
||||
term_btn.setMenu(term_menu)
|
||||
toolbar.addWidget(term_btn)
|
||||
|
||||
toolbar.addWidget(broadcast_button)
|
||||
|
||||
sftp_btn = QToolButton()
|
||||
sftp_btn.setText("📁 SFTP")
|
||||
sftp_btn.setToolTip("Ouvrir un gestionnaire de fichiers SFTP (sans session enregistree)")
|
||||
sftp_btn.clicked.connect(self.quick_sftp_requested.emit)
|
||||
toolbar.addWidget(sftp_btn)
|
||||
|
||||
toolbar.addStretch()
|
||||
|
||||
manage_btn = QToolButton()
|
||||
manage_btn.setText("⚙")
|
||||
manage_btn.setToolTip("Cles SSH et hotes connus")
|
||||
manage_btn.setPopupMode(QToolButton.InstantPopup)
|
||||
manage_menu = QMenu(manage_btn)
|
||||
manage_menu.addAction("Gestionnaire de cles SSH...", self.manage_keys_requested.emit)
|
||||
manage_menu.addAction("Gestionnaire des hotes connus...", self.manage_known_hosts_requested.emit)
|
||||
manage_btn.setMenu(manage_menu)
|
||||
toolbar.addWidget(manage_btn)
|
||||
|
||||
layout.addLayout(toolbar)
|
||||
|
||||
self.tree = QTreeWidget()
|
||||
self.tree.setHeaderHidden(True)
|
||||
self.tree.itemDoubleClicked.connect(self._on_tree_double_click)
|
||||
self.tree.setContextMenuPolicy(Qt.CustomContextMenu)
|
||||
self.tree.customContextMenuRequested.connect(self._on_tree_context_menu)
|
||||
layout.addWidget(self.tree)
|
||||
|
||||
# Sans minimum explicite, la largeur plancher est imposee par le contenu
|
||||
# (les noms de sessions), ce qui empeche de retrecir vraiment le panneau.
|
||||
self.widget.setMinimumWidth(120)
|
||||
self.tree.setMinimumWidth(0)
|
||||
|
||||
self.reload()
|
||||
|
||||
# ---------- arbre ----------
|
||||
def reload(self) -> None:
|
||||
self.tree.clear()
|
||||
|
||||
locals_header = QTreeWidgetItem(["💻 Terminaux locaux"])
|
||||
locals_header.setFlags(locals_header.flags() & ~Qt.ItemIsSelectable)
|
||||
font = locals_header.font(0)
|
||||
font.setBold(True)
|
||||
locals_header.setFont(0, font)
|
||||
self.tree.addTopLevelItem(locals_header)
|
||||
for shell, label in SHELL_LABELS.items():
|
||||
child = QTreeWidgetItem([f" {label}"])
|
||||
child.setData(0, LOCAL_SHELL_ROLE, shell)
|
||||
locals_header.addChild(child)
|
||||
|
||||
sessions_header = QTreeWidgetItem(["🔐 Sessions SSH"])
|
||||
sessions_header.setFlags(sessions_header.flags() & ~Qt.ItemIsSelectable)
|
||||
sessions_header.setFont(0, font)
|
||||
self.tree.addTopLevelItem(sessions_header)
|
||||
|
||||
folder_items: dict[str, QTreeWidgetItem] = {}
|
||||
|
||||
def get_folder(path: str) -> QTreeWidgetItem:
|
||||
if not path:
|
||||
return sessions_header
|
||||
if path in folder_items:
|
||||
return folder_items[path]
|
||||
parent_path, _, name = path.rpartition("/")
|
||||
parent_item = get_folder(parent_path)
|
||||
item = QTreeWidgetItem([f"📁 {name}"])
|
||||
item.setFlags(item.flags() & ~Qt.ItemIsSelectable)
|
||||
parent_item.addChild(item)
|
||||
folder_items[path] = item
|
||||
return item
|
||||
|
||||
for session in sorted(self.session_store.sessions, key=lambda s: s.name.lower()):
|
||||
parent = get_folder(session.group)
|
||||
item = QTreeWidgetItem([f"🖥 {session.name} ({session.host})"])
|
||||
item.setData(0, SESSION_ROLE, session.id)
|
||||
parent.addChild(item)
|
||||
|
||||
if not self.session_store.sessions:
|
||||
hint = QTreeWidgetItem([" (aucune session — bouton « + Session » ou « 📁 SFTP »)"])
|
||||
hint.setFlags(hint.flags() & ~Qt.ItemIsSelectable)
|
||||
hint.setForeground(0, Qt.gray)
|
||||
sessions_header.addChild(hint)
|
||||
|
||||
self.tree.expandAll()
|
||||
|
||||
def _session_from_item(self, item: QTreeWidgetItem) -> Session | None:
|
||||
if not item:
|
||||
return None
|
||||
session_id = item.data(0, SESSION_ROLE)
|
||||
if not session_id:
|
||||
return None
|
||||
return next((s for s in self.session_store.sessions if s.id == session_id), None)
|
||||
|
||||
def _shell_from_item(self, item: QTreeWidgetItem) -> str | None:
|
||||
if not item:
|
||||
return None
|
||||
return item.data(0, LOCAL_SHELL_ROLE)
|
||||
|
||||
def _on_tree_double_click(self, item: QTreeWidgetItem, _col: int) -> None:
|
||||
session = self._session_from_item(item)
|
||||
if session:
|
||||
self.open_session_requested.emit(session)
|
||||
return
|
||||
shell = self._shell_from_item(item)
|
||||
if shell:
|
||||
self.open_local_shell_requested.emit(shell)
|
||||
|
||||
def _on_tree_context_menu(self, pos) -> None:
|
||||
item = self.tree.itemAt(pos)
|
||||
menu = QMenu(self.widget)
|
||||
session = self._session_from_item(item) if item else None
|
||||
shell = self._shell_from_item(item) if item else None
|
||||
|
||||
if session:
|
||||
connect_action = menu.addAction("🖥 Ouvrir un terminal SSH")
|
||||
sftp_action = menu.addAction("📁 Ouvrir en SFTP (gestionnaire de fichiers)")
|
||||
menu.addSeparator()
|
||||
edit_action = menu.addAction("Modifier...")
|
||||
duplicate_action = menu.addAction("Dupliquer")
|
||||
delete_action = menu.addAction("Supprimer")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == connect_action:
|
||||
self.open_session_requested.emit(session)
|
||||
elif action == sftp_action:
|
||||
self.open_sftp_requested.emit(session)
|
||||
elif action == edit_action:
|
||||
self.edit_session(session)
|
||||
elif action == duplicate_action:
|
||||
self.duplicate_session(session)
|
||||
elif action == delete_action:
|
||||
self.delete_session(session)
|
||||
elif shell:
|
||||
open_action = menu.addAction("Ouvrir")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == open_action:
|
||||
self.open_local_shell_requested.emit(shell)
|
||||
else:
|
||||
new_action = menu.addAction("Nouvelle session...")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == new_action:
|
||||
self.new_session()
|
||||
|
||||
# ---------- CRUD ----------
|
||||
def new_session(self) -> None:
|
||||
dialog = SessionDialog(self.key_manager, parent=self._parent_widget)
|
||||
if dialog.exec():
|
||||
self.session_store.add(dialog.result_session())
|
||||
self.reload()
|
||||
|
||||
def edit_session(self, session: Session) -> None:
|
||||
dialog = SessionDialog(self.key_manager, session=session, parent=self._parent_widget)
|
||||
if dialog.exec():
|
||||
self.session_store.update(dialog.result_session())
|
||||
self.reload()
|
||||
|
||||
def duplicate_session(self, session: Session) -> None:
|
||||
new_session = copy.deepcopy(session)
|
||||
new_session.id = str(uuid.uuid4())
|
||||
new_session.name = f"{session.name} (copie)"
|
||||
self.session_store.add(new_session)
|
||||
self.reload()
|
||||
|
||||
def delete_session(self, session: Session) -> None:
|
||||
if QMessageBox.question(
|
||||
self._parent_widget, "Supprimer", f"Supprimer la session '{session.name}' ?"
|
||||
) == QMessageBox.Yes:
|
||||
self.session_store.delete(session.id)
|
||||
self.reload()
|
||||
@@ -1,492 +1,30 @@
|
||||
"""Fenetre principale : arborescence des sessions, onglets terminal/SFTP."""
|
||||
"""Fenetre principale : assemble le panneau des sessions, les onglets, les menus.
|
||||
|
||||
La logique metier vit dans des controleurs dedies (app.controllers) et les widgets
|
||||
d'onglet dans app.ui.tabs ; cette classe se contente de les construire et de les
|
||||
relier.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import re
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import Qt, Signal
|
||||
from PySide6.QtGui import QAction, QActionGroup, QKeySequence, QShortcut
|
||||
from PySide6.QtCore import Qt
|
||||
from PySide6.QtGui import QAction, QActionGroup, QKeySequence
|
||||
from PySide6.QtWidgets import (
|
||||
QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget,
|
||||
QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter,
|
||||
QStatusBar, QFrame, QApplication, QLineEdit, QFileDialog, QInputDialog,
|
||||
QMainWindow, QDockWidget, QMessageBox, QStatusBar, QApplication,
|
||||
)
|
||||
|
||||
from app.session_manager import SessionStore, Session, AUTH_KEY, AUTH_PASSWORD
|
||||
from app.session_manager import SessionStore
|
||||
from app.key_manager import KeyManager
|
||||
from app.ssh_connection import SSHConnection
|
||||
from app.local_terminal import (
|
||||
LocalTerminalConnection, SHELL_POWERSHELL, SHELL_CMD, SHELL_LABELS, DEFAULT_SHELL,
|
||||
)
|
||||
from app.local_terminal import SHELL_LABELS, DEFAULT_SHELL
|
||||
from app.terminal_widget import TerminalWidget
|
||||
from app.sftp_panel import SFTPPanel
|
||||
from app.dialogs.session_dialog import SessionDialog
|
||||
from app.dialogs.key_dialog import KeyManagerDialog
|
||||
from app.dialogs.known_hosts_dialog import KnownHostsDialog
|
||||
from app.dialogs.tunnel_dialog import TunnelDialog
|
||||
from app.tunnels import TunnelManager
|
||||
from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change
|
||||
from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme
|
||||
from app.controllers.broadcast_manager import BroadcastManager
|
||||
from app.controllers.session_controller import SessionController
|
||||
from app.controllers.connection_controller import ConnectionController
|
||||
|
||||
SESSION_ROLE = Qt.UserRole + 10
|
||||
LOCAL_SHELL_ROLE = Qt.UserRole + 11
|
||||
|
||||
# Repere une commande ssh tapee a la main dans un terminal, pour proposer le SFTP
|
||||
# sur cet hote sans avoir a creer de favori.
|
||||
SSH_COMMAND_RE = re.compile(r"\bssh\s+(?P<args>[^\r\n|;&]+)")
|
||||
# options ssh qui consomment la valeur suivante (a ne pas confondre avec l'hote)
|
||||
SSH_VALUE_OPTIONS = {
|
||||
"-p", "-i", "-l", "-o", "-F", "-J", "-b", "-c", "-D", "-L", "-R", "-W", "-e", "-m", "-Q",
|
||||
}
|
||||
HOST_RE = re.compile(r"^[A-Za-z0-9][\w.\-]*$")
|
||||
# sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles
|
||||
ANSI_ESCAPE_RE = re.compile(
|
||||
r"\x1b\[[0-9;?]*[a-zA-Z]" # CSI : couleurs, deplacements du curseur
|
||||
r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC : titre de fenetre
|
||||
r"|\x1b[()][A-Z0-9]" # jeux de caracteres
|
||||
r"|\x1b[=>]" # mode clavier
|
||||
r"|[\x00-\x08\x0b\x0c\x0e-\x1f]" # autres caracteres de controle
|
||||
)
|
||||
|
||||
|
||||
def _parse_ssh_args(args: str) -> dict | None:
|
||||
tokens = args.split()
|
||||
user, host, port = "", "", 22
|
||||
index = 0
|
||||
while index < len(tokens):
|
||||
token = tokens[index]
|
||||
if token.startswith("-"):
|
||||
if token in SSH_VALUE_OPTIONS:
|
||||
value = tokens[index + 1] if index + 1 < len(tokens) else ""
|
||||
if token == "-p" and value.isdigit():
|
||||
port = int(value)
|
||||
if token == "-l" and value:
|
||||
user = value
|
||||
index += 2
|
||||
continue
|
||||
if token.startswith("-p") and token[2:].isdigit(): # forme collee : -p2222
|
||||
port = int(token[2:])
|
||||
index += 1
|
||||
continue
|
||||
if not host:
|
||||
candidate = token
|
||||
if "@" in candidate:
|
||||
user, _, candidate = candidate.partition("@")
|
||||
if HOST_RE.match(candidate):
|
||||
host = candidate
|
||||
index += 1
|
||||
return {"user": user, "host": host, "port": port} if host else None
|
||||
|
||||
|
||||
def detect_ssh_target(text: str) -> dict | None:
|
||||
"""Renvoie la derniere cible ssh trouvee dans le texte d'un terminal."""
|
||||
result = None
|
||||
for match in SSH_COMMAND_RE.finditer(text):
|
||||
parsed = _parse_ssh_args(match.group("args"))
|
||||
if parsed:
|
||||
result = parsed
|
||||
return result
|
||||
|
||||
|
||||
class BaseTerminalTab(QWidget):
|
||||
"""Base commune a un onglet terminal : barre de statut + terminal, autour d'une
|
||||
connexion (SSH ou locale) qui expose la meme interface de signaux
|
||||
(connected / data_received / disconnected / error / send / resize)."""
|
||||
|
||||
def __init__(self, initial_status: str, parent=None):
|
||||
super().__init__(parent)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
self.bar = QHBoxLayout()
|
||||
self.bar.setContentsMargins(10, 6, 10, 6)
|
||||
self.status_label = QLabel(initial_status)
|
||||
self.status_label.setObjectName("statusLabel")
|
||||
self.bar.addWidget(self.status_label, 1)
|
||||
|
||||
# Apparait quand la connexion tombe : les serveurs de test redemarrent souvent,
|
||||
# inutile de fermer l'onglet et de tout ressaisir.
|
||||
self.reconnect_btn = QToolButton()
|
||||
self.reconnect_btn.setObjectName("primary")
|
||||
self.reconnect_btn.setText("⟳ Reconnecter")
|
||||
self.reconnect_btn.hide()
|
||||
self.reconnect_btn.clicked.connect(self.reconnect)
|
||||
self.bar.addWidget(self.reconnect_btn)
|
||||
|
||||
self.log_btn = QToolButton()
|
||||
self.log_btn.setText("⏺ Journal")
|
||||
self.log_btn.setToolTip("Enregistrer la session dans un fichier")
|
||||
self.log_btn.setCheckable(True)
|
||||
self.log_btn.clicked.connect(self.toggle_logging)
|
||||
self.bar.addWidget(self.log_btn)
|
||||
bar_frame = QFrame()
|
||||
bar_frame.setObjectName("tabBar")
|
||||
bar_frame.setLayout(self.bar)
|
||||
layout.addWidget(bar_frame)
|
||||
|
||||
self.splitter = QSplitter(Qt.Horizontal)
|
||||
self.terminal = TerminalWidget()
|
||||
self.splitter.addWidget(self.terminal)
|
||||
layout.addWidget(self.splitter, 1)
|
||||
|
||||
# barre de recherche, masquee tant qu'on n'a pas fait Ctrl+F
|
||||
self.search_bar = QFrame()
|
||||
self.search_bar.setObjectName("tabBar")
|
||||
search_layout = QHBoxLayout(self.search_bar)
|
||||
search_layout.setContentsMargins(10, 4, 10, 4)
|
||||
self.search_input = QLineEdit()
|
||||
self.search_input.setPlaceholderText("Rechercher dans l'historique... (Entree = precedent, Echap = fermer)")
|
||||
self.search_status = QLabel("")
|
||||
self.search_status.setObjectName("statusLabel")
|
||||
close_search = QToolButton()
|
||||
close_search.setText("✕")
|
||||
search_layout.addWidget(self.search_input, 1)
|
||||
search_layout.addWidget(self.search_status)
|
||||
search_layout.addWidget(close_search)
|
||||
self.search_bar.hide()
|
||||
layout.addWidget(self.search_bar)
|
||||
|
||||
self.terminal.search_requested.connect(self.open_search)
|
||||
self.search_input.returnPressed.connect(self._run_search)
|
||||
close_search.clicked.connect(self.close_search)
|
||||
QShortcut(QKeySequence(Qt.Key_Escape), self.search_input, activated=self.close_search)
|
||||
|
||||
self.connection = None # assigne par les sous-classes
|
||||
self._log_file = None
|
||||
|
||||
# ---------- journalisation ----------
|
||||
def toggle_logging(self) -> None:
|
||||
if self._log_file:
|
||||
self.stop_logging()
|
||||
return
|
||||
default = Path.home() / f"session-{datetime.now():%Y%m%d-%H%M%S}.log"
|
||||
path, _ = QFileDialog.getSaveFileName(self, "Journaliser la session", str(default), "Journal (*.log)")
|
||||
if not path:
|
||||
self.log_btn.setChecked(False)
|
||||
return
|
||||
try:
|
||||
self._log_file = open(path, "a", encoding="utf-8", errors="replace")
|
||||
except OSError as exc:
|
||||
QMessageBox.critical(self, "Erreur", f"Impossible d'ecrire le journal : {exc}")
|
||||
self.log_btn.setChecked(False)
|
||||
return
|
||||
self._log_file.write(f"\n===== Session du {datetime.now():%d/%m/%Y %H:%M:%S} =====\n")
|
||||
self.log_btn.setChecked(True)
|
||||
self.log_btn.setToolTip(f"Journalisation active : {path}")
|
||||
|
||||
def stop_logging(self) -> None:
|
||||
if self._log_file:
|
||||
try:
|
||||
self._log_file.close()
|
||||
except OSError:
|
||||
pass
|
||||
self._log_file = None
|
||||
self.log_btn.setChecked(False)
|
||||
self.log_btn.setToolTip("Enregistrer la session dans un fichier")
|
||||
|
||||
def _log_output(self, data: bytes) -> None:
|
||||
if not self._log_file:
|
||||
return
|
||||
# on retire les sequences d'echappement : un journal doit rester lisible
|
||||
text = ANSI_ESCAPE_RE.sub("", data.decode("utf-8", errors="replace"))
|
||||
try:
|
||||
self._log_file.write(text)
|
||||
self._log_file.flush()
|
||||
except OSError:
|
||||
self.stop_logging()
|
||||
|
||||
def _wire_connection(self) -> None:
|
||||
self.connection.connected.connect(self._on_connected)
|
||||
self.connection.data_received.connect(self.terminal.feed)
|
||||
self.connection.data_received.connect(self._log_output)
|
||||
self.connection.disconnected.connect(self._on_disconnected)
|
||||
self.connection.error.connect(self._on_error)
|
||||
self.terminal.data_to_send.connect(self.connection.send)
|
||||
self.terminal.resized.connect(self.connection.resize)
|
||||
|
||||
def _create_connection(self):
|
||||
"""Fabrique la connexion. Redefinie par chaque sous-classe : c'est ce qui
|
||||
permet de la recreer a l'identique lors d'une reconnexion."""
|
||||
raise NotImplementedError
|
||||
|
||||
def start_connection(self) -> None:
|
||||
self.connection = self._create_connection()
|
||||
self._wire_connection()
|
||||
self.connection.start()
|
||||
|
||||
def reconnect(self) -> None:
|
||||
if self.connection:
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(1500)
|
||||
self.terminal.reset_screen()
|
||||
self.reconnect_btn.hide()
|
||||
self.status_label.setText("○ Reconnexion...")
|
||||
self.start_connection()
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.reconnect_btn.hide()
|
||||
self.terminal.setFocus()
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
self.status_label.setText(f"● Deconnecte ({reason})")
|
||||
self.reconnect_btn.show()
|
||||
|
||||
def _on_error(self, message: str) -> None:
|
||||
self.status_label.setText(f"● Erreur : {message}")
|
||||
self.reconnect_btn.show()
|
||||
QMessageBox.critical(self, "Erreur", message)
|
||||
|
||||
# ---------- recherche dans l'historique ----------
|
||||
def open_search(self) -> None:
|
||||
self.search_bar.show()
|
||||
self.search_input.setFocus()
|
||||
self.search_input.selectAll()
|
||||
|
||||
def close_search(self) -> None:
|
||||
self.search_bar.hide()
|
||||
self.search_status.clear()
|
||||
self.terminal.setFocus()
|
||||
|
||||
def _run_search(self) -> None:
|
||||
needle = self.search_input.text()
|
||||
if not needle:
|
||||
return
|
||||
found = self.terminal.find_text(needle)
|
||||
self.search_status.setText("" if found else "aucun resultat")
|
||||
|
||||
def close_connection(self) -> None:
|
||||
self.stop_logging()
|
||||
if self.connection:
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(2000)
|
||||
|
||||
|
||||
class SessionTabWidget(BaseTerminalTab):
|
||||
"""Un onglet = une connexion SSH avec son terminal et un panneau SFTP optionnel."""
|
||||
|
||||
def __init__(self, session: Session, secret: str, parent=None):
|
||||
super().__init__(f"○ Connexion a {session.username}@{session.host}...", parent)
|
||||
self.session = session
|
||||
self.sftp_panel: SFTPPanel | None = None
|
||||
|
||||
self.sftp_toggle = QToolButton()
|
||||
self.sftp_toggle.setText("📁 Fichiers (SFTP)")
|
||||
self.sftp_toggle.setToolTip("Afficher / masquer le navigateur de fichiers SFTP")
|
||||
self.sftp_toggle.setCheckable(True)
|
||||
self.sftp_toggle.setChecked(True)
|
||||
self.sftp_toggle.setEnabled(False)
|
||||
self.bar.addWidget(self.sftp_toggle)
|
||||
|
||||
self.tunnel_btn = QToolButton()
|
||||
self.tunnel_btn.setText("🔀 Tunnels")
|
||||
self.tunnel_btn.setToolTip("Redirections de ports SSH (-L / -R)")
|
||||
self.tunnel_btn.setEnabled(False)
|
||||
self.tunnel_btn.clicked.connect(self.open_tunnels)
|
||||
self.bar.addWidget(self.tunnel_btn)
|
||||
|
||||
self.tunnel_manager: TunnelManager | None = None
|
||||
|
||||
self._secret = secret
|
||||
self.sftp_toggle.toggled.connect(self._toggle_sftp)
|
||||
self.start_connection()
|
||||
|
||||
def _create_connection(self):
|
||||
session = self.session
|
||||
connection = SSHConnection(
|
||||
host=session.host,
|
||||
port=session.port,
|
||||
username=session.username,
|
||||
auth_type=session.auth_type,
|
||||
key_path=session.key_path,
|
||||
password=self._secret if session.auth_type == AUTH_PASSWORD else "",
|
||||
passphrase=self._secret if session.auth_type == AUTH_KEY else "",
|
||||
# comme MobaXterm : le navigateur de fichiers est la des l'ouverture,
|
||||
# pas cache derriere une action supplementaire
|
||||
open_sftp_on_connect=True,
|
||||
)
|
||||
connection.sftp_ready.connect(self._on_sftp_ready)
|
||||
connection.password_prompt.connect(self._on_password_prompt)
|
||||
connection.host_key_prompt.connect(self._on_host_key_prompt)
|
||||
connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt)
|
||||
connection.passphrase_prompt.connect(self._on_passphrase_prompt)
|
||||
return connection
|
||||
|
||||
def open_tunnels(self) -> None:
|
||||
if not self.tunnel_manager:
|
||||
QMessageBox.information(self, "Tunnels", "La session doit etre connectee.")
|
||||
return
|
||||
TunnelDialog(self.tunnel_manager, parent=self).exec()
|
||||
actifs = len(self.tunnel_manager.tunnels)
|
||||
self.tunnel_btn.setText(f"🔀 Tunnels ({actifs})" if actifs else "🔀 Tunnels")
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.status_label.setText(f"● Connecte : {self.session.username}@{self.session.host}:{self.session.port}")
|
||||
transport = self.connection.client.get_transport() if self.connection.client else None
|
||||
if transport:
|
||||
self.tunnel_manager = TunnelManager(transport)
|
||||
self.tunnel_btn.setEnabled(True)
|
||||
# le bouton n'est active qu'une fois le canal SFTP reellement ouvert
|
||||
# (_on_sftp_ready) : certains serveurs n'exposent pas le sous-systeme SFTP
|
||||
super()._on_connected()
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
super()._on_disconnected(reason)
|
||||
self.sftp_toggle.setEnabled(False)
|
||||
|
||||
def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None:
|
||||
accepted = confirm_host_key(self, hostname, key_type, fingerprint)
|
||||
self.connection.confirm_host_key(accepted)
|
||||
|
||||
def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None:
|
||||
accepted = confirm_host_key_change(self, hostname, key_type, old_fp, new_fp)
|
||||
self.connection.confirm_host_key_change(accepted)
|
||||
|
||||
def _on_passphrase_prompt(self, key_path: str) -> None:
|
||||
text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}")
|
||||
self.connection.provide_passphrase(text if ok else None)
|
||||
|
||||
def _on_password_prompt(self, target: str) -> None:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :")
|
||||
self.connection.provide_password(text if ok else None)
|
||||
|
||||
def _on_sftp_ready(self, sftp) -> None:
|
||||
"""Le canal SFTP a ete ouvert en arriere-plan : on affiche le navigateur de
|
||||
fichiers a cote du terminal, sans action de l'utilisateur."""
|
||||
if self.sftp_panel is not None:
|
||||
return
|
||||
self.sftp_panel = SFTPPanel(sftp)
|
||||
self.splitter.insertWidget(0, self.sftp_panel)
|
||||
self.splitter.setSizes([460, 540])
|
||||
self.sftp_toggle.setEnabled(True)
|
||||
self.sftp_toggle.setChecked(True)
|
||||
self.terminal.setFocus()
|
||||
|
||||
def close_connection(self) -> None:
|
||||
if self.tunnel_manager:
|
||||
self.tunnel_manager.stop_all()
|
||||
self.tunnel_manager = None
|
||||
super().close_connection()
|
||||
|
||||
def _toggle_sftp(self, checked: bool) -> None:
|
||||
if self.sftp_panel is not None:
|
||||
self.sftp_panel.setVisible(checked)
|
||||
|
||||
|
||||
class SFTPTabWidget(QWidget):
|
||||
"""Onglet SFTP autonome (facon WinSCP) : double panneau de fichiers, sans terminal."""
|
||||
|
||||
def __init__(self, session: Session, secret: str, parent=None):
|
||||
super().__init__(parent)
|
||||
self.session = session
|
||||
self.sftp_panel: SFTPPanel | None = None
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
bar = QHBoxLayout()
|
||||
bar.setContentsMargins(10, 6, 10, 6)
|
||||
self.status_label = QLabel(f"○ Connexion SFTP a {session.username}@{session.host}...")
|
||||
self.status_label.setObjectName("statusLabel")
|
||||
bar.addWidget(self.status_label, 1)
|
||||
bar_frame = QFrame()
|
||||
bar_frame.setObjectName("tabBar")
|
||||
bar_frame.setLayout(bar)
|
||||
layout.addWidget(bar_frame)
|
||||
|
||||
self.container = QVBoxLayout()
|
||||
holder = QWidget()
|
||||
holder.setLayout(self.container)
|
||||
layout.addWidget(holder, 1)
|
||||
|
||||
self.connection = SSHConnection(
|
||||
host=session.host,
|
||||
port=session.port,
|
||||
username=session.username,
|
||||
auth_type=session.auth_type,
|
||||
key_path=session.key_path,
|
||||
password=secret if session.auth_type == AUTH_PASSWORD else "",
|
||||
passphrase=secret if session.auth_type == AUTH_KEY else "",
|
||||
open_shell=False,
|
||||
open_sftp_on_connect=True,
|
||||
)
|
||||
self.connection.sftp_ready.connect(self._on_sftp_ready)
|
||||
self.connection.password_prompt.connect(self._on_password_prompt)
|
||||
self.connection.error.connect(self._on_error)
|
||||
self.connection.disconnected.connect(self._on_disconnected)
|
||||
self.connection.host_key_prompt.connect(self._on_host_key_prompt)
|
||||
self.connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt)
|
||||
self.connection.passphrase_prompt.connect(self._on_passphrase_prompt)
|
||||
self.connection.start()
|
||||
|
||||
def _on_sftp_ready(self, sftp) -> None:
|
||||
self.sftp_panel = SFTPPanel(sftp)
|
||||
self.container.addWidget(self.sftp_panel)
|
||||
self.status_label.setText(
|
||||
f"● SFTP connecte : {self.session.username}@{self.session.host}:{self.session.port}"
|
||||
)
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
self.status_label.setText(f"● Deconnecte ({reason})")
|
||||
|
||||
def _on_error(self, message: str) -> None:
|
||||
self.status_label.setText(f"● Erreur : {message}")
|
||||
QMessageBox.critical(self, "Erreur SFTP", message)
|
||||
|
||||
def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None:
|
||||
self.connection.confirm_host_key(confirm_host_key(self, hostname, key_type, fingerprint))
|
||||
|
||||
def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None:
|
||||
self.connection.confirm_host_key_change(
|
||||
confirm_host_key_change(self, hostname, key_type, old_fp, new_fp)
|
||||
)
|
||||
|
||||
def _on_passphrase_prompt(self, key_path: str) -> None:
|
||||
text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}")
|
||||
self.connection.provide_passphrase(text if ok else None)
|
||||
|
||||
def _on_password_prompt(self, target: str) -> None:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :")
|
||||
self.connection.provide_password(text if ok else None)
|
||||
|
||||
def close_connection(self) -> None:
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(2000)
|
||||
|
||||
|
||||
class LocalTerminalTabWidget(BaseTerminalTab):
|
||||
"""Un onglet = un terminal Windows local (PowerShell / cmd), sans session enregistree."""
|
||||
|
||||
sftp_requested = Signal(object) # cible ssh detectee (ou None)
|
||||
|
||||
def __init__(self, shell: str = SHELL_POWERSHELL, parent=None):
|
||||
super().__init__(f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...", parent)
|
||||
self.shell = shell
|
||||
|
||||
# Meme dans un terminal local, on tape souvent "ssh user@ip" a la main :
|
||||
# ce bouton ouvre le SFTP sur cet hote, sans passer par un favori.
|
||||
self.sftp_btn = QToolButton()
|
||||
self.sftp_btn.setText("📁 SFTP sur cette connexion")
|
||||
self.sftp_btn.setToolTip(
|
||||
"Ouvre un gestionnaire de fichiers SFTP sur l'hote de la commande ssh tapee ici"
|
||||
)
|
||||
self.sftp_btn.clicked.connect(
|
||||
lambda: self.sftp_requested.emit(detect_ssh_target(self.terminal.buffer_text()))
|
||||
)
|
||||
self.bar.addWidget(self.sftp_btn)
|
||||
|
||||
self.start_connection()
|
||||
|
||||
def _create_connection(self):
|
||||
return LocalTerminalConnection(shell=self.shell)
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.status_label.setText(f"● Terminal local — {SHELL_LABELS.get(self.shell, self.shell)}")
|
||||
super()._on_connected()
|
||||
# re-exporte pour compatibilite (utilise par tests/test_detect.py et du code existant)
|
||||
from app.services.ssh_target import detect_ssh_target # noqa: F401
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
@@ -498,34 +36,61 @@ class MainWindow(QMainWindow):
|
||||
self.session_store = SessionStore()
|
||||
self.key_manager = KeyManager()
|
||||
|
||||
self.broadcast = BroadcastManager(self._iter_tabs, self)
|
||||
self.connection_controller = ConnectionController(
|
||||
self.session_store, self.key_manager, self.broadcast, parent_widget=self,
|
||||
)
|
||||
self.session_controller = SessionController(
|
||||
self.session_store, self.key_manager, self.broadcast.button, parent_widget=self,
|
||||
)
|
||||
self._wire_controllers()
|
||||
|
||||
self._build_menu()
|
||||
self._build_sidebar()
|
||||
self._build_tabs()
|
||||
self.setCentralWidget(self.connection_controller.tabs)
|
||||
self.setStatusBar(QStatusBar())
|
||||
|
||||
self.reload_sessions()
|
||||
|
||||
# demarre directement dans un terminal utilisable, sans avoir a en choisir un
|
||||
self.open_local_terminal(DEFAULT_SHELL)
|
||||
self.connection_controller.open_local_terminal(DEFAULT_SHELL)
|
||||
|
||||
# ---------- acces retro-compatible (tests, code externe) ----------
|
||||
@property
|
||||
def tabs(self):
|
||||
return self.connection_controller.tabs
|
||||
|
||||
def _iter_tabs(self):
|
||||
tabs = self.connection_controller.tabs
|
||||
return [tabs.widget(i) for i in range(tabs.count())]
|
||||
|
||||
# ---------- cablage entre controleurs ----------
|
||||
def _wire_controllers(self) -> None:
|
||||
sc, cc = self.session_controller, self.connection_controller
|
||||
sc.open_session_requested.connect(cc.open_session)
|
||||
sc.open_sftp_requested.connect(cc.open_sftp_session)
|
||||
sc.open_local_shell_requested.connect(cc.open_local_terminal)
|
||||
sc.quick_sftp_requested.connect(cc.quick_sftp)
|
||||
sc.manage_keys_requested.connect(self.open_key_manager)
|
||||
sc.manage_known_hosts_requested.connect(self.open_known_hosts_manager)
|
||||
cc.session_saved.connect(sc.reload)
|
||||
|
||||
# ---------- construction UI ----------
|
||||
def _build_menu(self) -> None:
|
||||
file_menu = self.menuBar().addMenu("&Fichier")
|
||||
new_session_action = QAction("Nouvelle session...", self)
|
||||
new_session_action.setShortcut(QKeySequence("Ctrl+N"))
|
||||
new_session_action.triggered.connect(self.new_session)
|
||||
new_session_action.triggered.connect(self.session_controller.new_session)
|
||||
file_menu.addAction(new_session_action)
|
||||
|
||||
file_menu.addSeparator()
|
||||
|
||||
quick_ssh_action = QAction("Connexion SSH rapide...", self)
|
||||
quick_ssh_action.setShortcut(QKeySequence("Ctrl+Shift+S"))
|
||||
quick_ssh_action.triggered.connect(self.quick_ssh)
|
||||
quick_ssh_action.triggered.connect(self.connection_controller.quick_ssh)
|
||||
file_menu.addAction(quick_ssh_action)
|
||||
|
||||
quick_sftp_action = QAction("Connexion SFTP (gestionnaire de fichiers)...", self)
|
||||
quick_sftp_action.setShortcut(QKeySequence("Ctrl+Shift+F"))
|
||||
quick_sftp_action.triggered.connect(self.quick_sftp)
|
||||
quick_sftp_action.triggered.connect(self.connection_controller.quick_sftp)
|
||||
file_menu.addAction(quick_sftp_action)
|
||||
|
||||
file_menu.addSeparator()
|
||||
@@ -534,12 +99,12 @@ class MainWindow(QMainWindow):
|
||||
f"Nouveau terminal local ({SHELL_LABELS.get(DEFAULT_SHELL, DEFAULT_SHELL)})", self
|
||||
)
|
||||
default_action.setShortcut(QKeySequence("Ctrl+Shift+T"))
|
||||
default_action.triggered.connect(lambda: self.open_local_terminal(DEFAULT_SHELL))
|
||||
default_action.triggered.connect(lambda: self.connection_controller.open_local_terminal(DEFAULT_SHELL))
|
||||
file_menu.addAction(default_action)
|
||||
|
||||
others_menu = file_menu.addMenu("Autre terminal local")
|
||||
for shell_id, label in SHELL_LABELS.items():
|
||||
others_menu.addAction(label, lambda s=shell_id: self.open_local_terminal(s))
|
||||
others_menu.addAction(label, lambda s=shell_id: self.connection_controller.open_local_terminal(s))
|
||||
|
||||
file_menu.addSeparator()
|
||||
quit_action = QAction("Quitter", self)
|
||||
@@ -591,357 +156,13 @@ class MainWindow(QMainWindow):
|
||||
)
|
||||
self.sessions_dock = dock
|
||||
|
||||
container = QWidget()
|
||||
layout = QVBoxLayout(container)
|
||||
layout.setContentsMargins(6, 6, 6, 6)
|
||||
layout.setSpacing(6)
|
||||
|
||||
toolbar = QHBoxLayout()
|
||||
new_btn = QToolButton()
|
||||
new_btn.setObjectName("primary")
|
||||
new_btn.setText("+ Session")
|
||||
new_btn.clicked.connect(self.new_session)
|
||||
toolbar.addWidget(new_btn)
|
||||
|
||||
term_btn = QToolButton()
|
||||
term_btn.setText("💻 Terminal")
|
||||
term_btn.setPopupMode(QToolButton.InstantPopup)
|
||||
term_menu = QMenu(term_btn)
|
||||
for shell_id, label in SHELL_LABELS.items():
|
||||
term_menu.addAction(label, lambda s=shell_id: self.open_local_terminal(s))
|
||||
term_btn.setMenu(term_menu)
|
||||
toolbar.addWidget(term_btn)
|
||||
|
||||
self.broadcast_btn = QToolButton()
|
||||
self.broadcast_btn.setText("⇉ Multi")
|
||||
self.broadcast_btn.setCheckable(True)
|
||||
self.broadcast_btn.setToolTip(
|
||||
"Envoi simultane : ce que vous tapez part vers tous les onglets terminal ouverts"
|
||||
)
|
||||
toolbar.addWidget(self.broadcast_btn)
|
||||
|
||||
sftp_btn = QToolButton()
|
||||
sftp_btn.setText("📁 SFTP")
|
||||
sftp_btn.setToolTip("Ouvrir un gestionnaire de fichiers SFTP (sans session enregistree)")
|
||||
sftp_btn.clicked.connect(self.quick_sftp)
|
||||
toolbar.addWidget(sftp_btn)
|
||||
|
||||
toolbar.addStretch()
|
||||
|
||||
manage_btn = QToolButton()
|
||||
manage_btn.setText("⚙")
|
||||
manage_btn.setToolTip("Cles SSH et hotes connus")
|
||||
manage_btn.setPopupMode(QToolButton.InstantPopup)
|
||||
manage_menu = QMenu(manage_btn)
|
||||
manage_menu.addAction("Gestionnaire de cles SSH...", self.open_key_manager)
|
||||
manage_menu.addAction("Gestionnaire des hotes connus...", self.open_known_hosts_manager)
|
||||
manage_btn.setMenu(manage_menu)
|
||||
toolbar.addWidget(manage_btn)
|
||||
|
||||
layout.addLayout(toolbar)
|
||||
|
||||
self.tree = QTreeWidget()
|
||||
self.tree.setHeaderHidden(True)
|
||||
self.tree.itemDoubleClicked.connect(self._on_tree_double_click)
|
||||
self.tree.setContextMenuPolicy(Qt.CustomContextMenu)
|
||||
self.tree.customContextMenuRequested.connect(self._on_tree_context_menu)
|
||||
layout.addWidget(self.tree)
|
||||
|
||||
# Sans minimum explicite, la largeur plancher est imposee par le contenu
|
||||
# (les noms de sessions), ce qui empeche de retrecir vraiment le panneau.
|
||||
container.setMinimumWidth(120)
|
||||
self.tree.setMinimumWidth(0)
|
||||
dock.setWidget(container)
|
||||
dock.setWidget(self.session_controller.widget)
|
||||
self.addDockWidget(Qt.LeftDockWidgetArea, dock)
|
||||
self.resizeDocks([dock], [300], Qt.Horizontal)
|
||||
# garde la case du menu Affichage synchronisee si on ferme le panneau par sa croix
|
||||
dock.visibilityChanged.connect(self.toggle_sidebar_action.setChecked)
|
||||
|
||||
def _build_tabs(self) -> None:
|
||||
self.tabs = QTabWidget()
|
||||
self.tabs.setTabsClosable(True)
|
||||
self.tabs.setMovable(True)
|
||||
# double-clic sur un onglet pour le renommer, clic droit pour le menu
|
||||
self.tabs.tabBarDoubleClicked.connect(self.rename_tab)
|
||||
self.tabs.tabBar().setContextMenuPolicy(Qt.CustomContextMenu)
|
||||
self.tabs.tabBar().customContextMenuRequested.connect(self._on_tab_context_menu)
|
||||
self.tabs.tabCloseRequested.connect(self._on_tab_close)
|
||||
self.setCentralWidget(self.tabs)
|
||||
|
||||
# ---------- sessions / arbre ----------
|
||||
def reload_sessions(self) -> None:
|
||||
self.tree.clear()
|
||||
|
||||
locals_header = QTreeWidgetItem(["💻 Terminaux locaux"])
|
||||
locals_header.setFlags(locals_header.flags() & ~Qt.ItemIsSelectable)
|
||||
font = locals_header.font(0)
|
||||
font.setBold(True)
|
||||
locals_header.setFont(0, font)
|
||||
self.tree.addTopLevelItem(locals_header)
|
||||
for shell, label in SHELL_LABELS.items():
|
||||
child = QTreeWidgetItem([f" {label}"])
|
||||
child.setData(0, LOCAL_SHELL_ROLE, shell)
|
||||
locals_header.addChild(child)
|
||||
|
||||
sessions_header = QTreeWidgetItem(["🔐 Sessions SSH"])
|
||||
sessions_header.setFlags(sessions_header.flags() & ~Qt.ItemIsSelectable)
|
||||
sessions_header.setFont(0, font)
|
||||
self.tree.addTopLevelItem(sessions_header)
|
||||
|
||||
folder_items: dict[str, QTreeWidgetItem] = {}
|
||||
|
||||
def get_folder(path: str) -> QTreeWidgetItem:
|
||||
if not path:
|
||||
return sessions_header
|
||||
if path in folder_items:
|
||||
return folder_items[path]
|
||||
parent_path, _, name = path.rpartition("/")
|
||||
parent_item = get_folder(parent_path)
|
||||
item = QTreeWidgetItem([f"📁 {name}"])
|
||||
item.setFlags(item.flags() & ~Qt.ItemIsSelectable)
|
||||
parent_item.addChild(item)
|
||||
folder_items[path] = item
|
||||
return item
|
||||
|
||||
for session in sorted(self.session_store.sessions, key=lambda s: s.name.lower()):
|
||||
parent = get_folder(session.group)
|
||||
item = QTreeWidgetItem([f"🖥 {session.name} ({session.host})"])
|
||||
item.setData(0, SESSION_ROLE, session.id)
|
||||
parent.addChild(item)
|
||||
|
||||
if not self.session_store.sessions:
|
||||
hint = QTreeWidgetItem([" (aucune session — bouton « + Session » ou « 📁 SFTP »)"])
|
||||
hint.setFlags(hint.flags() & ~Qt.ItemIsSelectable)
|
||||
hint.setForeground(0, Qt.gray)
|
||||
sessions_header.addChild(hint)
|
||||
|
||||
self.tree.expandAll()
|
||||
|
||||
def _session_from_item(self, item: QTreeWidgetItem) -> Session | None:
|
||||
if not item:
|
||||
return None
|
||||
session_id = item.data(0, SESSION_ROLE)
|
||||
if not session_id:
|
||||
return None
|
||||
return next((s for s in self.session_store.sessions if s.id == session_id), None)
|
||||
|
||||
def _shell_from_item(self, item: QTreeWidgetItem) -> str | None:
|
||||
if not item:
|
||||
return None
|
||||
return item.data(0, LOCAL_SHELL_ROLE)
|
||||
|
||||
def _on_tree_double_click(self, item: QTreeWidgetItem, _col: int) -> None:
|
||||
session = self._session_from_item(item)
|
||||
if session:
|
||||
self.open_session(session)
|
||||
return
|
||||
shell = self._shell_from_item(item)
|
||||
if shell:
|
||||
self.open_local_terminal(shell)
|
||||
|
||||
def _on_tree_context_menu(self, pos) -> None:
|
||||
item = self.tree.itemAt(pos)
|
||||
menu = QMenu(self)
|
||||
session = self._session_from_item(item) if item else None
|
||||
shell = self._shell_from_item(item) if item else None
|
||||
|
||||
if session:
|
||||
connect_action = menu.addAction("🖥 Ouvrir un terminal SSH")
|
||||
sftp_action = menu.addAction("📁 Ouvrir en SFTP (gestionnaire de fichiers)")
|
||||
menu.addSeparator()
|
||||
edit_action = menu.addAction("Modifier...")
|
||||
duplicate_action = menu.addAction("Dupliquer")
|
||||
delete_action = menu.addAction("Supprimer")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == connect_action:
|
||||
self.open_session(session)
|
||||
elif action == sftp_action:
|
||||
self.open_sftp_session(session)
|
||||
elif action == edit_action:
|
||||
self.edit_session(session)
|
||||
elif action == duplicate_action:
|
||||
self.duplicate_session(session)
|
||||
elif action == delete_action:
|
||||
self.delete_session(session)
|
||||
elif shell:
|
||||
open_action = menu.addAction("Ouvrir")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == open_action:
|
||||
self.open_local_terminal(shell)
|
||||
else:
|
||||
new_action = menu.addAction("Nouvelle session...")
|
||||
action = menu.exec(self.tree.viewport().mapToGlobal(pos))
|
||||
if action == new_action:
|
||||
self.new_session()
|
||||
|
||||
def new_session(self) -> None:
|
||||
dialog = SessionDialog(self.key_manager, parent=self)
|
||||
if dialog.exec():
|
||||
self.session_store.add(dialog.result_session())
|
||||
self.reload_sessions()
|
||||
|
||||
def edit_session(self, session: Session) -> None:
|
||||
dialog = SessionDialog(self.key_manager, session=session, parent=self)
|
||||
if dialog.exec():
|
||||
self.session_store.update(dialog.result_session())
|
||||
self.reload_sessions()
|
||||
|
||||
def duplicate_session(self, session: Session) -> None:
|
||||
new_session = copy.deepcopy(session)
|
||||
new_session.id = str(uuid.uuid4())
|
||||
new_session.name = f"{session.name} (copie)"
|
||||
self.session_store.add(new_session)
|
||||
self.reload_sessions()
|
||||
|
||||
def delete_session(self, session: Session) -> None:
|
||||
if QMessageBox.question(self, "Supprimer", f"Supprimer la session '{session.name}' ?") == QMessageBox.Yes:
|
||||
self.session_store.delete(session.id)
|
||||
self.reload_sessions()
|
||||
|
||||
# ---------- connexion / onglets ----------
|
||||
def open_session(self, session: Session) -> None:
|
||||
secret = ""
|
||||
if session.auth_type == AUTH_PASSWORD:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :")
|
||||
if not ok:
|
||||
return
|
||||
secret = text
|
||||
|
||||
tab = SessionTabWidget(session, secret)
|
||||
self._register_broadcast(tab)
|
||||
index = self.tabs.addTab(tab, f"🖥 {session.name}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def quick_sftp(self, target: dict | None = None) -> None:
|
||||
"""Ouvre un SFTP en saisissant directement les infos de connexion, sans favori.
|
||||
|
||||
Si l'onglet courant est un terminal ou une commande ssh a ete tapee, l'hote,
|
||||
l'utilisateur et le port sont pre-remplis automatiquement."""
|
||||
if target is None:
|
||||
target = self._current_tab_ssh_target()
|
||||
self._quick_connect(sftp=True, target=target)
|
||||
|
||||
def quick_ssh(self) -> None:
|
||||
self._quick_connect(sftp=False, target=self._current_tab_ssh_target())
|
||||
|
||||
def _current_tab_ssh_target(self) -> dict | None:
|
||||
widget = self.tabs.currentWidget()
|
||||
if isinstance(widget, SessionTabWidget):
|
||||
return {
|
||||
"user": widget.session.username,
|
||||
"host": widget.session.host,
|
||||
"port": widget.session.port,
|
||||
}
|
||||
if isinstance(widget, BaseTerminalTab):
|
||||
return detect_ssh_target(widget.terminal.buffer_text())
|
||||
return None
|
||||
|
||||
def _quick_connect(self, sftp: bool, target: dict | None = None) -> None:
|
||||
prefill = None
|
||||
if target:
|
||||
prefill = Session(
|
||||
name=target.get("host", ""),
|
||||
host=target.get("host", ""),
|
||||
username=target.get("user", ""),
|
||||
port=target.get("port", 22),
|
||||
)
|
||||
dialog = SessionDialog(
|
||||
self.key_manager,
|
||||
session=prefill,
|
||||
parent=self,
|
||||
connect_label="Se connecter",
|
||||
title="Connexion SFTP" if sftp else "Connexion SSH",
|
||||
)
|
||||
if not dialog.exec():
|
||||
return
|
||||
session = dialog.result_session()
|
||||
if dialog.should_save():
|
||||
self.session_store.add(session)
|
||||
self.reload_sessions()
|
||||
if sftp:
|
||||
self.open_sftp_session(session)
|
||||
else:
|
||||
self.open_session(session)
|
||||
|
||||
def open_sftp_session(self, session: Session) -> None:
|
||||
"""Ouvre un onglet SFTP autonome (double panneau) pour cette session."""
|
||||
secret = ""
|
||||
if session.auth_type == AUTH_PASSWORD:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :")
|
||||
if not ok:
|
||||
return
|
||||
secret = text
|
||||
|
||||
tab = SFTPTabWidget(session, secret)
|
||||
index = self.tabs.addTab(tab, f"📁 {session.name}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def open_local_terminal(self, shell: str = DEFAULT_SHELL) -> None:
|
||||
tab = LocalTerminalTabWidget(shell)
|
||||
tab.sftp_requested.connect(self.quick_sftp)
|
||||
self._register_broadcast(tab)
|
||||
label = SHELL_LABELS.get(shell, shell)
|
||||
index = self.tabs.addTab(tab, f"💻 {label}")
|
||||
self.tabs.setCurrentIndex(index)
|
||||
|
||||
def _broadcast(self, data: bytes, source) -> None:
|
||||
"""Envoie la meme frappe aux autres onglets terminal quand le mode Multi est actif."""
|
||||
if not self.broadcast_btn.isChecked():
|
||||
return
|
||||
for index in range(self.tabs.count()):
|
||||
tab = self.tabs.widget(index)
|
||||
if tab is source or not isinstance(tab, BaseTerminalTab):
|
||||
continue
|
||||
if tab.connection:
|
||||
# on ecrit directement dans la connexion : passer par le terminal
|
||||
# relancerait la diffusion et boucherait a l'infini
|
||||
tab.connection.send(data)
|
||||
|
||||
def _register_broadcast(self, tab) -> None:
|
||||
tab.terminal.data_to_send.connect(lambda data, t=tab: self._broadcast(data, t))
|
||||
|
||||
# ---------- renommage des onglets ----------
|
||||
def rename_tab(self, index: int) -> None:
|
||||
if index < 0:
|
||||
return
|
||||
actuel = self.tabs.tabText(index)
|
||||
nouveau, ok = QInputDialog.getText(
|
||||
self, "Renommer l'onglet", "Nom de l'onglet :", text=actuel
|
||||
)
|
||||
if ok and nouveau.strip():
|
||||
self.tabs.setTabText(index, nouveau.strip())
|
||||
|
||||
def _on_tab_context_menu(self, pos) -> None:
|
||||
bar = self.tabs.tabBar()
|
||||
index = bar.tabAt(pos)
|
||||
if index < 0:
|
||||
return
|
||||
menu = QMenu(self)
|
||||
act_rename = menu.addAction("Renommer... (double-clic)")
|
||||
menu.addSeparator()
|
||||
act_close = menu.addAction("Fermer cet onglet")
|
||||
act_close_others = menu.addAction("Fermer les autres onglets")
|
||||
act_close_others.setEnabled(self.tabs.count() > 1)
|
||||
|
||||
action = menu.exec(bar.mapToGlobal(pos))
|
||||
if action == act_rename:
|
||||
self.rename_tab(index)
|
||||
elif action == act_close:
|
||||
self._on_tab_close(index)
|
||||
elif action == act_close_others:
|
||||
# on ferme par la fin : retirer un onglet decale les suivants
|
||||
cible = self.tabs.widget(index)
|
||||
for other in range(self.tabs.count() - 1, -1, -1):
|
||||
if self.tabs.widget(other) is not cible:
|
||||
self._on_tab_close(other)
|
||||
|
||||
def _on_tab_close(self, index: int) -> None:
|
||||
widget = self.tabs.widget(index)
|
||||
if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)):
|
||||
widget.close_connection()
|
||||
self.tabs.removeTab(index)
|
||||
|
||||
# ---------- cles / a propos ----------
|
||||
# ---------- cles / affichage / a propos ----------
|
||||
def toggle_sidebar(self, visible: bool | None = None) -> None:
|
||||
"""Affiche ou masque le panneau des sessions (F9)."""
|
||||
if visible is None:
|
||||
@@ -976,8 +197,5 @@ class MainWindow(QMainWindow):
|
||||
)
|
||||
|
||||
def closeEvent(self, event) -> None:
|
||||
for i in range(self.tabs.count()):
|
||||
widget = self.tabs.widget(i)
|
||||
if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)):
|
||||
widget.close_connection()
|
||||
self.connection_controller.close_all()
|
||||
super().closeEvent(event)
|
||||
|
||||
0
app/services/__init__.py
Normal file
0
app/services/__init__.py
Normal file
55
app/services/ssh_target.py
Normal file
55
app/services/ssh_target.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""Detection d'une cible ssh (user/host/port) tapee a la main dans un terminal.
|
||||
|
||||
Sert a proposer le SFTP sur l'hote d'une commande `ssh ...` sans avoir a creer
|
||||
un favori au prealable.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# Repere une commande ssh tapee a la main dans un terminal.
|
||||
SSH_COMMAND_RE = re.compile(r"\bssh\s+(?P<args>[^\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
|
||||
0
app/ui/__init__.py
Normal file
0
app/ui/__init__.py
Normal file
450
app/ui/tabs.py
Normal file
450
app/ui/tabs.py
Normal file
@@ -0,0 +1,450 @@
|
||||
"""Widgets d'onglet : terminal SSH, terminal local, SFTP autonome.
|
||||
|
||||
Chaque onglet encapsule une connexion (SSH ou locale) qui expose la meme
|
||||
interface de signaux (connected / data_received / disconnected / error / send /
|
||||
resize), ce qui permet a `BaseTerminalTab` de rester independante du type de
|
||||
connexion sous-jacent.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import Qt, Signal
|
||||
from PySide6.QtGui import QKeySequence, QShortcut
|
||||
from PySide6.QtWidgets import (
|
||||
QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMessageBox, QSplitter,
|
||||
QFrame, QLineEdit, QFileDialog,
|
||||
)
|
||||
|
||||
from app.session_manager import Session, AUTH_KEY, AUTH_PASSWORD
|
||||
from app.ssh_connection import SSHConnection
|
||||
from app.local_terminal import LocalTerminalConnection, SHELL_POWERSHELL, SHELL_LABELS
|
||||
from app.terminal_widget import TerminalWidget
|
||||
from app.sftp_panel import SFTPPanel
|
||||
from app.dialogs.tunnel_dialog import TunnelDialog
|
||||
from app.tunnels import TunnelManager
|
||||
from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_host_key_change
|
||||
from app.services.ssh_target import detect_ssh_target
|
||||
|
||||
# sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles
|
||||
ANSI_ESCAPE_RE = re.compile(
|
||||
r"\x1b\[[0-9;?]*[a-zA-Z]" # CSI : couleurs, deplacements du curseur
|
||||
r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC : titre de fenetre
|
||||
r"|\x1b[()][A-Z0-9]" # jeux de caracteres
|
||||
r"|\x1b[=>]" # mode clavier
|
||||
r"|[\x00-\x08\x0b\x0c\x0e-\x1f]" # autres caracteres de controle
|
||||
)
|
||||
|
||||
|
||||
class BaseTerminalTab(QWidget):
|
||||
"""Base commune a un onglet terminal : barre de statut + terminal, autour d'une
|
||||
connexion (SSH ou locale) qui expose la meme interface de signaux
|
||||
(connected / data_received / disconnected / error / send / resize)."""
|
||||
|
||||
def __init__(self, initial_status: str, parent=None):
|
||||
super().__init__(parent)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
self.bar = QHBoxLayout()
|
||||
self.bar.setContentsMargins(10, 6, 10, 6)
|
||||
self.status_label = QLabel(initial_status)
|
||||
self.status_label.setObjectName("statusLabel")
|
||||
self.bar.addWidget(self.status_label, 1)
|
||||
|
||||
# Apparait quand la connexion tombe : les serveurs de test redemarrent souvent,
|
||||
# inutile de fermer l'onglet et de tout ressaisir.
|
||||
self.reconnect_btn = QToolButton()
|
||||
self.reconnect_btn.setObjectName("primary")
|
||||
self.reconnect_btn.setText("⟳ Reconnecter")
|
||||
self.reconnect_btn.hide()
|
||||
self.reconnect_btn.clicked.connect(self.reconnect)
|
||||
self.bar.addWidget(self.reconnect_btn)
|
||||
|
||||
self.log_btn = QToolButton()
|
||||
self.log_btn.setText("⏺ Journal")
|
||||
self.log_btn.setToolTip("Enregistrer la session dans un fichier")
|
||||
self.log_btn.setCheckable(True)
|
||||
self.log_btn.clicked.connect(self.toggle_logging)
|
||||
self.bar.addWidget(self.log_btn)
|
||||
bar_frame = QFrame()
|
||||
bar_frame.setObjectName("tabBar")
|
||||
bar_frame.setLayout(self.bar)
|
||||
layout.addWidget(bar_frame)
|
||||
|
||||
self.splitter = QSplitter(Qt.Horizontal)
|
||||
self.terminal = TerminalWidget()
|
||||
self.splitter.addWidget(self.terminal)
|
||||
layout.addWidget(self.splitter, 1)
|
||||
|
||||
# barre de recherche, masquee tant qu'on n'a pas fait Ctrl+F
|
||||
self.search_bar = QFrame()
|
||||
self.search_bar.setObjectName("tabBar")
|
||||
search_layout = QHBoxLayout(self.search_bar)
|
||||
search_layout.setContentsMargins(10, 4, 10, 4)
|
||||
self.search_input = QLineEdit()
|
||||
self.search_input.setPlaceholderText("Rechercher dans l'historique... (Entree = precedent, Echap = fermer)")
|
||||
self.search_status = QLabel("")
|
||||
self.search_status.setObjectName("statusLabel")
|
||||
close_search = QToolButton()
|
||||
close_search.setText("✕")
|
||||
search_layout.addWidget(self.search_input, 1)
|
||||
search_layout.addWidget(self.search_status)
|
||||
search_layout.addWidget(close_search)
|
||||
self.search_bar.hide()
|
||||
layout.addWidget(self.search_bar)
|
||||
|
||||
self.terminal.search_requested.connect(self.open_search)
|
||||
self.search_input.returnPressed.connect(self._run_search)
|
||||
close_search.clicked.connect(self.close_search)
|
||||
QShortcut(QKeySequence(Qt.Key_Escape), self.search_input, activated=self.close_search)
|
||||
|
||||
self.connection = None # assigne par les sous-classes
|
||||
self._log_file = None
|
||||
|
||||
# ---------- journalisation ----------
|
||||
def toggle_logging(self) -> None:
|
||||
if self._log_file:
|
||||
self.stop_logging()
|
||||
return
|
||||
default = Path.home() / f"session-{datetime.now():%Y%m%d-%H%M%S}.log"
|
||||
path, _ = QFileDialog.getSaveFileName(self, "Journaliser la session", str(default), "Journal (*.log)")
|
||||
if not path:
|
||||
self.log_btn.setChecked(False)
|
||||
return
|
||||
try:
|
||||
self._log_file = open(path, "a", encoding="utf-8", errors="replace")
|
||||
except OSError as exc:
|
||||
QMessageBox.critical(self, "Erreur", f"Impossible d'ecrire le journal : {exc}")
|
||||
self.log_btn.setChecked(False)
|
||||
return
|
||||
self._log_file.write(f"\n===== Session du {datetime.now():%d/%m/%Y %H:%M:%S} =====\n")
|
||||
self.log_btn.setChecked(True)
|
||||
self.log_btn.setToolTip(f"Journalisation active : {path}")
|
||||
|
||||
def stop_logging(self) -> None:
|
||||
if self._log_file:
|
||||
try:
|
||||
self._log_file.close()
|
||||
except OSError:
|
||||
pass
|
||||
self._log_file = None
|
||||
self.log_btn.setChecked(False)
|
||||
self.log_btn.setToolTip("Enregistrer la session dans un fichier")
|
||||
|
||||
def _log_output(self, data: bytes) -> None:
|
||||
if not self._log_file:
|
||||
return
|
||||
# on retire les sequences d'echappement : un journal doit rester lisible
|
||||
text = ANSI_ESCAPE_RE.sub("", data.decode("utf-8", errors="replace"))
|
||||
try:
|
||||
self._log_file.write(text)
|
||||
self._log_file.flush()
|
||||
except OSError:
|
||||
self.stop_logging()
|
||||
|
||||
def _wire_connection(self) -> None:
|
||||
self.connection.connected.connect(self._on_connected)
|
||||
self.connection.data_received.connect(self.terminal.feed)
|
||||
self.connection.data_received.connect(self._log_output)
|
||||
self.connection.disconnected.connect(self._on_disconnected)
|
||||
self.connection.error.connect(self._on_error)
|
||||
self.terminal.data_to_send.connect(self.connection.send)
|
||||
self.terminal.resized.connect(self.connection.resize)
|
||||
|
||||
def _create_connection(self):
|
||||
"""Fabrique la connexion. Redefinie par chaque sous-classe : c'est ce qui
|
||||
permet de la recreer a l'identique lors d'une reconnexion."""
|
||||
raise NotImplementedError
|
||||
|
||||
def start_connection(self) -> None:
|
||||
self.connection = self._create_connection()
|
||||
self._wire_connection()
|
||||
self.connection.start()
|
||||
|
||||
def reconnect(self) -> None:
|
||||
if self.connection:
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(1500)
|
||||
self.terminal.reset_screen()
|
||||
self.reconnect_btn.hide()
|
||||
self.status_label.setText("○ Reconnexion...")
|
||||
self.start_connection()
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.reconnect_btn.hide()
|
||||
self.terminal.setFocus()
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
self.status_label.setText(f"● Deconnecte ({reason})")
|
||||
self.reconnect_btn.show()
|
||||
|
||||
def _on_error(self, message: str) -> None:
|
||||
self.status_label.setText(f"● Erreur : {message}")
|
||||
self.reconnect_btn.show()
|
||||
QMessageBox.critical(self, "Erreur", message)
|
||||
|
||||
# ---------- recherche dans l'historique ----------
|
||||
def open_search(self) -> None:
|
||||
self.search_bar.show()
|
||||
self.search_input.setFocus()
|
||||
self.search_input.selectAll()
|
||||
|
||||
def close_search(self) -> None:
|
||||
self.search_bar.hide()
|
||||
self.search_status.clear()
|
||||
self.terminal.setFocus()
|
||||
|
||||
def _run_search(self) -> None:
|
||||
needle = self.search_input.text()
|
||||
if not needle:
|
||||
return
|
||||
found = self.terminal.find_text(needle)
|
||||
self.search_status.setText("" if found else "aucun resultat")
|
||||
|
||||
def close_connection(self) -> None:
|
||||
self.stop_logging()
|
||||
if self.connection:
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(2000)
|
||||
|
||||
|
||||
class SessionTabWidget(BaseTerminalTab):
|
||||
"""Un onglet = une connexion SSH avec son terminal et un panneau SFTP optionnel."""
|
||||
|
||||
def __init__(self, session: Session, secret: str, parent=None):
|
||||
super().__init__(f"○ Connexion a {session.username}@{session.host}...", parent)
|
||||
self.session = session
|
||||
self.sftp_panel: SFTPPanel | None = None
|
||||
|
||||
self.sftp_toggle = QToolButton()
|
||||
self.sftp_toggle.setText("📁 Fichiers (SFTP)")
|
||||
self.sftp_toggle.setToolTip("Afficher / masquer le navigateur de fichiers SFTP")
|
||||
self.sftp_toggle.setCheckable(True)
|
||||
self.sftp_toggle.setChecked(True)
|
||||
self.sftp_toggle.setEnabled(False)
|
||||
self.bar.addWidget(self.sftp_toggle)
|
||||
|
||||
self.tunnel_btn = QToolButton()
|
||||
self.tunnel_btn.setText("🔀 Tunnels")
|
||||
self.tunnel_btn.setToolTip("Redirections de ports SSH (-L / -R)")
|
||||
self.tunnel_btn.setEnabled(False)
|
||||
self.tunnel_btn.clicked.connect(self.open_tunnels)
|
||||
self.bar.addWidget(self.tunnel_btn)
|
||||
|
||||
self.tunnel_manager: TunnelManager | None = None
|
||||
|
||||
self._secret = secret
|
||||
self.sftp_toggle.toggled.connect(self._toggle_sftp)
|
||||
self.start_connection()
|
||||
|
||||
def _create_connection(self):
|
||||
session = self.session
|
||||
connection = SSHConnection(
|
||||
host=session.host,
|
||||
port=session.port,
|
||||
username=session.username,
|
||||
auth_type=session.auth_type,
|
||||
key_path=session.key_path,
|
||||
password=self._secret if session.auth_type == AUTH_PASSWORD else "",
|
||||
passphrase=self._secret if session.auth_type == AUTH_KEY else "",
|
||||
# comme MobaXterm : le navigateur de fichiers est la des l'ouverture,
|
||||
# pas cache derriere une action supplementaire
|
||||
open_sftp_on_connect=True,
|
||||
)
|
||||
connection.sftp_ready.connect(self._on_sftp_ready)
|
||||
connection.password_prompt.connect(self._on_password_prompt)
|
||||
connection.host_key_prompt.connect(self._on_host_key_prompt)
|
||||
connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt)
|
||||
connection.passphrase_prompt.connect(self._on_passphrase_prompt)
|
||||
return connection
|
||||
|
||||
def open_tunnels(self) -> None:
|
||||
if not self.tunnel_manager:
|
||||
QMessageBox.information(self, "Tunnels", "La session doit etre connectee.")
|
||||
return
|
||||
TunnelDialog(self.tunnel_manager, parent=self).exec()
|
||||
actifs = len(self.tunnel_manager.tunnels)
|
||||
self.tunnel_btn.setText(f"🔀 Tunnels ({actifs})" if actifs else "🔀 Tunnels")
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.status_label.setText(f"● Connecte : {self.session.username}@{self.session.host}:{self.session.port}")
|
||||
transport = self.connection.client.get_transport() if self.connection.client else None
|
||||
if transport:
|
||||
self.tunnel_manager = TunnelManager(transport)
|
||||
self.tunnel_btn.setEnabled(True)
|
||||
# le bouton n'est active qu'une fois le canal SFTP reellement ouvert
|
||||
# (_on_sftp_ready) : certains serveurs n'exposent pas le sous-systeme SFTP
|
||||
super()._on_connected()
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
super()._on_disconnected(reason)
|
||||
self.sftp_toggle.setEnabled(False)
|
||||
|
||||
def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None:
|
||||
accepted = confirm_host_key(self, hostname, key_type, fingerprint)
|
||||
self.connection.confirm_host_key(accepted)
|
||||
|
||||
def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None:
|
||||
accepted = confirm_host_key_change(self, hostname, key_type, old_fp, new_fp)
|
||||
self.connection.confirm_host_key_change(accepted)
|
||||
|
||||
def _on_passphrase_prompt(self, key_path: str) -> None:
|
||||
text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}")
|
||||
self.connection.provide_passphrase(text if ok else None)
|
||||
|
||||
def _on_password_prompt(self, target: str) -> None:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :")
|
||||
self.connection.provide_password(text if ok else None)
|
||||
|
||||
def _on_sftp_ready(self, sftp) -> None:
|
||||
"""Le canal SFTP a ete ouvert en arriere-plan : on affiche le navigateur de
|
||||
fichiers a cote du terminal, sans action de l'utilisateur."""
|
||||
if self.sftp_panel is not None:
|
||||
return
|
||||
self.sftp_panel = SFTPPanel(sftp, transfer_sftp_factory=self._open_transfer_channel)
|
||||
self.splitter.insertWidget(0, self.sftp_panel)
|
||||
self.splitter.setSizes([460, 540])
|
||||
self.sftp_toggle.setEnabled(True)
|
||||
self.sftp_toggle.setChecked(True)
|
||||
self.terminal.setFocus()
|
||||
|
||||
def _open_transfer_channel(self):
|
||||
"""Ouvre un second canal SFTP, dedie aux transferts, pour ne jamais faire
|
||||
attendre la navigation derriere un gros envoi/reception (voir SFTPPanel)."""
|
||||
import paramiko
|
||||
return paramiko.SFTPClient.from_transport(self.connection.client.get_transport())
|
||||
|
||||
def close_connection(self) -> None:
|
||||
if self.sftp_panel:
|
||||
self.sftp_panel.shutdown()
|
||||
if self.tunnel_manager:
|
||||
self.tunnel_manager.stop_all()
|
||||
self.tunnel_manager = None
|
||||
super().close_connection()
|
||||
|
||||
def _toggle_sftp(self, checked: bool) -> None:
|
||||
if self.sftp_panel is not None:
|
||||
self.sftp_panel.setVisible(checked)
|
||||
|
||||
|
||||
class SFTPTabWidget(QWidget):
|
||||
"""Onglet SFTP autonome (facon WinSCP) : double panneau de fichiers, sans terminal."""
|
||||
|
||||
def __init__(self, session: Session, secret: str, parent=None):
|
||||
super().__init__(parent)
|
||||
self.session = session
|
||||
self.sftp_panel: SFTPPanel | None = None
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
bar = QHBoxLayout()
|
||||
bar.setContentsMargins(10, 6, 10, 6)
|
||||
self.status_label = QLabel(f"○ Connexion SFTP a {session.username}@{session.host}...")
|
||||
self.status_label.setObjectName("statusLabel")
|
||||
bar.addWidget(self.status_label, 1)
|
||||
bar_frame = QFrame()
|
||||
bar_frame.setObjectName("tabBar")
|
||||
bar_frame.setLayout(bar)
|
||||
layout.addWidget(bar_frame)
|
||||
|
||||
self.container = QVBoxLayout()
|
||||
holder = QWidget()
|
||||
holder.setLayout(self.container)
|
||||
layout.addWidget(holder, 1)
|
||||
|
||||
self.connection = SSHConnection(
|
||||
host=session.host,
|
||||
port=session.port,
|
||||
username=session.username,
|
||||
auth_type=session.auth_type,
|
||||
key_path=session.key_path,
|
||||
password=secret if session.auth_type == AUTH_PASSWORD else "",
|
||||
passphrase=secret if session.auth_type == AUTH_KEY else "",
|
||||
open_shell=False,
|
||||
open_sftp_on_connect=True,
|
||||
)
|
||||
self.connection.sftp_ready.connect(self._on_sftp_ready)
|
||||
self.connection.password_prompt.connect(self._on_password_prompt)
|
||||
self.connection.error.connect(self._on_error)
|
||||
self.connection.disconnected.connect(self._on_disconnected)
|
||||
self.connection.host_key_prompt.connect(self._on_host_key_prompt)
|
||||
self.connection.host_key_changed_prompt.connect(self._on_host_key_changed_prompt)
|
||||
self.connection.passphrase_prompt.connect(self._on_passphrase_prompt)
|
||||
self.connection.start()
|
||||
|
||||
def _on_sftp_ready(self, sftp) -> None:
|
||||
self.sftp_panel = SFTPPanel(sftp, transfer_sftp_factory=self._open_transfer_channel)
|
||||
self.container.addWidget(self.sftp_panel)
|
||||
self.status_label.setText(
|
||||
f"● SFTP connecte : {self.session.username}@{self.session.host}:{self.session.port}"
|
||||
)
|
||||
|
||||
def _open_transfer_channel(self):
|
||||
import paramiko
|
||||
return paramiko.SFTPClient.from_transport(self.connection.client.get_transport())
|
||||
|
||||
def _on_disconnected(self, reason: str) -> None:
|
||||
self.status_label.setText(f"● Deconnecte ({reason})")
|
||||
|
||||
def _on_error(self, message: str) -> None:
|
||||
self.status_label.setText(f"● Erreur : {message}")
|
||||
QMessageBox.critical(self, "Erreur SFTP", message)
|
||||
|
||||
def _on_host_key_prompt(self, hostname: str, key_type: str, fingerprint: str) -> None:
|
||||
self.connection.confirm_host_key(confirm_host_key(self, hostname, key_type, fingerprint))
|
||||
|
||||
def _on_host_key_changed_prompt(self, hostname: str, key_type: str, old_fp: str, new_fp: str) -> None:
|
||||
self.connection.confirm_host_key_change(
|
||||
confirm_host_key_change(self, hostname, key_type, old_fp, new_fp)
|
||||
)
|
||||
|
||||
def _on_passphrase_prompt(self, key_path: str) -> None:
|
||||
text, ok = ask_secret(self, "Passphrase requise", f"Passphrase pour la cle :\n{key_path}")
|
||||
self.connection.provide_passphrase(text if ok else None)
|
||||
|
||||
def _on_password_prompt(self, target: str) -> None:
|
||||
text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {target} :")
|
||||
self.connection.provide_password(text if ok else None)
|
||||
|
||||
def close_connection(self) -> None:
|
||||
if self.sftp_panel:
|
||||
self.sftp_panel.shutdown()
|
||||
self.connection.close_connection()
|
||||
self.connection.wait(2000)
|
||||
|
||||
|
||||
class LocalTerminalTabWidget(BaseTerminalTab):
|
||||
"""Un onglet = un terminal Windows local (PowerShell / cmd), sans session enregistree."""
|
||||
|
||||
sftp_requested = Signal(object) # cible ssh detectee (ou None)
|
||||
|
||||
def __init__(self, shell: str = SHELL_POWERSHELL, parent=None):
|
||||
super().__init__(f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...", parent)
|
||||
self.shell = shell
|
||||
|
||||
# Meme dans un terminal local, on tape souvent "ssh user@ip" a la main :
|
||||
# ce bouton ouvre le SFTP sur cet hote, sans passer par un favori.
|
||||
self.sftp_btn = QToolButton()
|
||||
self.sftp_btn.setText("📁 SFTP sur cette connexion")
|
||||
self.sftp_btn.setToolTip(
|
||||
"Ouvre un gestionnaire de fichiers SFTP sur l'hote de la commande ssh tapee ici"
|
||||
)
|
||||
self.sftp_btn.clicked.connect(
|
||||
lambda: self.sftp_requested.emit(detect_ssh_target(self.terminal.buffer_text()))
|
||||
)
|
||||
self.bar.addWidget(self.sftp_btn)
|
||||
|
||||
self.start_connection()
|
||||
|
||||
def _create_connection(self):
|
||||
return LocalTerminalConnection(shell=self.shell)
|
||||
|
||||
def _on_connected(self) -> None:
|
||||
self.status_label.setText(f"● Terminal local — {SHELL_LABELS.get(self.shell, self.shell)}")
|
||||
super()._on_connected()
|
||||
Reference in New Issue
Block a user