diff --git a/README.md b/README.md index ca65330..0aadd25 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,12 @@ Client SSH/SFTP personnel (type MobaXterm/Termius/Bitvise), 100% local, sans com - Confort d'affichage : zoom `Ctrl+molette` ou `Ctrl+ +/-` (`Ctrl+0` remet a zero, la taille est memorisee), barre de defilement de l'historique, `Maj+PagePrec/Suiv`, `Ctrl+Fin` pour revenir en direct, et recherche dans l'historique avec `Ctrl+F` +- **Tunnels SSH** (redirection de ports) : bouton "Tunnels" dans la barre d'un onglet SSH. + Redirection locale (`-L`) pour atteindre depuis votre poste un service qui n'ecoute que sur la + boucle locale du serveur, et redirection distante (`-R`) dans l'autre sens +- **Envoi simultane** : le bouton "⇉ Multi" du panneau lateral diffuse ce que vous tapez vers tous + les onglets terminal ouverts, pour piloter plusieurs serveurs d'un coup +- Journalisation de session dans un fichier (bouton "Journal"), sequences ANSI retirees pour rester lisible - Reconnexion en un clic quand une session tombe (bouton dans la barre de l'onglet) : pratique avec des serveurs de test qui redemarrent souvent - Les hotes deja approuves par votre commande `ssh` (`~/.ssh/known_hosts`) sont reconnus : pas de @@ -126,7 +132,8 @@ app/ terminal_widget.py emulateur de terminal ANSI (pyte + QPainter) sftp_panel.py client SFTP a deux panneaux main_window.py fenetre principale (arbre de sessions + onglets) - dialogs/ boites de dialogue (session, cles, connexion) + tunnels.py tunnels SSH (redirection de ports -L / -R) + dialogs/ boites de dialogue (session, cles, hotes, tunnels, connexion) resources/ theme.qss feuille de style Qt "bleu nuit" data/ diff --git a/app/dialogs/tunnel_dialog.py b/app/dialogs/tunnel_dialog.py new file mode 100644 index 0000000..766e8f1 --- /dev/null +++ b/app/dialogs/tunnel_dialog.py @@ -0,0 +1,114 @@ +"""Gestion des tunnels SSH (redirections de ports) d'une session ouverte.""" +from __future__ import annotations + +from PySide6.QtCore import Qt +from PySide6.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QFormLayout, QTreeWidget, QTreeWidgetItem, + QPushButton, QComboBox, QSpinBox, QLineEdit, QLabel, QMessageBox, QAbstractItemView, +) + +from app.tunnels import ( + DIRECTION_LOCAL, DIRECTION_REMOTE, DIRECTION_LABELS, TunnelManager, describe, +) + + +class TunnelDialog(QDialog): + def __init__(self, manager: TunnelManager, parent=None): + super().__init__(parent) + self.manager = manager + self.setWindowTitle("Tunnels SSH (redirection de ports)") + self.setMinimumSize(660, 400) + + layout = QVBoxLayout(self) + + info = QLabel( + "Une redirection locale ouvre un port sur votre poste et l'envoie, a travers " + "la connexion SSH, vers une machine joignable depuis le serveur — pratique pour " + "atteindre un service qui n'ecoute que sur la boucle locale du serveur.
" + "Une redirection distante fait l'inverse : un port ouvert sur le serveur est " + "relaye vers une machine joignable depuis votre poste." + ) + info.setWordWrap(True) + layout.addWidget(info) + + self.tree = QTreeWidget() + self.tree.setHeaderLabels(["Sens", "Redirection", "Etat"]) + self.tree.setRootIsDecorated(False) + self.tree.setSelectionMode(QAbstractItemView.SingleSelection) + self.tree.setColumnWidth(0, 110) + self.tree.setColumnWidth(1, 380) + layout.addWidget(self.tree) + + form = QFormLayout() + self.direction = QComboBox() + for key, label in DIRECTION_LABELS.items(): + self.direction.addItem(label, key) + self.listen_port = QSpinBox(); self.listen_port.setRange(1, 65535); self.listen_port.setValue(8080) + self.dest_host = QLineEdit("127.0.0.1") + self.dest_port = QSpinBox(); self.dest_port.setRange(1, 65535); self.dest_port.setValue(80) + + form.addRow("Sens :", self.direction) + form.addRow("Port a ouvrir :", self.listen_port) + form.addRow("Machine de destination :", self.dest_host) + form.addRow("Port de destination :", self.dest_port) + layout.addLayout(form) + + buttons = QHBoxLayout() + self.add_btn = QPushButton("Ouvrir le tunnel") + self.add_btn.setObjectName("primary") + self.remove_btn = QPushButton("Fermer le tunnel selectionne") + buttons.addWidget(self.add_btn) + buttons.addWidget(self.remove_btn) + buttons.addStretch() + close_btn = QPushButton("Fermer") + close_btn.clicked.connect(self.accept) + buttons.addWidget(close_btn) + layout.addLayout(buttons) + + self.add_btn.clicked.connect(self.add_tunnel) + self.remove_btn.clicked.connect(self.remove_tunnel) + self.refresh() + + def refresh(self) -> None: + self.tree.clear() + for tunnel in self.manager.tunnels: + etat = f"Erreur : {tunnel.error}" if tunnel.error else "Actif" + item = QTreeWidgetItem([ + DIRECTION_LABELS.get(tunnel.direction, tunnel.direction), + describe(tunnel.config), + etat, + ]) + item.setData(0, Qt.UserRole, id(tunnel)) + self.tree.addTopLevelItem(item) + self.remove_btn.setEnabled(bool(self.manager.tunnels)) + + def add_tunnel(self) -> None: + if not self.dest_host.text().strip(): + QMessageBox.warning(self, "Destination manquante", "Indiquez la machine de destination.") + return + config = { + "direction": self.direction.currentData(), + "listen_port": self.listen_port.value(), + "dest_host": self.dest_host.text().strip(), + "dest_port": self.dest_port.value(), + } + tunnel = self.manager.start(config) + if tunnel.error: + QMessageBox.critical( + self, "Tunnel impossible", + f"{describe(config)}\n\n{tunnel.error}\n\n" + "Le port est peut-etre deja utilise sur le poste.", + ) + self.manager.stop(tunnel) + self.refresh() + + def remove_tunnel(self) -> None: + item = self.tree.currentItem() + if not item: + return + target = item.data(0, Qt.UserRole) + for tunnel in list(self.manager.tunnels): + if id(tunnel) == target: + self.manager.stop(tunnel) + break + self.refresh() diff --git a/app/main_window.py b/app/main_window.py index 8686dd9..0363831 100644 --- a/app/main_window.py +++ b/app/main_window.py @@ -1,128 +1,48 @@ """Fenetre principale : arborescence des sessions, onglets terminal/SFTP.""" - - from __future__ import annotations - - - - import copy - - import re - - import uuid - - from datetime import datetime - - from pathlib import Path - - - - from PySide6.QtCore import Qt, Signal - - from PySide6.QtGui import QAction, QActionGroup, QKeySequence, QShortcut - - from PySide6.QtWidgets import ( - - QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget, - - QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter, - - QStatusBar, QFrame, QApplication, QLineEdit, QFileDialog, - - ) - - - - from app.session_manager import SessionStore, Session, AUTH_KEY, AUTH_PASSWORD - - from app.key_manager import KeyManager - - from app.ssh_connection import SSHConnection - - from app.local_terminal import ( LocalTerminalConnection, SHELL_POWERSHELL, SHELL_CMD, SHELL_LABELS, 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 - - - - SESSION_ROLE = Qt.UserRole + 10 - - LOCAL_SHELL_ROLE = Qt.UserRole + 11 - - - - # Repere une commande ssh tapee a la main dans un terminal, pour proposer le SFTP - - # sur cet hote sans avoir a creer de favori. - - SSH_COMMAND_RE = re.compile(r"\bssh\s+(?P[^\r\n|;&]+)") - - # options ssh qui consomment la valeur suivante (a ne pas confondre avec l'hote) - - SSH_VALUE_OPTIONS = { - - "-p", "-i", "-l", "-o", "-F", "-J", "-b", "-c", "-D", "-L", "-R", "-W", "-e", "-m", "-Q", - - } - - HOST_RE = re.compile(r"^[A-Za-z0-9][\w.\-]*$") - - # sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles - - ANSI_ESCAPE_RE = re.compile( r"\x1b\[[0-9;?]*[a-zA-Z]" # CSI : couleurs, deplacements du curseur r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC : titre de fenetre @@ -132,1363 +52,484 @@ ANSI_ESCAPE_RE = re.compile( ) - - - - 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() - - - - - - class MainWindow(QMainWindow): - - def __init__(self): - - super().__init__() - - self.setWindowTitle("PySSH Client") - - self.resize(1280, 820) - - - - self.session_store = SessionStore() - - self.key_manager = KeyManager() - - - - self._build_menu() - - self._build_sidebar() - - self._build_tabs() - - self.setStatusBar(QStatusBar()) - - - - self.reload_sessions() - - - - # demarre directement dans un terminal utilisable, sans avoir a en choisir un - - self.open_local_terminal(DEFAULT_SHELL) - - - - # ---------- construction UI ---------- - - def _build_menu(self) -> None: - - file_menu = self.menuBar().addMenu("&Fichier") - - new_session_action = QAction("Nouvelle session...", self) - - new_session_action.setShortcut(QKeySequence("Ctrl+N")) - - new_session_action.triggered.connect(self.new_session) - - file_menu.addAction(new_session_action) - - - - file_menu.addSeparator() - - - - quick_ssh_action = QAction("Connexion SSH rapide...", self) - - quick_ssh_action.setShortcut(QKeySequence("Ctrl+Shift+S")) - - quick_ssh_action.triggered.connect(self.quick_ssh) - - file_menu.addAction(quick_ssh_action) - - - - quick_sftp_action = QAction("Connexion SFTP (gestionnaire de fichiers)...", self) - - quick_sftp_action.setShortcut(QKeySequence("Ctrl+Shift+F")) - - quick_sftp_action.triggered.connect(self.quick_sftp) - - file_menu.addAction(quick_sftp_action) - - - - file_menu.addSeparator() - - - - default_action = QAction( f"Nouveau terminal local ({SHELL_LABELS.get(DEFAULT_SHELL, DEFAULT_SHELL)})", self ) @@ -1500,1039 +541,375 @@ class MainWindow(QMainWindow): for shell_id, label in SHELL_LABELS.items(): others_menu.addAction(label, lambda s=shell_id: self.open_local_terminal(s)) - - - - file_menu.addSeparator() - - quit_action = QAction("Quitter", self) - - quit_action.triggered.connect(self.close) - - file_menu.addAction(quit_action) - - - - keys_menu = self.menuBar().addMenu("&Cles SSH") - - manage_keys_action = QAction("Gestionnaire de cles...", self) - - manage_keys_action.triggered.connect(self.open_key_manager) - - keys_menu.addAction(manage_keys_action) - - - - view_menu = self.menuBar().addMenu("&Affichage") - - theme_menu = view_menu.addMenu("Theme") - - self.theme_group = QActionGroup(self) - - self.theme_group.setExclusive(True) - - for theme_id, label in THEME_LABELS.items(): - - action = QAction(label, self, checkable=True) - - action.setChecked(theme_id == current_theme()) - - action.triggered.connect(lambda _checked=False, t=theme_id: self.change_theme(t)) - - self.theme_group.addAction(action) - - theme_menu.addAction(action) - - - - hosts_menu = self.menuBar().addMenu("&Hotes connus") - - manage_hosts_action = QAction("Gestionnaire des hotes connus...", self) - - manage_hosts_action.triggered.connect(self.open_known_hosts_manager) - - hosts_menu.addAction(manage_hosts_action) - - - - help_menu = self.menuBar().addMenu("&Aide") - - about_action = QAction("A propos", self) - - about_action.triggered.connect(self.show_about) - - help_menu.addAction(about_action) - - - - def _build_sidebar(self) -> None: - - dock = QDockWidget("Sessions", self) - - dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) - - - - container = QWidget() - - layout = QVBoxLayout(container) - - layout.setContentsMargins(6, 6, 6, 6) - - layout.setSpacing(6) - - - - toolbar = QHBoxLayout() - - new_btn = QToolButton() - - new_btn.setObjectName("primary") - - new_btn.setText("+ Session") - - new_btn.clicked.connect(self.new_session) - - toolbar.addWidget(new_btn) - - - - term_btn = QToolButton() - - term_btn.setText("💻 Terminal") - - term_btn.setPopupMode(QToolButton.InstantPopup) - - term_menu = QMenu(term_btn) 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) - - - - dock.setWidget(container) - - self.addDockWidget(Qt.LeftDockWidgetArea, dock) - - - - def _build_tabs(self) -> None: - - self.tabs = QTabWidget() - - self.tabs.setTabsClosable(True) - - self.tabs.setMovable(True) - - self.tabs.tabCloseRequested.connect(self._on_tab_close) - - self.setCentralWidget(self.tabs) - - - - # ---------- sessions / arbre ---------- - - def reload_sessions(self) -> None: - - self.tree.clear() - - - - locals_header = QTreeWidgetItem(["💻 Terminaux locaux"]) - - locals_header.setFlags(locals_header.flags() & ~Qt.ItemIsSelectable) - - font = locals_header.font(0) - - font.setBold(True) - - locals_header.setFont(0, font) - - self.tree.addTopLevelItem(locals_header) - - for shell, label in SHELL_LABELS.items(): - - child = QTreeWidgetItem([f" {label}"]) - - child.setData(0, LOCAL_SHELL_ROLE, shell) - - locals_header.addChild(child) - - - - sessions_header = QTreeWidgetItem(["🔐 Sessions SSH"]) - - sessions_header.setFlags(sessions_header.flags() & ~Qt.ItemIsSelectable) - - sessions_header.setFont(0, font) - - self.tree.addTopLevelItem(sessions_header) - - - - folder_items: dict[str, QTreeWidgetItem] = {} - - - - def get_folder(path: str) -> QTreeWidgetItem: - - if not path: - - return sessions_header - - if path in folder_items: - - return folder_items[path] - - parent_path, _, name = path.rpartition("/") - - parent_item = get_folder(parent_path) - - item = QTreeWidgetItem([f"📁 {name}"]) - - item.setFlags(item.flags() & ~Qt.ItemIsSelectable) - - parent_item.addChild(item) - - folder_items[path] = item - - return item - - - - for session in sorted(self.session_store.sessions, key=lambda s: s.name.lower()): - - parent = get_folder(session.group) - - item = QTreeWidgetItem([f"🖥 {session.name} ({session.host})"]) - - item.setData(0, SESSION_ROLE, session.id) - - parent.addChild(item) - - - - if not self.session_store.sessions: - - hint = QTreeWidgetItem([" (aucune session — bouton « + Session » ou « 📁 SFTP »)"]) - - hint.setFlags(hint.flags() & ~Qt.ItemIsSelectable) - - hint.setForeground(0, Qt.gray) - - sessions_header.addChild(hint) - - - - self.tree.expandAll() - - - - def _session_from_item(self, item: QTreeWidgetItem) -> Session | None: - - if not item: - - return None - - session_id = item.data(0, SESSION_ROLE) - - if not session_id: - - return None - - return next((s for s in self.session_store.sessions if s.id == session_id), None) - - - - def _shell_from_item(self, item: QTreeWidgetItem) -> str | None: - - if not item: - - return None - - return item.data(0, LOCAL_SHELL_ROLE) - - - - def _on_tree_double_click(self, item: QTreeWidgetItem, _col: int) -> None: - - session = self._session_from_item(item) - - if session: - - self.open_session(session) - - return - - shell = self._shell_from_item(item) - - if shell: - - self.open_local_terminal(shell) - - - - def _on_tree_context_menu(self, pos) -> None: - - item = self.tree.itemAt(pos) - - menu = QMenu(self) - - session = self._session_from_item(item) if item else None - - shell = self._shell_from_item(item) if item else None - - - - if session: - - connect_action = menu.addAction("🖥 Ouvrir un terminal SSH") - - sftp_action = menu.addAction("📁 Ouvrir en SFTP (gestionnaire de fichiers)") - - menu.addSeparator() - - edit_action = menu.addAction("Modifier...") - - duplicate_action = menu.addAction("Dupliquer") - - delete_action = menu.addAction("Supprimer") - - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - - if action == connect_action: - - self.open_session(session) - - elif action == sftp_action: - - self.open_sftp_session(session) - - elif action == edit_action: - - self.edit_session(session) - - elif action == duplicate_action: - - self.duplicate_session(session) - - elif action == delete_action: - - self.delete_session(session) - - elif shell: - - open_action = menu.addAction("Ouvrir") - - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - - if action == open_action: - - self.open_local_terminal(shell) - - else: - - new_action = menu.addAction("Nouvelle session...") - - action = menu.exec(self.tree.viewport().mapToGlobal(pos)) - - if action == new_action: - - self.new_session() - - - - def new_session(self) -> None: - - dialog = SessionDialog(self.key_manager, parent=self) - - if dialog.exec(): - - self.session_store.add(dialog.result_session()) - - self.reload_sessions() - - - - def edit_session(self, session: Session) -> None: - - dialog = SessionDialog(self.key_manager, session=session, parent=self) - - if dialog.exec(): - - self.session_store.update(dialog.result_session()) - - self.reload_sessions() - - - - def duplicate_session(self, session: Session) -> None: - - new_session = copy.deepcopy(session) - - new_session.id = str(uuid.uuid4()) - - new_session.name = f"{session.name} (copie)" - - self.session_store.add(new_session) - - self.reload_sessions() - - - - def delete_session(self, session: Session) -> None: - - if QMessageBox.question(self, "Supprimer", f"Supprimer la session '{session.name}' ?") == QMessageBox.Yes: - - self.session_store.delete(session.id) - - self.reload_sessions() - - - - # ---------- connexion / onglets ---------- - - def open_session(self, session: Session) -> None: - - secret = "" - - if session.auth_type == AUTH_PASSWORD: - - text, ok = ask_secret(self, "Mot de passe", f"Mot de passe pour {session.username}@{session.host} :") - - if not ok: - - return - - secret = text - - - - tab = SessionTabWidget(session, secret) - - + 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)) def _on_tab_close(self, index: int) -> None: - - widget = self.tabs.widget(index) - - if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): - - widget.close_connection() - - self.tabs.removeTab(index) - - - - # ---------- cles / a propos ---------- - - def change_theme(self, theme: str) -> None: - - """Applique un theme a chaud : feuille de style de l'interface + couleurs des - - terminaux deja ouverts (qui ont leur propre palette, hors QSS).""" - - app = QApplication.instance() - - if app is None: - - return - - apply_theme(app, theme) - - save_theme(theme) - - for widget in self.findChildren(TerminalWidget): - - widget.refresh_theme() - - - - def open_key_manager(self) -> None: - - dialog = KeyManagerDialog(self.key_manager, parent=self) - - dialog.exec() - - - - def open_known_hosts_manager(self) -> None: - - dialog = KnownHostsDialog(parent=self) - - dialog.exec() - - - - def show_about(self) -> None: - - QMessageBox.information( - - self, "A propos", - - "PySSH Client\n\nClient SSH/SFTP local avec gestion de sessions, de cles, " - - "et terminaux locaux integres.\nBase sur PySide6, paramiko, pywinpty et pyte.", - - ) - - - - def closeEvent(self, event) -> None: - - for i in range(self.tabs.count()): - - widget = self.tabs.widget(i) - - if isinstance(widget, (BaseTerminalTab, SFTPTabWidget)): - - widget.close_connection() - - super().closeEvent(event) - - diff --git a/app/tunnels.py b/app/tunnels.py new file mode 100644 index 0000000..dbeefa4 --- /dev/null +++ b/app/tunnels.py @@ -0,0 +1,189 @@ +"""Tunnels SSH (redirection de ports), facon `ssh -L` et `ssh -R`. + +Redirection locale (-L) : un port ouvert sur votre poste est relaye, a travers la +connexion SSH, vers un hote joignable depuis le serveur. Typiquement pour atteindre +l'interface web d'un service qui n'ecoute que sur la boucle locale du serveur. + +Redirection distante (-R) : l'inverse, un port ouvert sur le serveur est relaye vers +une machine joignable depuis votre poste. + +Chaque tunnel tourne dans son propre thread : un relais bloque ne doit gener ni +l'interface ni les autres tunnels. +""" +from __future__ import annotations + +import select +import socket +import socketserver +import threading + +import paramiko + +DIRECTION_LOCAL = "local" +DIRECTION_REMOTE = "remote" + +DIRECTION_LABELS = { + DIRECTION_LOCAL: "Locale (-L)", + DIRECTION_REMOTE: "Distante (-R)", +} + + +def describe(tunnel: dict) -> str: + """Resume lisible, dans le sens de lecture du flux.""" + if tunnel.get("direction") == DIRECTION_REMOTE: + return (f"serveur:{tunnel['listen_port']} -> " + f"{tunnel['dest_host']}:{tunnel['dest_port']} (via ce poste)") + return (f"localhost:{tunnel['listen_port']} -> " + f"{tunnel['dest_host']}:{tunnel['dest_port']} (via le serveur)") + + +def _relay(sock: socket.socket, chan: paramiko.Channel) -> None: + """Recopie les octets dans les deux sens jusqu'a fermeture.""" + try: + while True: + readable, _, _ = select.select([sock, chan], [], [], 1.0) + if sock in readable: + data = sock.recv(4096) + if not data: + break + chan.sendall(data) + if chan in readable: + data = chan.recv(4096) + if not data: + break + sock.sendall(data) + except (OSError, socket.error): + pass + finally: + for closeable in (chan, sock): + try: + closeable.close() + except Exception: + pass + + +class _ForwardServer(socketserver.ThreadingTCPServer): + daemon_threads = True + allow_reuse_address = True + + +class _ForwardHandler(socketserver.BaseRequestHandler): + transport: paramiko.Transport + dest_host: str + dest_port: int + + def handle(self): + try: + chan = self.transport.open_channel( + "direct-tcpip", (self.dest_host, self.dest_port), self.request.getpeername() + ) + except Exception: + return + if chan is None: + return + _relay(self.request, chan) + + +class Tunnel: + """Un tunnel actif, avec de quoi l'arreter.""" + + def __init__(self, config: dict): + self.config = config + self.error: str | None = None + self._server: _ForwardServer | None = None + self._thread: threading.Thread | None = None + self._transport: paramiko.Transport | None = None + self._stop = threading.Event() + + @property + def direction(self) -> str: + return self.config.get("direction", DIRECTION_LOCAL) + + def start(self, transport: paramiko.Transport) -> None: + self._transport = transport + if self.direction == DIRECTION_REMOTE: + self._start_remote(transport) + else: + self._start_local(transport) + + # ---------- redirection locale (-L) ---------- + def _start_local(self, transport: paramiko.Transport) -> None: + config = self.config + handler = type( + "Handler", (_ForwardHandler,), + {"transport": transport, + "dest_host": config["dest_host"], + "dest_port": int(config["dest_port"])}, + ) + bind = config.get("listen_host") or "127.0.0.1" + self._server = _ForwardServer((bind, int(config["listen_port"])), handler) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + + # ---------- redirection distante (-R) ---------- + def _start_remote(self, transport: paramiko.Transport) -> None: + config = self.config + bind = config.get("listen_host") or "" + transport.request_port_forward(bind, int(config["listen_port"])) + self._thread = threading.Thread(target=self._accept_remote, args=(transport,), daemon=True) + self._thread.start() + + def _accept_remote(self, transport: paramiko.Transport) -> None: + while not self._stop.is_set(): + chan = transport.accept(1) + if chan is None: + continue + try: + sock = socket.create_connection( + (self.config["dest_host"], int(self.config["dest_port"])), timeout=10 + ) + except OSError: + try: + chan.close() + except Exception: + pass + continue + threading.Thread(target=_relay, args=(sock, chan), daemon=True).start() + + def stop(self) -> None: + self._stop.set() + if self._server: + try: + self._server.shutdown() + self._server.server_close() + except Exception: + pass + self._server = None + if self.direction == DIRECTION_REMOTE and self._transport: + try: + self._transport.cancel_port_forward( + self.config.get("listen_host") or "", int(self.config["listen_port"]) + ) + except Exception: + pass + + +class TunnelManager: + """Tunnels d'une session SSH donnee.""" + + def __init__(self, transport: paramiko.Transport): + self.transport = transport + self.tunnels: list[Tunnel] = [] + + def start(self, config: dict) -> Tunnel: + tunnel = Tunnel(config) + try: + tunnel.start(self.transport) + except Exception as exc: # noqa: BLE001 + tunnel.error = str(exc) + self.tunnels.append(tunnel) + return tunnel + + def stop(self, tunnel: Tunnel) -> None: + tunnel.stop() + if tunnel in self.tunnels: + self.tunnels.remove(tunnel) + + def stop_all(self) -> None: + for tunnel in list(self.tunnels): + self.stop(tunnel) diff --git a/tests/test_tunnels.py b/tests/test_tunnels.py new file mode 100644 index 0000000..62bb9ff --- /dev/null +++ b/tests/test_tunnels.py @@ -0,0 +1,87 @@ +"""Tunnels SSH : la redirection locale relaie-t-elle reellement le trafic ? + +Le test monte un vrai serveur TCP local et un faux transport SSH dont le canal +'direct-tcpip' est un simple socket vers ce serveur : cela valide toute la +mecanique de relais sans avoir besoin d'un serveur SSH. +""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +import socket +import threading +import time + +from app.tunnels import TunnelManager, DIRECTION_LOCAL, describe + + +class EchoServer(threading.Thread): + """Serveur de destination : renvoie ce qu'il recoit, prefixe.""" + + def __init__(self): + super().__init__(daemon=True) + self.sock = socket.socket() + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.sock.bind(("127.0.0.1", 0)) + self.sock.listen(5) + self.port = self.sock.getsockname()[1] + + def run(self): + while True: + try: + conn, _ = self.sock.accept() + except OSError: + return + data = conn.recv(1024) + conn.sendall(b"echo:" + data) + conn.close() + + +class FakeTransport: + """Imite paramiko.Transport : ouvre un vrai socket vers la destination.""" + + def open_channel(self, kind, dest, src): + assert kind == "direct-tcpip" + return socket.create_connection(dest, timeout=5) + + +echo = EchoServer() +echo.start() +print(f"serveur de destination sur le port {echo.port}") + +manager = TunnelManager(FakeTransport()) +config = { + "direction": DIRECTION_LOCAL, + "listen_port": 0, # port libre choisi par le systeme + "dest_host": "127.0.0.1", + "dest_port": echo.port, +} +tunnel = manager.start(config) +assert tunnel.error is None, tunnel.error +port_local = tunnel._server.server_address[1] +print("tunnel local ouvert sur le port", port_local) +print("description :", describe({**config, "listen_port": port_local})) + +time.sleep(0.3) +client = socket.create_connection(("127.0.0.1", port_local), timeout=5) +client.sendall(b"bonjour") +reponse = client.recv(1024) +client.close() +print("envoye : b'bonjour'") +print("recu :", reponse) +assert reponse == b"echo:bonjour", reponse + +print() +print("=== fermeture du tunnel ===") +manager.stop_all() +time.sleep(0.3) +assert not manager.tunnels +try: + socket.create_connection(("127.0.0.1", port_local), timeout=1).close() + ferme = False +except OSError: + ferme = True +print("le port n'accepte plus de connexion :", ferme) +assert ferme, "le port doit etre libere apres fermeture" + +print() +print("TUNNELS SSH : OK")