"""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, QMessageBox, 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.openssh.config import OpenSSHHost, session_from_openssh_host from app.bastion import BastionTarget, resolve_bastion_target, BastionResolutionError 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 _resolve_bastion(self, session: Session) -> tuple[bool, BastionTarget | None, str]: """Renvoie (ok, bastion_target, bastion_password). ok=False signifie que l'ouverture doit etre annulee (bastion invalide, ou mot de passe du bastion non fourni) ; l'appelant ne doit alors rien ouvrir.""" try: target = resolve_bastion_target(session.bastion, self.session_store) except BastionResolutionError as exc: QMessageBox.critical(self._parent_widget, "Bastion invalide", str(exc)) return False, None, "" if target is None: return True, None, "" if target.auth_type != AUTH_PASSWORD: return True, target, "" text, ok = ask_secret( self._parent_widget, "Mot de passe du bastion", f"Mot de passe pour {target.label} (bastion) :", ) if not ok: return False, None, "" return True, target, text def open_session(self, session: Session) -> None: ok, bastion_target, bastion_password = self._resolve_bastion(session) if not ok: return 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, bastion_target=bastion_target, bastion_password=bastion_password) 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.""" ok, bastion_target, bastion_password = self._resolve_bastion(session) if not ok: return 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, bastion_target=bastion_target, bastion_password=bastion_password) index = self.tabs.addTab(tab, f"📁 {session.name}") self.tabs.setCurrentIndex(index) def open_openssh_host(self, host: OpenSSHHost) -> None: """Connexion directe a un Host de ~/.ssh/config, sans l'enregistrer comme session. Un ProxyJump eventuel est resolu et applique via app.bastion.""" self.open_session(session_from_openssh_host(host)) 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", session_store=self.session_store, ) 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()