pyssh-client/app/controllers/connection_controller.py

"""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()