"""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_DYNAMIC, 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.
"
"Une redirection dynamique ouvre un proxy SOCKS sur votre poste : chaque "
"application qui s'y connecte choisit elle-meme sa destination, atteinte a travers "
"la connexion SSH (pratique pour naviguer comme si vous etiez sur le reseau du serveur)."
)
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)
self.form = form
self.direction.currentIndexChanged.connect(self._update_dest_fields)
self._update_dest_fields()
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 _update_dest_fields(self) -> None:
"""La destination d'une redirection dynamique (-D) est negociee par
chaque client SOCKS : les champs n'ont pas de sens dans ce mode."""
has_dest = self.direction.currentData() != DIRECTION_DYNAMIC
self.form.setRowVisible(self.dest_host, has_dest)
self.form.setRowVisible(self.dest_port, has_dest)
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:
direction = self.direction.currentData()
if direction != DIRECTION_DYNAMIC and not self.dest_host.text().strip():
QMessageBox.warning(self, "Destination manquante", "Indiquez la machine de destination.")
return
config = {"direction": direction, "listen_port": self.listen_port.value()}
if direction != DIRECTION_DYNAMIC:
config["dest_host"] = self.dest_host.text().strip()
config["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()