diff --git a/app/bastion.py b/app/bastion.py new file mode 100644 index 0000000..dab41ef --- /dev/null +++ b/app/bastion.py @@ -0,0 +1,74 @@ +"""Resolution d'un BastionConfig (session/openssh/manuel) en parametres de +connexion concrets, independamment de la source choisie dans l'editeur. + +La connexion elle-meme (ouverture du canal direct-tcpip a travers le bastion) +est geree par SSHConnection (app.ssh_connection), qui consomme un BastionTarget +sans se soucier de sa provenance. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from app.session_manager import ( + SessionStore, BastionConfig, + BASTION_NONE, BASTION_SESSION, BASTION_OPENSSH, BASTION_MANUAL, + AUTH_AUTO, AUTH_KEY, +) + + +class BastionResolutionError(Exception): + """Le bastion configure ne peut pas etre resolu (reference cassee, config + invalide...). Empeche de lancer une connexion vouee a l'echec.""" + + +@dataclass +class BastionTarget: + """Parametres de connexion resolus pour le saut a franchir avant la cible.""" + host: str + port: int + username: str + auth_type: str + key_path: str = "" + + @property + def label(self) -> str: + return f"{self.username}@{self.host}" if self.username else self.host + + +def resolve_bastion_target( + bastion: BastionConfig | None, session_store: SessionStore | None = None, +) -> BastionTarget | None: + """Renvoie None si aucun bastion n'est configure. Leve BastionResolutionError + si le mode choisi ne peut pas etre resolu en ce moment (session supprimee, + alias OpenSSH introuvable, config manuelle incomplete...).""" + if bastion is None or bastion.mode in (BASTION_NONE, ""): + return None + + if bastion.mode == BASTION_MANUAL: + if not bastion.host.strip(): + raise BastionResolutionError("Bastion manuel : aucune adresse renseignee.") + return BastionTarget(bastion.host.strip(), bastion.port, bastion.username, bastion.auth_type, bastion.key_path) + + if bastion.mode == BASTION_SESSION: + if session_store is None: + raise BastionResolutionError("Bastion vers une session PySSH : aucun registre de sessions disponible.") + target = next((s for s in session_store.sessions if s.id == bastion.session_id), None) + if target is None: + raise BastionResolutionError( + "La session utilisee comme bastion est introuvable (supprimee ou renommee)." + ) + return BastionTarget(target.host, target.port, target.username, target.auth_type, target.key_path) + + if bastion.mode == BASTION_OPENSSH: + from app.openssh.config import load_openssh_hosts + + host = next((h for h in load_openssh_hosts() if h.alias == bastion.openssh_alias), None) + if host is None: + raise BastionResolutionError( + f"Host OpenSSH « {bastion.openssh_alias} » introuvable dans ~/.ssh/config." + ) + key_path = host.identity_files[0] if host.identity_files else "" + auth_type = AUTH_KEY if key_path else AUTH_AUTO + return BastionTarget(host.hostname or host.alias, host.port, host.user, auth_type, key_path) + + raise BastionResolutionError(f"Mode de bastion inconnu : {bastion.mode}") diff --git a/app/controllers/connection_controller.py b/app/controllers/connection_controller.py index cfc9b19..1fc1f82 100644 --- a/app/controllers/connection_controller.py +++ b/app/controllers/connection_controller.py @@ -3,7 +3,7 @@ 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 PySide6.QtWidgets import QTabWidget, QWidget, QMenu, QMessageBox, QInputDialog from app.session_manager import SessionStore, Session, AUTH_PASSWORD from app.key_manager import KeyManager @@ -12,6 +12,7 @@ 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 @@ -45,7 +46,30 @@ class ConnectionController(QObject): 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( @@ -55,13 +79,16 @@ class ConnectionController(QObject): return secret = text - tab = SessionTabWidget(session, secret) + 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( @@ -71,13 +98,13 @@ class ConnectionController(QObject): return secret = text - tab = SFTPTabWidget(session, secret) + 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. Le ProxyJump eventuel n'est pas encore applique (Phase 3).""" + 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: @@ -127,6 +154,7 @@ class ConnectionController(QObject): 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 diff --git a/app/controllers/session_controller.py b/app/controllers/session_controller.py index 91659d3..07095ce 100644 --- a/app/controllers/session_controller.py +++ b/app/controllers/session_controller.py @@ -168,11 +168,7 @@ class SessionController(QObject): item = QTreeWidgetItem([label]) item.setData(0, OPENSSH_HOST_ROLE, host) if host.proxy_jump: - item.setToolTip( - 0, - f"ProxyJump : {host.proxy_jump} (pas encore applique automatiquement, " - "connexion directe pour l'instant)", - ) + item.setToolTip(0, f"ProxyJump : via {host.proxy_jump}") openssh_header.addChild(item) if not hosts: @@ -264,13 +260,15 @@ class SessionController(QObject): # ---------- CRUD ---------- def new_session(self) -> None: - dialog = SessionDialog(self.key_manager, parent=self._parent_widget) + dialog = SessionDialog(self.key_manager, parent=self._parent_widget, session_store=self.session_store) 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) + dialog = SessionDialog( + self.key_manager, session=session, parent=self._parent_widget, session_store=self.session_store, + ) if dialog.exec(): self.session_store.update(dialog.result_session()) self.reload() @@ -295,7 +293,7 @@ class SessionController(QObject): d'ajuster avant d'enregistrer (comme une connexion rapide).""" dialog = SessionDialog( self.key_manager, session=session_from_openssh_host(host), parent=self._parent_widget, - title=f"Importer « {host.alias} » comme session", + title=f"Importer « {host.alias} » comme session", session_store=self.session_store, ) if dialog.exec(): self.session_store.add(dialog.result_session()) diff --git a/app/dialogs/session_dialog.py b/app/dialogs/session_dialog.py index 8b2465c..f04f35d 100644 --- a/app/dialogs/session_dialog.py +++ b/app/dialogs/session_dialog.py @@ -4,25 +4,33 @@ from __future__ import annotations from PySide6.QtWidgets import ( QDialog, QFormLayout, QLineEdit, QSpinBox, QComboBox, QDialogButtonBox, QHBoxLayout, QPushButton, QTextEdit, QFileDialog, QWidget, QVBoxLayout, - QCheckBox, QMessageBox, + QCheckBox, QMessageBox, QGroupBox, QStackedWidget, ) -from app.session_manager import Session, AUTH_AUTO, AUTH_KEY, AUTH_PASSWORD, AUTH_AGENT +from app.session_manager import ( + Session, SessionStore, BastionConfig, AUTH_AUTO, AUTH_KEY, AUTH_PASSWORD, AUTH_AGENT, + BASTION_NONE, BASTION_SESSION, BASTION_OPENSSH, BASTION_MANUAL, +) from app.key_manager import KeyManager +from app.openssh.config import load_openssh_hosts class SessionDialog(QDialog): """Formulaire de session. En mode `connect_label`, il sert de connexion rapide : on saisit les infos et on se connecte directement, l'enregistrement etant optionnel.""" - def __init__(self, key_manager: KeyManager, session: Session | None = None, - parent=None, connect_label: str = "", title: str = ""): + def __init__( + self, key_manager: KeyManager, session: Session | None = None, + parent=None, connect_label: str = "", title: str = "", + session_store: SessionStore | None = None, + ): super().__init__(parent) self.key_manager = key_manager + self.session_store = session_store self.session = session or Session() self.quick_mode = bool(connect_label) self.setWindowTitle(title or ("Nouvelle session" if session is None else "Modifier la session")) - self.setMinimumWidth(420) + self.setMinimumWidth(460) layout = QVBoxLayout(self) form = QFormLayout() @@ -71,6 +79,7 @@ class SessionDialog(QDialog): form.addRow("Notes :", self.notes_edit) layout.addLayout(form) + layout.addWidget(self._build_bastion_group()) self.save_check = QCheckBox("Enregistrer dans mes sessions (optionnel)") if self.quick_mode: @@ -114,6 +123,143 @@ class SessionDialog(QDialog): self.key_combo.addItem(path, path) self.key_combo.setCurrentIndex(self.key_combo.count() - 1) + # ---------- Bastion / Jump Host ---------- + def _build_bastion_group(self) -> QGroupBox: + """Un seul saut pour l'instant (voir BastionConfig) : Aucun / une session + PySSH existante / un Host de ~/.ssh/config / une configuration manuelle.""" + bastion = self.session.bastion + group = QGroupBox("Bastion / Jump Host") + v = QVBoxLayout(group) + + self.bastion_mode_combo = QComboBox() + self.bastion_mode_combo.addItem("Aucun", BASTION_NONE) + self.bastion_mode_combo.addItem("Une session PySSH existante", BASTION_SESSION) + self.bastion_mode_combo.addItem("Un Host de ~/.ssh/config", BASTION_OPENSSH) + self.bastion_mode_combo.addItem("Configuration manuelle", BASTION_MANUAL) + v.addWidget(self.bastion_mode_combo) + + self.bastion_stack = QStackedWidget() + v.addWidget(self.bastion_stack) + + self.bastion_stack.addWidget(QWidget()) # BASTION_NONE : rien a afficher + + session_page = QWidget() + session_form = QFormLayout(session_page) + session_form.setContentsMargins(0, 4, 0, 0) + self.bastion_session_combo = QComboBox() + others = [ + s for s in (self.session_store.sessions if self.session_store else []) + if s.id != self.session.id + ] + if others: + for s in sorted(others, key=lambda s: s.name.lower()): + self.bastion_session_combo.addItem(f"{s.name} ({s.username}@{s.host})", s.id) + else: + self.bastion_session_combo.addItem("(aucune autre session enregistree)", "") + self.bastion_session_combo.setEnabled(False) + session_form.addRow("Session :", self.bastion_session_combo) + self.bastion_stack.addWidget(session_page) + + openssh_page = QWidget() + openssh_form = QFormLayout(openssh_page) + openssh_form.setContentsMargins(0, 4, 0, 0) + self.bastion_openssh_combo = QComboBox() + hosts = load_openssh_hosts() + if hosts: + for h in hosts: + self.bastion_openssh_combo.addItem(f"{h.alias} ({h.display_target})", h.alias) + else: + self.bastion_openssh_combo.addItem("(aucun host dans ~/.ssh/config)", "") + self.bastion_openssh_combo.setEnabled(False) + openssh_form.addRow("Host :", self.bastion_openssh_combo) + self.bastion_stack.addWidget(openssh_page) + + manual_page = QWidget() + manual_form = QFormLayout(manual_page) + manual_form.setContentsMargins(0, 4, 0, 0) + self.bastion_host_edit = QLineEdit(bastion.host if bastion.mode == BASTION_MANUAL else "") + self.bastion_port_spin = QSpinBox() + self.bastion_port_spin.setRange(1, 65535) + self.bastion_port_spin.setValue(bastion.port or 22) + self.bastion_user_edit = QLineEdit(bastion.username if bastion.mode == BASTION_MANUAL else "") + self.bastion_auth_combo = QComboBox() + self.bastion_auth_combo.addItem("Automatique", AUTH_AUTO) + self.bastion_auth_combo.addItem("Cle SSH choisie", AUTH_KEY) + self.bastion_auth_combo.addItem("Mot de passe (saisi a la connexion)", AUTH_PASSWORD) + self.bastion_auth_combo.addItem("Agent SSH", AUTH_AGENT) + self.bastion_key_combo = QComboBox() + for entry in self.key_manager.keys: + self.bastion_key_combo.addItem(f"{entry.name} ({entry.key_type})", entry.private_path) + self.bastion_browse_key_btn = QPushButton("Parcourir...") + bastion_key_row = QWidget() + bastion_key_layout = QHBoxLayout(bastion_key_row) + bastion_key_layout.setContentsMargins(0, 0, 0, 0) + bastion_key_layout.addWidget(self.bastion_key_combo, 1) + bastion_key_layout.addWidget(self.bastion_browse_key_btn) + + manual_form.addRow("Hote :", self.bastion_host_edit) + manual_form.addRow("Port :", self.bastion_port_spin) + manual_form.addRow("Utilisateur :", self.bastion_user_edit) + manual_form.addRow("Authentification :", self.bastion_auth_combo) + manual_form.addRow("Cle privee :", bastion_key_row) + self.bastion_stack.addWidget(manual_page) + + self.bastion_auth_combo.currentIndexChanged.connect(self._update_bastion_key_row) + self.bastion_browse_key_btn.clicked.connect(self._browse_bastion_key) + self.bastion_mode_combo.currentIndexChanged.connect(self.bastion_stack.setCurrentIndex) + + mode_idx = self.bastion_mode_combo.findData(bastion.mode) + self.bastion_mode_combo.setCurrentIndex(mode_idx if mode_idx >= 0 else 0) + self.bastion_stack.setCurrentIndex(self.bastion_mode_combo.currentIndex()) + + if bastion.mode == BASTION_SESSION: + found = self.bastion_session_combo.findData(bastion.session_id) + if found >= 0: + self.bastion_session_combo.setCurrentIndex(found) + elif bastion.mode == BASTION_OPENSSH: + found = self.bastion_openssh_combo.findData(bastion.openssh_alias) + if found >= 0: + self.bastion_openssh_combo.setCurrentIndex(found) + elif bastion.mode == BASTION_MANUAL: + auth_idx = self.bastion_auth_combo.findData(bastion.auth_type) + self.bastion_auth_combo.setCurrentIndex(auth_idx if auth_idx >= 0 else 0) + if bastion.key_path: + existing = self.bastion_key_combo.findData(bastion.key_path) + if existing >= 0: + self.bastion_key_combo.setCurrentIndex(existing) + else: + self.bastion_key_combo.addItem(bastion.key_path, bastion.key_path) + self.bastion_key_combo.setCurrentIndex(self.bastion_key_combo.count() - 1) + + self._update_bastion_key_row() + return group + + def _update_bastion_key_row(self) -> None: + is_key_auth = self.bastion_auth_combo.currentData() == AUTH_KEY + self.bastion_key_combo.setEnabled(is_key_auth) + self.bastion_browse_key_btn.setEnabled(is_key_auth) + + def _browse_bastion_key(self) -> None: + path, _ = QFileDialog.getOpenFileName(self, "Choisir une cle privee (bastion)") + if path: + self.bastion_key_combo.addItem(path, path) + self.bastion_key_combo.setCurrentIndex(self.bastion_key_combo.count() - 1) + + def _result_bastion(self) -> BastionConfig: + mode = self.bastion_mode_combo.currentData() + bastion = BastionConfig(mode=mode) + if mode == BASTION_SESSION: + bastion.session_id = self.bastion_session_combo.currentData() or "" + elif mode == BASTION_OPENSSH: + bastion.openssh_alias = self.bastion_openssh_combo.currentData() or "" + elif mode == BASTION_MANUAL: + bastion.host = self.bastion_host_edit.text().strip() + bastion.port = self.bastion_port_spin.value() + bastion.username = self.bastion_user_edit.text().strip() + bastion.auth_type = self.bastion_auth_combo.currentData() + bastion.key_path = self.bastion_key_combo.currentData() or "" + return bastion + def should_save(self) -> bool: """En connexion rapide, indique si la session doit aussi etre enregistree.""" return self.save_check.isChecked() if self.quick_mode else True @@ -137,6 +283,19 @@ class SessionDialog(QDialog): ) self.auth_combo.setFocus() return + + bastion_mode = self.bastion_mode_combo.currentData() + if bastion_mode == BASTION_SESSION and not self.bastion_session_combo.currentData(): + QMessageBox.warning(self, "Bastion incomplet", "Choisissez une session PySSH a utiliser comme bastion.") + return + if bastion_mode == BASTION_OPENSSH and not self.bastion_openssh_combo.currentData(): + QMessageBox.warning(self, "Bastion incomplet", "Choisissez un Host de ~/.ssh/config a utiliser comme bastion.") + return + if bastion_mode == BASTION_MANUAL and not self.bastion_host_edit.text().strip(): + QMessageBox.warning(self, "Bastion incomplet", "Indiquez l'adresse du bastion.") + self.bastion_host_edit.setFocus() + return + self.session.name = self.name_edit.text().strip() or self.host_edit.text().strip() self.session.group = self.group_edit.text().strip() self.session.host = self.host_edit.text().strip() @@ -145,6 +304,7 @@ class SessionDialog(QDialog): self.session.auth_type = self.auth_combo.currentData() self.session.key_path = self.key_combo.currentData() or "" self.session.notes = self.notes_edit.toPlainText() + self.session.bastion = self._result_bastion() self.accept() def result_session(self) -> Session: diff --git a/app/openssh/config.py b/app/openssh/config.py index 275351b..f4d370a 100644 --- a/app/openssh/config.py +++ b/app/openssh/config.py @@ -14,7 +14,7 @@ from pathlib import Path import paramiko -from app.session_manager import Session, AUTH_AUTO, AUTH_KEY +from app.session_manager import Session, BastionConfig, AUTH_AUTO, AUTH_KEY, BASTION_OPENSSH @dataclass @@ -87,9 +87,10 @@ def load_openssh_hosts(path: Path | None = None) -> list[OpenSSHHost]: def session_from_openssh_host(host: OpenSSHHost) -> Session: """Convertit un Host OpenSSH en Session PySSH (utilisation directe ou import). - Le ProxyJump n'est pas encore applique automatiquement (necessite le support - Bastion/Jump Host) : on le signale dans les notes plutot que de le perdre - silencieusement.""" + Si ProxyJump est present, le bastion resultant reference l'alias OpenSSH + correspondant (mode BASTION_OPENSSH) : il sera resolu et applique a la + connexion par app.bastion, tant que cet alias reste present dans + ~/.ssh/config (voir app.bastion.resolve_bastion_target).""" session = Session( name=host.alias, group="OpenSSH", @@ -105,9 +106,7 @@ def session_from_openssh_host(host: OpenSSHHost) -> Session: notes = [f"Importe depuis ~/.ssh/config (Host {host.alias})."] if host.proxy_jump: - notes.append( - f"ProxyJump detecte vers « {host.proxy_jump} » — pas encore applique " - "automatiquement (connexion directe pour l'instant)." - ) + session.bastion = BastionConfig(mode=BASTION_OPENSSH, openssh_alias=host.proxy_jump) + notes.append(f"ProxyJump vers « {host.proxy_jump} » (voir section Bastion / Jump Host).") session.notes = "\n".join(notes) return session diff --git a/app/session_manager.py b/app/session_manager.py index 796e7a7..5238d66 100644 --- a/app/session_manager.py +++ b/app/session_manager.py @@ -16,6 +16,42 @@ AUTH_KEY = "key" # cle privee explicitement choisie AUTH_PASSWORD = "password" AUTH_AGENT = "agent" +# Sources possibles d'un jump host (ProxyJump / bastion) - voir BastionConfig. +BASTION_NONE = "none" # connexion directe, sans bastion +BASTION_SESSION = "session" # reutilise une session PySSH existante (par id) +BASTION_OPENSSH = "openssh" # reutilise un Host de ~/.ssh/config (par alias) +BASTION_MANUAL = "manual" # parametres saisis directement + + +@dataclass +class BastionConfig: + """Jump host d'une session. Un seul saut pour l'instant (mode != "chain"), + mais rien n'empeche d'ajouter plus tard une liste de sauts a la place d'un + bastion unique : le mecanisme de resolution (app.bastion) traite deja ceci + comme "le saut a franchir avant la cible", independamment du nombre total. + + Jamais de mot de passe stocke ici : comme pour Session.auth_type, un bastion + en mode mot de passe le redemande a la connexion.""" + mode: str = BASTION_NONE + session_id: str = "" # mode == BASTION_SESSION + openssh_alias: str = "" # mode == BASTION_OPENSSH + host: str = "" # mode == BASTION_MANUAL + port: int = 22 + username: str = "" + auth_type: str = AUTH_AUTO + key_path: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + @staticmethod + def from_dict(data: dict) -> "BastionConfig": + b = BastionConfig() + for k in BastionConfig.__dataclass_fields__: + if k in data: + setattr(b, k, data[k]) + return b + @dataclass class Session: @@ -28,6 +64,7 @@ class Session: auth_type: str = AUTH_AUTO # auto | key | password | agent key_path: str = "" notes: str = "" + bastion: BastionConfig = field(default_factory=BastionConfig) def to_dict(self) -> dict: return asdict(self) @@ -37,6 +74,8 @@ class Session: known = {f: data.get(f) for f in Session.__dataclass_fields__ if f in data} s = Session() for k, v in known.items(): + if k == "bastion" and isinstance(v, dict): + v = BastionConfig.from_dict(v) setattr(s, k, v) return s diff --git a/app/ssh_connection.py b/app/ssh_connection.py index bfd9844..d9fe140 100644 --- a/app/ssh_connection.py +++ b/app/ssh_connection.py @@ -3,12 +3,12 @@ from __future__ import annotations import queue import threading -from pathlib import Path import paramiko from PySide6.QtCore import QThread, Signal from app.known_hosts_manager import KnownHostsStore +from app.bastion import BastionTarget from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) KNOWN_HOSTS_FILE = DATA_DIR / "known_hosts" @@ -55,6 +55,8 @@ class SSHConnection(QThread): rows: int = 24, open_shell: bool = True, open_sftp_on_connect: bool = False, + bastion_target: BastionTarget | None = None, + bastion_password: str = "", parent=None, ): super().__init__(parent) @@ -70,9 +72,12 @@ class SSHConnection(QThread): self.passphrase = passphrase self.cols = cols self.rows = rows + self.bastion_target = bastion_target + self.bastion_password = bastion_password self.client: paramiko.SSHClient | None = None self.channel: paramiko.Channel | None = None + self._bastion_client: paramiko.SSHClient | None = None self._stop = threading.Event() self._trust_event = threading.Event() @@ -132,10 +137,10 @@ class SSHConnection(QThread): self._passphrase_event.set() # ---------- mot de passe demande en secours (appele depuis le thread SSH) ---------- - def _ask_password(self) -> str | None: + def _ask_password(self, label: str | None = None) -> str | None: self._password_event.clear() self._password_result = None - self.password_prompt.emit(f"{self.username}@{self.host}") + self.password_prompt.emit(label or f"{self.username}@{self.host}") self._password_event.wait() # debloque via provide_password() depuis le thread GUI return self._password_result @@ -143,76 +148,127 @@ class SSHConnection(QThread): self._password_result = text or None self._password_event.set() + # ---------- connexion (cible ou bastion, meme logique pour les deux) ---------- + def _new_client(self) -> paramiko.SSHClient: + """Client SSH pret a etre connecte : cles d'hote systeme + notre known_hosts, + et confirmation TOFU via cette meme SSHConnection pour une cle inconnue/modifiee.""" + client = paramiko.SSHClient() + DATA_DIR.mkdir(parents=True, exist_ok=True) + # Les hotes deja approuves par la commande ssh (~/.ssh/known_hosts) sont + # reconnus : inutile de reconfirmer une empreinte que votre ssh connait deja. + # paramiko ne reecrit jamais ce fichier ; les nouvelles cles vont dans le notre. + try: + client.load_system_host_keys() + except Exception: + pass + if KNOWN_HOSTS_FILE.exists(): + client.load_host_keys(str(KNOWN_HOSTS_FILE)) + client.set_missing_host_key_policy(TrustPrompt(self)) + return client + + def _connect_host( + self, client: paramiko.SSHClient, host: str, port: int, username: str, auth_type: str, + key_path: str, password: str, passphrase: str, sock=None, label: str | None = None, + ) -> None: + """Connecte `client` a host:port, en gerant l'authentification, la demande + interactive de mot de passe de secours (mode auto) et une cle d'hote modifiee. + Utilise a l'identique pour la cible et pour un eventuel bastion.""" + label = label or f"{username}@{host}" + auto = auth_type == "auto" + connect_kwargs = dict( + hostname=host, + port=port, + username=username, + timeout=15, + # en mode auto on se comporte comme la commande ssh : agent SSH + # et cles par defaut de ~/.ssh, sans rien avoir a designer + allow_agent=(auto or auth_type == "agent"), + look_for_keys=auto, + sock=sock, + ) + if auth_type == "key": + if not key_path: + raise paramiko.SSHException( + f"{label} : aucune cle privee selectionnee. Choisissez une cle, ou " + "l'authentification automatique / par mot de passe." + ) + connect_kwargs["pkey"] = self._load_private_key(key_path, passphrase) + elif auth_type == "password": + connect_kwargs["password"] = password + elif auto and password: + connect_kwargs["password"] = password + + try: + try: + client.connect(**connect_kwargs) + except paramiko.AuthenticationException: + # Mode auto : ni l'agent ni les cles ~/.ssh n'ont convenu, + # on demande alors un mot de passe plutot que d'echouer. + if not auto or connect_kwargs.get("password"): + raise + asked = self._ask_password(label) + if not asked: + raise + connect_kwargs["password"] = asked + connect_kwargs["look_for_keys"] = False + connect_kwargs["allow_agent"] = False + client.connect(**connect_kwargs) + except paramiko.BadHostKeyException as exc: + # La cle presentee par le serveur ne correspond plus a celle enregistree + # (frequent avec des IP DHCP reattribuees a une autre machine). + accepted = self._ask_host_key_change(exc.hostname, exc.got_key, exc.expected_key) + if not accepted: + raise paramiko.SSHException( + f"Cle d'hote modifiee refusee pour {exc.hostname} (connexion annulee)." + ) from exc + self.known_hosts.replace_host_key(exc.hostname, exc.got_key) + # Corrige aussi la copie en memoire utilisee par ce client pour la reconnexion + # (load_host_keys() fusionnerait avec l'ancienne entree encore en memoire). + live_keys = client.get_host_keys() + if exc.hostname in live_keys: + del live_keys[exc.hostname] + live_keys.add(exc.hostname, exc.got_key.get_name(), exc.got_key) + client.connect(**connect_kwargs) + + def _open_bastion_channel(self) -> paramiko.Channel: + """Connexion au bastion puis canal direct-tcpip vers la cible reelle, a + passer en `sock=` de la connexion finale (mecanisme ProxyJump standard).""" + target = self.bastion_target + bastion_client = self._new_client() + label = f"{target.label} (bastion)" + try: + self._connect_host( + bastion_client, target.host, target.port, target.username, target.auth_type, + target.key_path, self.bastion_password, "", label=label, + ) + except Exception as exc: # noqa: BLE001 + try: + bastion_client.close() + except Exception: + pass + raise paramiko.SSHException(f"Bastion {target.label} : {exc}") from exc + + self._bastion_client = bastion_client + try: + return bastion_client.get_transport().open_channel( + "direct-tcpip", (self.host, self.port), ("127.0.0.1", 0), + ) + except Exception as exc: # noqa: BLE001 + raise paramiko.SSHException( + f"Bastion {target.label} : impossible d'ouvrir un canal vers " + f"{self.host}:{self.port} ({exc})" + ) from exc + # ---------- cycle de vie du thread ---------- def run(self) -> None: try: - self.client = paramiko.SSHClient() - DATA_DIR.mkdir(parents=True, exist_ok=True) - # Les hotes deja approuves par la commande ssh (~/.ssh/known_hosts) sont - # reconnus : inutile de reconfirmer une empreinte que votre ssh connait deja. - # paramiko ne reecrit jamais ce fichier ; les nouvelles cles vont dans le notre. - try: - self.client.load_system_host_keys() - except Exception: - pass - if KNOWN_HOSTS_FILE.exists(): - self.client.load_host_keys(str(KNOWN_HOSTS_FILE)) - self.client.set_missing_host_key_policy(TrustPrompt(self)) + self.client = self._new_client() - auto = self.auth_type == "auto" - connect_kwargs = dict( - hostname=self.host, - port=self.port, - username=self.username, - timeout=15, - # en mode auto on se comporte comme la commande ssh : agent SSH - # et cles par defaut de ~/.ssh, sans rien avoir a designer - allow_agent=(auto or self.auth_type == "agent"), - look_for_keys=auto, + sock = self._open_bastion_channel() if self.bastion_target is not None else None + self._connect_host( + self.client, self.host, self.port, self.username, self.auth_type, + self.key_path, self.password, self.passphrase, sock=sock, ) - if self.auth_type == "key": - if not self.key_path: - raise paramiko.SSHException( - "Aucune cle privee selectionnee. Choisissez une cle, ou l'authentification " - "automatique / par mot de passe." - ) - connect_kwargs["pkey"] = self._load_private_key(self.key_path, self.passphrase) - elif self.auth_type == "password": - connect_kwargs["password"] = self.password - elif auto and self.password: - connect_kwargs["password"] = self.password - - try: - try: - self.client.connect(**connect_kwargs) - except paramiko.AuthenticationException: - # Mode auto : ni l'agent ni les cles ~/.ssh n'ont convenu, - # on demande alors un mot de passe plutot que d'echouer. - if not auto or connect_kwargs.get("password"): - raise - password = self._ask_password() - if not password: - raise - connect_kwargs["password"] = password - connect_kwargs["look_for_keys"] = False - connect_kwargs["allow_agent"] = False - self.client.connect(**connect_kwargs) - except paramiko.BadHostKeyException as exc: - # La cle presentee par le serveur ne correspond plus a celle enregistree - # (frequent avec des IP DHCP reattribuees a une autre machine). - accepted = self._ask_host_key_change(exc.hostname, exc.got_key, exc.expected_key) - if not accepted: - raise paramiko.SSHException( - f"Cle d'hote modifiee refusee pour {exc.hostname} (connexion annulee)." - ) from exc - self.known_hosts.replace_host_key(exc.hostname, exc.got_key) - # Corrige aussi la copie en memoire utilisee par ce client pour la reconnexion - # (load_host_keys() fusionnerait avec l'ancienne entree encore en memoire). - live_keys = self.client.get_host_keys() - if exc.hostname in live_keys: - del live_keys[exc.hostname] - live_keys.add(exc.hostname, exc.got_key.get_name(), exc.got_key) - self.client.connect(**connect_kwargs) # Ouvert ici (dans le thread) et non depuis l'interface : open_sftp() fait un # aller-retour reseau qui figerait la fenetre s'il etait appele au clic. @@ -341,3 +397,8 @@ class SSHConnection(QThread): self.client.close() except Exception: pass + try: + if self._bastion_client: + self._bastion_client.close() + except Exception: + pass diff --git a/app/ui/tabs.py b/app/ui/tabs.py index f7c0f30..25e0681 100644 --- a/app/ui/tabs.py +++ b/app/ui/tabs.py @@ -215,9 +215,14 @@ class BaseTerminalTab(QWidget): 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): + def __init__( + self, session: Session, secret: str, bastion_target=None, bastion_password: str = "", + parent=None, + ): super().__init__(f"○ Connexion a {session.username}@{session.host}...", parent) self.session = session + self._bastion_target = bastion_target + self._bastion_password = bastion_password self.sftp_panel: SFTPPanel | None = None self.sftp_toggle = QToolButton() @@ -254,6 +259,8 @@ class SessionTabWidget(BaseTerminalTab): # comme MobaXterm : le navigateur de fichiers est la des l'ouverture, # pas cache derriere une action supplementaire open_sftp_on_connect=True, + bastion_target=self._bastion_target, + bastion_password=self._bastion_password, ) connection.sftp_ready.connect(self._on_sftp_ready) connection.password_prompt.connect(self._on_password_prompt) @@ -334,7 +341,10 @@ class SessionTabWidget(BaseTerminalTab): class SFTPTabWidget(QWidget): """Onglet SFTP autonome (facon WinSCP) : double panneau de fichiers, sans terminal.""" - def __init__(self, session: Session, secret: str, parent=None): + def __init__( + self, session: Session, secret: str, bastion_target=None, bastion_password: str = "", + parent=None, + ): super().__init__(parent) self.session = session self.sftp_panel: SFTPPanel | None = None @@ -368,6 +378,8 @@ class SFTPTabWidget(QWidget): passphrase=secret if session.auth_type == AUTH_KEY else "", open_shell=False, open_sftp_on_connect=True, + bastion_target=bastion_target, + bastion_password=bastion_password, ) self.connection.sftp_ready.connect(self._on_sftp_ready) self.connection.password_prompt.connect(self._on_password_prompt) diff --git a/tests/test_bastion.py b/tests/test_bastion.py new file mode 100644 index 0000000..812573f --- /dev/null +++ b/tests/test_bastion.py @@ -0,0 +1,178 @@ +"""ProxyJump / Bastion : resolution (app.bastion) pour les 3 sources (session, +OpenSSH, manuel), et etablissement du canal direct-tcpip par SSHConnection via +un bastion entierement mocke (aucun reseau reel).""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +from PySide6.QtWidgets import QApplication + +from app.session_manager import ( + SessionStore, Session, BastionConfig, + AUTH_AUTO, AUTH_KEY, + BASTION_NONE, BASTION_SESSION, BASTION_OPENSSH, BASTION_MANUAL, +) +from app.bastion import resolve_bastion_target, BastionResolutionError, BastionTarget + +app = QApplication.instance() or QApplication([]) + +tmp = Path(tempfile.mkdtemp()) + +print("=== 1. Aucun bastion -> None ===") +assert resolve_bastion_target(BastionConfig(mode=BASTION_NONE)) is None +assert resolve_bastion_target(None) is None +print(" OK") + +print() +print("=== 2. Bastion manuel ===") +manual = BastionConfig( + mode=BASTION_MANUAL, host="bastion.example.com", port=2222, + username="jump", auth_type=AUTH_KEY, key_path="/tmp/key", +) +target = resolve_bastion_target(manual) +print(" target:", target) +assert target == BastionTarget("bastion.example.com", 2222, "jump", AUTH_KEY, "/tmp/key") +print(" OK") + +print() +print("=== 3. Bastion manuel incomplet -> erreur claire ===") +try: + resolve_bastion_target(BastionConfig(mode=BASTION_MANUAL, host="")) + ok = False +except BastionResolutionError: + ok = True +assert ok +print(" OK") + +print() +print("=== 4. Bastion = session PySSH existante ===") +store = SessionStore(path=tmp / "sessions.json") +bastion_session = Session(name="bastion-srv", host="10.0.0.1", port=22, username="root", auth_type=AUTH_AUTO) +store.add(bastion_session) +target = resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id=bastion_session.id), store) +print(" target:", target) +assert target.host == "10.0.0.1" and target.username == "root" +print(" OK") + +print() +print("=== 5. Bastion = session supprimee -> erreur claire ===") +try: + resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id="inconnu"), store) + ok = False +except BastionResolutionError as exc: + ok = True + print(" message:", exc) +assert ok +print(" OK") + +print() +print("=== 6. Bastion sans registre de sessions fourni -> erreur claire (pas de crash) ===") +try: + resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id="x"), None) + ok = False +except BastionResolutionError: + ok = True +assert ok +print(" OK") + +print() +print("=== 7. Bastion = Host OpenSSH ===") +config_path = tmp / "sshconfig" +config_path.write_text( + "Host bastion\n HostName bastion.example.com\n User jump\n IdentityFile ~/.ssh/jumpkey\n", + encoding="utf-8", +) +with patch("app.openssh.config.default_config_path", return_value=config_path): + target = resolve_bastion_target(BastionConfig(mode=BASTION_OPENSSH, openssh_alias="bastion")) +print(" target:", target) +assert target.host == "bastion.example.com" +assert target.username == "jump" +assert target.auth_type == AUTH_KEY +print(" OK") + +print() +print("=== 8. Bastion = alias OpenSSH introuvable -> erreur claire ===") +with patch("app.openssh.config.default_config_path", return_value=config_path): + try: + resolve_bastion_target(BastionConfig(mode=BASTION_OPENSSH, openssh_alias="n-existe-pas")) + ok = False + except BastionResolutionError: + ok = True +assert ok +print(" OK") + +print() +print("=== 9. SSHConnection : ouverture du canal direct-tcpip via un bastion mocke ===") +from app.ssh_connection import SSHConnection + +bastion_target = BastionTarget(host="bastion.example.com", port=22, username="jump", auth_type=AUTH_AUTO) +conn = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target) + +fake_channel = MagicMock(name="direct-tcpip-channel") +fake_transport = MagicMock() +fake_transport.open_channel.return_value = fake_channel + +with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient: + fake_bastion_client = MagicMock() + fake_bastion_client.get_transport.return_value = fake_transport + FakeSSHClient.return_value = fake_bastion_client + + channel = conn._open_bastion_channel() + +hostname_used = fake_bastion_client.connect.call_args.kwargs.get("hostname") +print(" connect() vers le bastion, hostname =", hostname_used) +assert hostname_used == "bastion.example.com" +assert fake_transport.open_channel.call_args.args[0] == "direct-tcpip" +assert fake_transport.open_channel.call_args.args[1] == ("10.20.20.15", 22) +assert channel is fake_channel +assert conn._bastion_client is fake_bastion_client +print(" OK") + +print() +print("=== 10. Cible connectee avec sock= (pas de socket TCP direct) ===") +conn2 = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target) +with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient: + fake_bastion_client2 = MagicMock() + fake_target_client2 = MagicMock() + fake_transport2 = MagicMock() + fake_transport2.open_channel.return_value = fake_channel + fake_bastion_client2.get_transport.return_value = fake_transport2 + # comme dans run() : le 1er SSHClient() cree est celui de la cible (self.client, + # assigne avant l'appel a _open_bastion_channel), le 2e celui du bastion (cree a + # l'interieur de _open_bastion_channel) + FakeSSHClient.side_effect = [fake_target_client2, fake_bastion_client2] + + conn2.client = conn2._new_client() + sock = conn2._open_bastion_channel() + conn2._connect_host( + conn2.client, conn2.host, conn2.port, conn2.username, conn2.auth_type, + conn2.key_path, conn2.password, conn2.passphrase, sock=sock, + ) + +print(" cible connect() sock=canal du bastion :", fake_target_client2.connect.call_args.kwargs.get("sock") is fake_channel) +assert fake_target_client2.connect.call_args.kwargs.get("sock") is fake_channel +print(" OK") + +print() +print("=== 11. Echec de connexion au bastion -> message clair, jamais de fuite d'exception brute ===") +conn3 = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target) +with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient: + fake_bastion_client3 = MagicMock() + fake_bastion_client3.connect.side_effect = OSError("Connexion refusee") + FakeSSHClient.return_value = fake_bastion_client3 + try: + conn3._open_bastion_channel() + ok = False + except Exception as exc: # noqa: BLE001 + ok = True + message = str(exc) + print(" message:", message) + assert "Bastion" in message +assert ok +print(" OK") + +print() +print("BASTION / PROXYJUMP : TOUS LES TESTS OK") diff --git a/tests/test_session_dialog_bastion.py b/tests/test_session_dialog_bastion.py new file mode 100644 index 0000000..0c8f10a --- /dev/null +++ b/tests/test_session_dialog_bastion.py @@ -0,0 +1,126 @@ +"""Editeur de session : section Bastion / Jump Host (construction, pre-remplissage +a l'edition, et resultat renvoye pour chacun des 4 modes).""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +import tempfile +from pathlib import Path +from unittest.mock import patch + +from PySide6.QtWidgets import QApplication + +from app.session_manager import ( + SessionStore, Session, BastionConfig, + BASTION_NONE, BASTION_SESSION, BASTION_OPENSSH, BASTION_MANUAL, AUTH_KEY, +) +from app.key_manager import KeyManager +from app.dialogs.session_dialog import SessionDialog + +app = QApplication.instance() or QApplication([]) + +tmp = Path(tempfile.mkdtemp()) +key_manager = KeyManager(registry_path=tmp / "keys.json") +store = SessionStore(path=tmp / "sessions.json") +bastion_session = Session(name="bastion-srv", host="10.0.0.1", username="root") +store.add(bastion_session) +target_session = Session(name="cible", host="10.20.20.15", username="root") +store.add(target_session) + +config_path = tmp / "sshconfig" +config_path.write_text("Host bastion-openssh\n HostName 1.2.3.4\n User jump\n", encoding="utf-8") + +print("=== 1. Par defaut : mode Aucun ===") +dlg = SessionDialog(key_manager, session=Session(), session_store=store) +assert dlg.bastion_mode_combo.currentData() == BASTION_NONE +assert dlg._result_bastion().mode == BASTION_NONE +dlg.close() +print(" OK") + +print() +print("=== 2. La session en cours d'edition n'apparait pas dans sa propre liste de bastions ===") +dlg = SessionDialog(key_manager, session=target_session, session_store=store) +ids_proposes = [dlg.bastion_session_combo.itemData(i) for i in range(dlg.bastion_session_combo.count())] +print(" sessions proposees comme bastion:", ids_proposes) +assert target_session.id not in ids_proposes +assert bastion_session.id in ids_proposes +dlg.close() +print(" OK") + +print() +print("=== 3. Mode Session : selection puis resultat ===") +dlg = SessionDialog(key_manager, session=Session(), session_store=store) +idx = dlg.bastion_mode_combo.findData(BASTION_SESSION) +dlg.bastion_mode_combo.setCurrentIndex(idx) +idx_session = dlg.bastion_session_combo.findData(bastion_session.id) +dlg.bastion_session_combo.setCurrentIndex(idx_session) +result = dlg._result_bastion() +print(" resultat:", result) +assert result.mode == BASTION_SESSION +assert result.session_id == bastion_session.id +dlg.close() +print(" OK") + +print() +print("=== 4. Mode OpenSSH : liste chargee depuis ~/.ssh/config, puis resultat ===") +with patch("app.openssh.config.default_config_path", return_value=config_path): + dlg = SessionDialog(key_manager, session=Session(), session_store=store) + idx = dlg.bastion_mode_combo.findData(BASTION_OPENSSH) + dlg.bastion_mode_combo.setCurrentIndex(idx) + idx_host = dlg.bastion_openssh_combo.findData("bastion-openssh") + assert idx_host >= 0, "l'alias detecte doit apparaitre dans la liste" + dlg.bastion_openssh_combo.setCurrentIndex(idx_host) + result = dlg._result_bastion() +print(" resultat:", result) +assert result.mode == BASTION_OPENSSH +assert result.openssh_alias == "bastion-openssh" +dlg.close() +print(" OK") + +print() +print("=== 5. Mode manuel : champs -> resultat ===") +dlg = SessionDialog(key_manager, session=Session(), session_store=store) +idx = dlg.bastion_mode_combo.findData(BASTION_MANUAL) +dlg.bastion_mode_combo.setCurrentIndex(idx) +dlg.bastion_host_edit.setText("bastion.example.com") +dlg.bastion_port_spin.setValue(2222) +dlg.bastion_user_edit.setText("jump") +result = dlg._result_bastion() +print(" resultat:", result) +assert result.mode == BASTION_MANUAL +assert result.host == "bastion.example.com" +assert result.port == 2222 +assert result.username == "jump" +dlg.close() +print(" OK") + +print() +print("=== 6. Edition d'une session avec bastion deja configure : pre-remplissage correct ===") +pre_filled = Session( + name="avec-bastion", host="10.20.20.15", username="root", + bastion=BastionConfig(mode=BASTION_MANUAL, host="old-bastion", port=2200, username="olduser"), +) +dlg = SessionDialog(key_manager, session=pre_filled, session_store=store) +print(" mode pre-selectionne:", dlg.bastion_mode_combo.currentData()) +print(" page affichee (stack index):", dlg.bastion_stack.currentIndex()) +assert dlg.bastion_mode_combo.currentData() == BASTION_MANUAL +assert dlg.bastion_stack.currentIndex() == dlg.bastion_mode_combo.currentIndex() +assert dlg.bastion_host_edit.text() == "old-bastion" +assert dlg.bastion_port_spin.value() == 2200 +assert dlg.bastion_user_edit.text() == "olduser" +dlg.close() +print(" OK") + +print() +print("=== 7. Cle bastion visible seulement en authentification par cle ===") +dlg = SessionDialog(key_manager, session=Session(), session_store=store) +idx = dlg.bastion_mode_combo.findData(BASTION_MANUAL) +dlg.bastion_mode_combo.setCurrentIndex(idx) +assert not dlg.bastion_key_combo.isEnabled() +idx_key = dlg.bastion_auth_combo.findData(AUTH_KEY) +dlg.bastion_auth_combo.setCurrentIndex(idx_key) +assert dlg.bastion_key_combo.isEnabled() +dlg.close() +print(" OK") + +print() +print("SESSION DIALOG - BASTION : TOUS LES TESTS OK")