commit b977842186cb4ad7a9d86ba0d393f28d4b43ae67 Author: snoopy Date: Fri Sep 4 06:50:10 2026 +0200 Client SSH/SFTP personnel pour Windows Reunit en un seul outil les fonctions utiles de MobaXterm, Termius et Bitvise, sans compte ni service tiers, avec toutes les donnees en local. - Terminaux multi-onglets : sessions SSH et terminaux Windows locaux (PowerShell / cmd) via la pseudo-console native ConPTY - Gestionnaire de fichiers SFTP facon WinSCP : double panneau, colonnes triables, glisser-deposer, renommage, permissions, file de transferts. Ouvrable sur une connexion en cours sans creer de favori - Gestion des cles SSH : generation Ed25519/RSA, import, export - Gestion du fichier known_hosts, avec remplacement guide d'une cle d'hote modifiee (cas frequent des serveurs en DHCP) - Trois themes (clair, bleu nuit, noir), commutables a chaud - Authentification automatique reprenant le comportement de la commande ssh (agent SSH et cles ~/.ssh), mot de passe demande seulement en secours Aucun mot de passe n'est stocke sur le disque. Co-Authored-By: Claude Opus 5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d5cceed --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +venv/ +__pycache__/ +*.pyc + +# generes localement (chemins absolus / dependants du poste) +*.lnk +resources/app.ico +data/settings.json + +# artefacts de construction PyInstaller +build/ +dist/ +data/sessions.json +data/keys.json +data/known_hosts +data/keys/* +!data/keys/.gitkeep diff --git a/PySSHClient.spec b/PySSHClient.spec new file mode 100644 index 0000000..a8882ed --- /dev/null +++ b/PySSHClient.spec @@ -0,0 +1,65 @@ +# -*- mode: python ; coding: utf-8 -*- +"""Recette PyInstaller : construit un dossier portable dans dist/PySSH Client/. + +Choix "onedir" (dossier) plutot que "onefile" : avec PySide6, un executable unique +doit decompresser ~200 Mo a chaque lancement, ce qui ajoute plusieurs secondes +d'attente a chaque demarrage. Le dossier demarre instantanement et se copie tout +aussi bien sur une cle USB. +""" + +block_cipher = None + +a = Analysis( + ["main.py"], + pathex=[], + binaries=[], + # la feuille de style est lue au runtime : elle doit voyager avec l'application + datas=[("resources/theme.qss", "resources")], + hiddenimports=["winpty"], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + # modules inutiles ici : les exclure allege sensiblement le resultat + excludes=[ + "tkinter", "PySide6.QtQml", "PySide6.QtQuick", "PySide6.QtQuick3D", + "PySide6.QtWebEngineCore", "PySide6.QtWebEngineWidgets", "PySide6.Qt3DCore", + "PySide6.QtMultimedia", "PySide6.QtCharts", "PySide6.QtDataVisualization", + "PySide6.QtBluetooth", "PySide6.QtPositioning", "PySide6.QtSensors", + ], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher, + noarchive=False, +) + +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) + +exe = EXE( + pyz, + a.scripts, + [], + exclude_binaries=True, + name="PySSH Client", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=False, + console=False, # application graphique : pas de fenetre de console + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, + icon="resources/app.ico", +) + +coll = COLLECT( + exe, + a.binaries, + a.zipfiles, + a.datas, + strip=False, + upx=False, + upx_exclude=[], + name="PySSH Client", +) diff --git a/README.md b/README.md new file mode 100644 index 0000000..d20bb54 --- /dev/null +++ b/README.md @@ -0,0 +1,123 @@ +# PySSH Client + +Client SSH/SFTP personnel (type MobaXterm/Termius/Bitvise), 100% local, sans compte ni service tiers. + +## Fonctionnalites + +- Sessions SSH multiples dans des onglets, avec terminal ANSI colore (theme "bleu nuit") +- Terminaux locaux Windows (PowerShell / Invite de commandes) accessibles en un clic, **sans avoir besoin d'une session enregistree** โ€” pratique pour un usage ponctuel, comme le terminal Windows natif +- Carnet d'adresses de serveurs (JSON local, `data/sessions.json`) organise par dossiers, connexion en un clic +- Gestion des cles SSH : generation (Ed25519 / RSA), import de cles existantes, export (cle publique/privee) +- Client SFTP integre facon WinSCP : double panneau local/distant avec colonnes triables (nom, taille, date, droits), + glisser-deposer entre les panneaux, renommage (F2), suppression (Suppr), creation de dossier, permissions distantes + (chmod), et file de transferts avec progression. **Aucun favori n'est necessaire pour l'utiliser** โ€” voir ci-dessous +- **SFTP sur la connexion en cours** : depuis n'importe quel terminal ou vous avez tape `ssh user@hote` a la main, le + bouton "๐Ÿ“ SFTP sur cette connexion" lit l'hote, l'utilisateur et le port dans la commande saisie et ouvre le + gestionnaire de fichiers dessus. Pense pour les serveurs de test en DHCP, qu'on n'a aucune envie d'enregistrer +- **Authentification automatique** (mode par defaut) : comme la commande `ssh`, l'application utilise l'agent SSH et les + cles de `~/.ssh` sans que vous ayez a designer quoi que ce soit, et ne demande un mot de passe qu'en dernier recours +- Verification des empreintes de cles d'hote (TOFU) avant premiere connexion, comme un client SSH classique +- Gestionnaire des hotes connus (menu "Hotes connus" ou bouton "โš™" du panneau lateral) : si une adresse en DHCP change de machine, la cle d'hote presentee ne correspond plus a celle enregistree โ€” l'application le detecte et propose de remplacer l'ancienne entree directement au moment de la connexion, sans avoir a vider le fichier `known_hosts` a la main. Le gestionnaire permet aussi un nettoyage manuel (par hote ou en une fois). + +Aucun mot de passe n'est stocke sur le disque : l'authentification par mot de passe est demandee a chaque connexion, et les passphrases de cles sont demandees uniquement en memoire au moment de la connexion. + +## Installation (Windows) + +Necessite Python 3.10+ installe sur le poste (verifiable avec `py -3 --version`). + +```bash +setup.bat +``` + +Ce script cree un environnement virtuel local (`venv/`) et y installe les dependances (PySide6, paramiko, pyte, cryptography, pywinpty) โ€” rien n'est installe globalement sur le poste. + +## Lancement + +Trois facons, au choix : + +- **Raccourci Windows** (recommande) : double-cliquez sur `PySSH Client.lnk` a la racine du projet. + Il lance l'application via `pythonw.exe`, donc **sans fenetre de console noire**. Vous pouvez le + copier sur le Bureau, ou l'epingler a la barre des taches. + Pour le (re)creer โ€” apres un deplacement du projet, ou sur un autre poste : + + ```bash + powershell -ExecutionPolicy Bypass -File tools\create_shortcut.ps1 -Desktop + ``` + + (`-Desktop` et `-StartMenu` sont optionnels et ajoutent une copie a ces emplacements.) + +- **En ligne de commande**, sans console residuelle : + + ```bash + venv\Scripts\pythonw.exe main.py + ``` + +- **`run.bat`** : garde une fenetre de console ouverte, ce qui est pratique pour voir les erreurs + si l'application refuse de demarrer. + +## Version portable (sans Python installe) + +```bash +build_exe.bat +``` + +Produit un dossier autonome `dist\PySSH Client\` contenant l'executable et tout ce dont il a besoin, +interpreteur Python compris. Copiez ce dossier ou vous voulez (y compris sur une cle USB) : il +fonctionne sur n'importe quel poste Windows, **meme sans Python installe**. + +Le format retenu est un dossier plutot qu'un fichier `.exe` unique : avec PySide6, un executable +unique doit decompresser environ 200 Mo a chaque lancement, ce qui ajoute plusieurs secondes +d'attente a chaque demarrage. Le dossier demarre immediatement. + +**Ou sont stockees les donnees ?** A cote de l'executable, dans son propre sous-dossier `data\` +(mode portable : vos sessions et cles suivent la cle USB). Si cet emplacement n'est pas accessible +en ecriture โ€” installation dans `Program Files`, support en lecture seule โ€” l'application bascule +automatiquement sur `%APPDATA%\PySSHClient\`. + +## Premiers pas + +1. **Terminal local immediat** : bouton "๐Ÿ’ป Terminal" dans le panneau lateral (ou `Ctrl+Shift+T`), ou double-clic sur "PowerShell"/"Invite de commandes" dans la section "Terminaux locaux" de l'arborescence โ€” aucune session a creer. +2. **Nouvelle session SSH** (bouton "+ Session" ou `Ctrl+N`) : renseignez l'hote, le port, l'utilisateur et le mode d'authentification. +3. **Cles SSH** (menu "Cles SSH > Gestionnaire de cles") : generez une nouvelle paire de cles ou importez-en une existante, puis selectionnez-la dans le formulaire de session. +4. Double-cliquez sur une session dans l'arborescence pour ouvrir un onglet terminal SSH. +5. **SFTP โ€” trois facons d'y arriver**, selon la situation : + - *Vous avez tape `ssh user@hote` dans un terminal* : bouton **"๐Ÿ“ SFTP sur cette connexion"** dans la barre de + l'onglet. L'hote et l'utilisateur sont deduits de la commande, le formulaire est pre-rempli. Rien a enregistrer. + - *Connexion ponctuelle* : bouton **"๐Ÿ“ SFTP"** du panneau lateral, ou `Ctrl+Shift+F`. La case "Enregistrer dans mes + sessions" est decochee par defaut. + - *Session enregistree* : clic droit > **"Ouvrir en SFTP"** (onglet dedie). Dans une session SSH ouverte par + l'application, le panneau de fichiers est deja affiche a cote du terminal, comme dans MobaXterm. + + Dans tous les cas : glisser-deposer entre les deux panneaux, F2 renomme, Suppr supprime, F5 actualise, et le clic + droit sur le panneau distant donne acces aux permissions (chmod). + +## Structure du projet + +``` +main.py point d'entree +app/ + theme.py feuille de style "bleu nuit" + icone d'application + session_manager.py base de sessions (JSON) + key_manager.py generation/import/export de cles SSH + ssh_connection.py connexion SSH en arriere-plan (paramiko + QThread) + known_hosts_manager.py gestion du fichier known_hosts (lister/retirer/remplacer une cle d'hote) + local_terminal.py terminal local Windows (PowerShell/cmd via pywinpty) + 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) +resources/ + theme.qss feuille de style Qt "bleu nuit" +data/ + sessions.json vos serveurs enregistres (non versionne) + keys.json registre des cles connues (non versionne) + keys/ cles generees par l'application (non versionne) + known_hosts empreintes d'hotes approuvees (non versionne) +``` + +## Limites connues / pistes d'amelioration + +- L'emulation de terminal couvre les usages courants (bash, vim, htop...) mais n'est pas un xterm parfaitement complet. +- Le terminal local repose sur `pywinpty` (pseudo-console Windows). Si un onglet local reste bloque sur un ecran vide, tapez une touche : le rafraichissement de la console Windows peut etre paresseux dans certains contextes (notamment les sessions Windows non interactives) et se declenche a la premiere frappe. +- Le SFTP ne propose pas encore l'edition de fichier distant en place (telecharger / modifier / renvoyer a la main). +- Pas encore d'export/import global du carnet de sessions (le fichier `data/sessions.json` peut etre copie manuellement). diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/dialogs/__init__.py b/app/dialogs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/dialogs/connect_prompt.py b/app/dialogs/connect_prompt.py new file mode 100644 index 0000000..94963fd --- /dev/null +++ b/app/dialogs/connect_prompt.py @@ -0,0 +1,42 @@ +"""Dialogues utilises au moment de la connexion : saisie de secret et confirmation de cle d'hote.""" +from __future__ import annotations + +from PySide6.QtWidgets import QInputDialog, QLineEdit, QMessageBox, QWidget + + +def ask_secret(parent: QWidget, title: str, label: str) -> tuple[str, bool]: + text, ok = QInputDialog.getText(parent, title, label, QLineEdit.Password) + return text, ok + + +def confirm_host_key(parent: QWidget, hostname: str, key_type: str, fingerprint: str) -> bool: + msg = ( + f"L'empreinte de la cle du serveur '{hostname}' n'est pas connue.\n\n" + f"Type : {key_type}\n" + f"Empreinte : {fingerprint}\n\n" + "Verifiez cette empreinte aupres de l'administrateur du serveur si vous n'etes pas certain.\n" + "Faire confiance a cette cle et se connecter ?" + ) + answer = QMessageBox.question( + parent, "Nouvelle cle d'hote detectee", msg, + QMessageBox.Yes | QMessageBox.No, QMessageBox.No, + ) + return answer == QMessageBox.Yes + + +def confirm_host_key_change(parent: QWidget, hostname: str, key_type: str, old_fingerprint: str, new_fingerprint: str) -> bool: + msg = ( + f"La cle presentee par le serveur '{hostname}' ne correspond plus a celle enregistree.\n\n" + f"Ancienne empreinte : {old_fingerprint}\n" + f"Nouvelle empreinte : {new_fingerprint}\n\n" + "C'est generalement normal si ce serveur a une adresse IP attribuee par DHCP et " + "qu'elle a ete reutilisee par une autre machine, ou si le serveur a ete reinstalle. " + "Mais cela peut aussi indiquer une interception de la connexion (attaque de type " + "'man-in-the-middle') : verifiez la nouvelle empreinte si vous avez un doute.\n\n" + "Remplacer l'ancienne cle enregistree et se connecter ?" + ) + answer = QMessageBox.warning( + parent, "Cle d'hote modifiee", msg, + QMessageBox.Yes | QMessageBox.No, QMessageBox.No, + ) + return answer == QMessageBox.Yes diff --git a/app/dialogs/key_dialog.py b/app/dialogs/key_dialog.py new file mode 100644 index 0000000..cb212cc --- /dev/null +++ b/app/dialogs/key_dialog.py @@ -0,0 +1,198 @@ +"""Gestionnaire de cles SSH : generation, import, export.""" +from __future__ import annotations + +from pathlib import Path + +from PySide6.QtCore import Qt +from PySide6.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QListWidget, QListWidgetItem, QPushButton, + QMessageBox, QFileDialog, QFormLayout, QLineEdit, QComboBox, QSpinBox, + QDialogButtonBox, QApplication, QLabel, +) + +from app.key_manager import KeyManager, KEY_TYPE_ED25519, KEY_TYPE_RSA + + +class GenerateKeyDialog(QDialog): + def __init__(self, parent=None): + super().__init__(parent) + self.setWindowTitle("Generer une nouvelle cle SSH") + layout = QVBoxLayout(self) + form = QFormLayout() + + self.name_edit = QLineEdit() + self.type_combo = QComboBox() + self.type_combo.addItem("Ed25519 (recommande)", KEY_TYPE_ED25519) + self.type_combo.addItem("RSA", KEY_TYPE_RSA) + self.bits_spin = QSpinBox() + self.bits_spin.setRange(2048, 8192) + self.bits_spin.setSingleStep(1024) + self.bits_spin.setValue(4096) + self.passphrase_edit = QLineEdit() + self.passphrase_edit.setEchoMode(QLineEdit.Password) + self.comment_edit = QLineEdit() + + form.addRow("Nom :", self.name_edit) + form.addRow("Type :", self.type_combo) + form.addRow("Taille (RSA) :", self.bits_spin) + form.addRow("Passphrase (optionnel) :", self.passphrase_edit) + form.addRow("Commentaire :", self.comment_edit) + layout.addLayout(form) + + self.type_combo.currentIndexChanged.connect( + lambda: self.bits_spin.setEnabled(self.type_combo.currentData() == KEY_TYPE_RSA) + ) + self.bits_spin.setEnabled(False) + + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + def values(self) -> dict: + return dict( + name=self.name_edit.text().strip() or "cle_ssh", + key_type=self.type_combo.currentData(), + rsa_bits=self.bits_spin.value(), + passphrase=self.passphrase_edit.text(), + comment=self.comment_edit.text().strip(), + ) + + +class KeyManagerDialog(QDialog): + def __init__(self, key_manager: KeyManager, parent=None): + super().__init__(parent) + self.key_manager = key_manager + self.setWindowTitle("Gestionnaire de cles SSH") + self.setMinimumSize(560, 360) + + layout = QVBoxLayout(self) + self.list = QListWidget() + layout.addWidget(self.list) + + self.info_label = QLabel("") + self.info_label.setWordWrap(True) + layout.addWidget(self.info_label) + + btn_row = QHBoxLayout() + self.generate_btn = QPushButton("Generer...") + self.import_btn = QPushButton("Importer...") + self.export_pub_btn = QPushButton("Exporter cle publique...") + self.export_priv_btn = QPushButton("Exporter cle privee...") + self.copy_pub_btn = QPushButton("Copier cle publique") + self.remove_btn = QPushButton("Supprimer") + for b in (self.generate_btn, self.import_btn, self.export_pub_btn, + self.export_priv_btn, self.copy_pub_btn, self.remove_btn): + btn_row.addWidget(b) + layout.addLayout(btn_row) + + close_row = QHBoxLayout() + close_row.addStretch() + close_btn = QPushButton("Fermer") + close_btn.clicked.connect(self.accept) + close_row.addWidget(close_btn) + layout.addLayout(close_row) + + self.generate_btn.clicked.connect(self.generate_key) + self.import_btn.clicked.connect(self.import_key) + self.export_pub_btn.clicked.connect(self.export_public) + self.export_priv_btn.clicked.connect(self.export_private) + self.copy_pub_btn.clicked.connect(self.copy_public) + self.remove_btn.clicked.connect(self.remove_key) + self.list.currentItemChanged.connect(self._update_info) + + self.refresh() + + def refresh(self) -> None: + self.list.clear() + for entry in self.key_manager.keys: + item = QListWidgetItem(f"{entry.name} [{entry.key_type}]") + item.setData(Qt.UserRole, entry.id) + self.list.addItem(item) + self.info_label.setText("") + + def _current_entry(self): + item = self.list.currentItem() + if not item: + return None + key_id = item.data(Qt.UserRole) + return next((k for k in self.key_manager.keys if k.id == key_id), None) + + def _update_info(self) -> None: + entry = self._current_entry() + if not entry: + self.info_label.setText("") + return + self.info_label.setText( + f"Chemin : {entry.private_path}\nEmpreinte : {entry.fingerprint}\n\n{entry.public_key_text}" + ) + + def generate_key(self) -> None: + dlg = GenerateKeyDialog(self) + if dlg.exec() != QDialog.Accepted: + return + vals = dlg.values() + try: + self.key_manager.generate_key(**vals) + except Exception as exc: # noqa: BLE001 + QMessageBox.critical(self, "Erreur", str(exc)) + return + self.refresh() + + def import_key(self) -> None: + priv_path, _ = QFileDialog.getOpenFileName(self, "Selectionner la cle privee a importer") + if not priv_path: + return + name = Path(priv_path).stem + try: + self.key_manager.import_key(name=name, private_key_path=priv_path) + except Exception as exc: # noqa: BLE001 + QMessageBox.critical(self, "Erreur", str(exc)) + return + self.refresh() + + def export_public(self) -> None: + entry = self._current_entry() + if not entry: + return + dest, _ = QFileDialog.getSaveFileName(self, "Exporter la cle publique", f"{entry.name}.pub") + if dest: + self.key_manager.export_public_key(entry, dest) + QMessageBox.information(self, "Export", "Cle publique exportee.") + + def export_private(self) -> None: + entry = self._current_entry() + if not entry: + return + if QMessageBox.warning( + self, "Attention", + "La cle privee est sensible : quiconque y a acces peut se connecter a vos serveurs. Continuer ?", + QMessageBox.Yes | QMessageBox.No, + ) != QMessageBox.Yes: + return + dest, _ = QFileDialog.getSaveFileName(self, "Exporter la cle privee", entry.name) + if dest: + self.key_manager.export_private_key(entry, dest) + QMessageBox.information(self, "Export", "Cle privee exportee.") + + def copy_public(self) -> None: + entry = self._current_entry() + if not entry: + return + QApplication.clipboard().setText(entry.public_key_text) + QMessageBox.information(self, "Copie", "Cle publique copiee dans le presse-papiers.") + + def remove_key(self) -> None: + entry = self._current_entry() + if not entry: + return + delete_files = QMessageBox.question( + self, "Supprimer", + "Retirer aussi les fichiers de cle du disque (irreversible) ?\n" + "Non = seulement retirer de la liste.", + QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel, + ) + if delete_files == QMessageBox.Cancel: + return + self.key_manager.remove(entry.id, delete_files=(delete_files == QMessageBox.Yes)) + self.refresh() diff --git a/app/dialogs/known_hosts_dialog.py b/app/dialogs/known_hosts_dialog.py new file mode 100644 index 0000000..fd4b652 --- /dev/null +++ b/app/dialogs/known_hosts_dialog.py @@ -0,0 +1,95 @@ +"""Gestionnaire des hotes SSH connus (fichier known_hosts local).""" +from __future__ import annotations + +from PySide6.QtCore import Qt +from PySide6.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QTreeWidget, QTreeWidgetItem, QPushButton, + QMessageBox, QLabel, QAbstractItemView, +) + +from app.known_hosts_manager import KnownHostsStore + + +class KnownHostsDialog(QDialog): + def __init__(self, parent=None): + super().__init__(parent) + self.store = KnownHostsStore() + self.setWindowTitle("Gestionnaire des hotes connus") + self.setMinimumSize(560, 380) + + layout = QVBoxLayout(self) + + info = QLabel( + "Cles des serveurs approuvees lors d'une premiere connexion. Si un serveur change " + "d'adresse (ex. reattribution DHCP), l'application vous proposera automatiquement de " + "remplacer son entree au moment de la connexion. Vous pouvez aussi nettoyer manuellement ici." + ) + info.setWordWrap(True) + layout.addWidget(info) + + self.tree = QTreeWidget() + self.tree.setHeaderLabels(["Hote", "Type de cle", "Empreinte"]) + self.tree.setRootIsDecorated(False) + self.tree.setSelectionMode(QAbstractItemView.ExtendedSelection) + self.tree.setColumnWidth(0, 200) + self.tree.setColumnWidth(1, 110) + layout.addWidget(self.tree) + + self.count_label = QLabel("") + self.count_label.setStyleSheet("color: #93a4c2;") + layout.addWidget(self.count_label) + + btn_row = QHBoxLayout() + self.refresh_btn = QPushButton("Actualiser") + self.remove_btn = QPushButton("Supprimer la selection") + self.clear_btn = QPushButton("Tout effacer") + btn_row.addWidget(self.refresh_btn) + btn_row.addWidget(self.remove_btn) + btn_row.addWidget(self.clear_btn) + btn_row.addStretch() + close_btn = QPushButton("Fermer") + close_btn.clicked.connect(self.accept) + btn_row.addWidget(close_btn) + layout.addLayout(btn_row) + + self.refresh_btn.clicked.connect(self.refresh) + self.remove_btn.clicked.connect(self.remove_selected) + self.clear_btn.clicked.connect(self.clear_all) + + self.refresh() + + def refresh(self) -> None: + self.tree.clear() + entries = self.store.list_entries() + for entry in entries: + item = QTreeWidgetItem([entry.hostname, entry.key_type, entry.fingerprint]) + item.setData(0, Qt.UserRole, entry.hostname) + self.tree.addTopLevelItem(item) + self.count_label.setText(f"{len(entries)} hote(s) enregistre(s).") + self.remove_btn.setEnabled(bool(entries)) + self.clear_btn.setEnabled(bool(entries)) + + def remove_selected(self) -> None: + items = self.tree.selectedItems() + if not items: + return + hostnames = {item.data(0, Qt.UserRole) for item in items} + if QMessageBox.question( + self, "Supprimer", + f"Retirer {len(hostnames)} hote(s) du fichier known_hosts ?\n" + "Une confirmation d'empreinte sera redemandee a la prochaine connexion.", + ) != QMessageBox.Yes: + return + for hostname in hostnames: + self.store.remove_host(hostname) + self.refresh() + + def clear_all(self) -> None: + if QMessageBox.question( + self, "Tout effacer", + "Vider entierement le fichier known_hosts ?\n" + "Toutes les connexions redemanderont une confirmation d'empreinte.", + ) != QMessageBox.Yes: + return + self.store.clear_all() + self.refresh() diff --git a/app/dialogs/session_dialog.py b/app/dialogs/session_dialog.py new file mode 100644 index 0000000..8b2465c --- /dev/null +++ b/app/dialogs/session_dialog.py @@ -0,0 +1,151 @@ +"""Boite de dialogue de creation/edition d'une session SSH.""" +from __future__ import annotations + +from PySide6.QtWidgets import ( + QDialog, QFormLayout, QLineEdit, QSpinBox, QComboBox, QDialogButtonBox, + QHBoxLayout, QPushButton, QTextEdit, QFileDialog, QWidget, QVBoxLayout, + QCheckBox, QMessageBox, +) + +from app.session_manager import Session, AUTH_AUTO, AUTH_KEY, AUTH_PASSWORD, AUTH_AGENT +from app.key_manager import KeyManager + + +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 = ""): + super().__init__(parent) + self.key_manager = key_manager + 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) + + layout = QVBoxLayout(self) + form = QFormLayout() + + self.name_edit = QLineEdit(self.session.name) + self.group_edit = QLineEdit(self.session.group) + self.group_edit.setPlaceholderText("ex: Production/Web") + self.host_edit = QLineEdit(self.session.host) + self.port_spin = QSpinBox() + self.port_spin.setRange(1, 65535) + self.port_spin.setValue(self.session.port or 22) + self.user_edit = QLineEdit(self.session.username) + + self.auth_combo = QComboBox() + self.auth_combo.addItem("Automatique (comme la commande ssh)", AUTH_AUTO) + self.auth_combo.addItem("Cle SSH choisie", AUTH_KEY) + self.auth_combo.addItem("Mot de passe (saisi a la connexion)", AUTH_PASSWORD) + self.auth_combo.addItem("Agent SSH", AUTH_AGENT) + self.auth_combo.setToolTip( + "Automatique : utilise l'agent SSH et vos cles ~/.ssh comme le fait la commande ssh,\n" + "et demande un mot de passe seulement si aucune cle ne convient." + ) + idx = self.auth_combo.findData(self.session.auth_type) + self.auth_combo.setCurrentIndex(idx if idx >= 0 else 0) + + self.key_combo = QComboBox() + self._reload_keys() + self.browse_key_btn = QPushButton("Parcourir...") + + key_row = QWidget() + key_layout = QHBoxLayout(key_row) + key_layout.setContentsMargins(0, 0, 0, 0) + key_layout.addWidget(self.key_combo, 1) + key_layout.addWidget(self.browse_key_btn) + + self.notes_edit = QTextEdit(self.session.notes) + self.notes_edit.setFixedHeight(60) + + form.addRow("Nom :", self.name_edit) + form.addRow("Dossier :", self.group_edit) + form.addRow("Hote :", self.host_edit) + form.addRow("Port :", self.port_spin) + form.addRow("Utilisateur :", self.user_edit) + form.addRow("Authentification :", self.auth_combo) + form.addRow("Cle privee :", key_row) + form.addRow("Notes :", self.notes_edit) + + layout.addLayout(form) + + self.save_check = QCheckBox("Enregistrer dans mes sessions (optionnel)") + if self.quick_mode: + # decoche par defaut : les serveurs de test changent d'IP (DHCP), les + # enregistrer systematiquement encombre la liste pour rien + self.save_check.setChecked(False) + layout.addWidget(self.save_check) + + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + if self.quick_mode: + buttons.button(QDialogButtonBox.Ok).setText(connect_label) + buttons.accepted.connect(self._on_accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + self.auth_combo.currentIndexChanged.connect(self._update_key_row) + self.browse_key_btn.clicked.connect(self._browse_key) + self._update_key_row() + + if self.session.key_path: + existing = self.key_combo.findData(self.session.key_path) + if existing >= 0: + self.key_combo.setCurrentIndex(existing) + else: + self.key_combo.addItem(self.session.key_path, self.session.key_path) + self.key_combo.setCurrentIndex(self.key_combo.count() - 1) + + def _reload_keys(self) -> None: + self.key_combo.clear() + for entry in self.key_manager.keys: + self.key_combo.addItem(f"{entry.name} ({entry.key_type})", entry.private_path) + + def _update_key_row(self) -> None: + is_key_auth = self.auth_combo.currentData() == AUTH_KEY + self.key_combo.setEnabled(is_key_auth) + self.browse_key_btn.setEnabled(is_key_auth) + + def _browse_key(self) -> None: + path, _ = QFileDialog.getOpenFileName(self, "Choisir une cle privee") + if path: + self.key_combo.addItem(path, path) + self.key_combo.setCurrentIndex(self.key_combo.count() - 1) + + 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 + + def _on_accept(self) -> None: + if not self.host_edit.text().strip(): + QMessageBox.warning(self, "Hote manquant", "Indiquez l'adresse du serveur (nom d'hote ou IP).") + self.host_edit.setFocus() + return + if not self.user_edit.text().strip(): + QMessageBox.warning(self, "Utilisateur manquant", "Indiquez le nom d'utilisateur.") + self.user_edit.setFocus() + return + if self.auth_combo.currentData() == AUTH_KEY and not self.key_combo.currentData(): + QMessageBox.warning( + self, "Cle manquante", + "Aucune cle privee n'est selectionnee.\n\n" + "Choisissez une cle (bouton ยซ Parcourir... ยป), ou passez l'authentification " + "sur ยซ Automatique ยป โ€” qui utilisera l'agent SSH et vos cles ~/.ssh comme le " + "fait la commande ssh.", + ) + self.auth_combo.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() + self.session.port = self.port_spin.value() + self.session.username = self.user_edit.text().strip() + 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.accept() + + def result_session(self) -> Session: + return self.session diff --git a/app/key_manager.py b/app/key_manager.py new file mode 100644 index 0000000..f829b51 --- /dev/null +++ b/app/key_manager.py @@ -0,0 +1,201 @@ +"""Gestion des cles SSH : generation, import, export, registre local.""" +from __future__ import annotations + +import json +import shutil +import uuid +from dataclasses import dataclass, field, asdict +from pathlib import Path + +import paramiko +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa, ed25519 + +from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +KEYS_FILE = DATA_DIR / "keys.json" +DEFAULT_KEY_DIR = DATA_DIR / "keys" + +KEY_TYPE_ED25519 = "ed25519" +KEY_TYPE_RSA = "rsa" + + +@dataclass +class KeyEntry: + id: str = field(default_factory=lambda: str(uuid.uuid4())) + name: str = "" + key_type: str = KEY_TYPE_ED25519 + private_path: str = "" + public_path: str = "" + fingerprint: str = "" + public_key_text: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + @staticmethod + def from_dict(data: dict) -> "KeyEntry": + e = KeyEntry() + for k in KeyEntry.__dataclass_fields__: + if k in data: + setattr(e, k, data[k]) + return e + + +class KeyManager: + def __init__(self, registry_path: Path = KEYS_FILE): + self.registry_path = registry_path + self.keys: list[KeyEntry] = [] + self.load() + + # ---------- registre ---------- + def load(self) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + DEFAULT_KEY_DIR.mkdir(parents=True, exist_ok=True) + if not self.registry_path.exists(): + self.keys = [] + self.save() + return + try: + raw = json.loads(self.registry_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + raw = {"keys": []} + self.keys = [KeyEntry.from_dict(k) for k in raw.get("keys", [])] + + def save(self) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + payload = {"keys": [k.to_dict() for k in self.keys]} + self.registry_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + + def remove(self, key_id: str, delete_files: bool = False) -> None: + entry = next((k for k in self.keys if k.id == key_id), None) + if entry and delete_files: + for p in (entry.private_path, entry.public_path): + try: + if p and Path(p).exists(): + Path(p).unlink() + except OSError: + pass + self.keys = [k for k in self.keys if k.id != key_id] + self.save() + + # ---------- generation ---------- + def generate_key( + self, + name: str, + key_type: str = KEY_TYPE_ED25519, + rsa_bits: int = 4096, + passphrase: str | None = None, + save_dir: Path | None = None, + comment: str = "", + ) -> KeyEntry: + save_dir = save_dir or DEFAULT_KEY_DIR + save_dir.mkdir(parents=True, exist_ok=True) + safe_name = "".join(c for c in name if c.isalnum() or c in ("-", "_")) or "id_key" + filename = f"id_{key_type}_{safe_name}" + priv_path = save_dir / filename + pub_path = save_dir / f"{filename}.pub" + + if key_type == KEY_TYPE_ED25519: + private_key = ed25519.Ed25519PrivateKey.generate() + else: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=rsa_bits) + + encryption = ( + serialization.BestAvailableEncryption(passphrase.encode("utf-8")) + if passphrase + else serialization.NoEncryption() + ) + priv_bytes = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.OpenSSH, + encryption_algorithm=encryption, + ) + pub_bytes = private_key.public_key().public_bytes( + encoding=serialization.Encoding.OpenSSH, + format=serialization.PublicFormat.OpenSSH, + ) + comment_suffix = f" {comment}".rstrip() if comment else "" + pub_text = pub_bytes.decode("utf-8") + comment_suffix + "\n" + + priv_path.write_bytes(priv_bytes) + pub_path.write_text(pub_text, encoding="utf-8") + try: + priv_path.chmod(0o600) + except OSError: + pass + + fingerprint = self._fingerprint(pub_bytes) + entry = KeyEntry( + name=name, + key_type=key_type, + private_path=str(priv_path), + public_path=str(pub_path), + fingerprint=fingerprint, + public_key_text=pub_text.strip(), + ) + self.keys.append(entry) + self.save() + return entry + + # ---------- import ---------- + def import_key(self, name: str, private_key_path: str, public_key_path: str | None = None) -> KeyEntry: + priv_path = Path(private_key_path) + key_type, pub_text = self._inspect_private_key(priv_path) + + pub_path_str = "" + if public_key_path and Path(public_key_path).exists(): + pub_path_str = public_key_path + pub_text = Path(public_key_path).read_text(encoding="utf-8").strip() + else: + guess = priv_path.with_suffix(priv_path.suffix + ".pub") + if guess.exists(): + pub_path_str = str(guess) + pub_text = guess.read_text(encoding="utf-8").strip() + + fingerprint = pub_text.split()[1] if pub_text and len(pub_text.split()) > 1 else "" + entry = KeyEntry( + name=name, + key_type=key_type, + private_path=str(priv_path), + public_path=pub_path_str, + fingerprint=fingerprint[:16], + public_key_text=pub_text, + ) + self.keys.append(entry) + self.save() + return entry + + def _inspect_private_key(self, priv_path: Path) -> tuple[str, str]: + """Tente de determiner le type de cle sans mot de passe (best effort).""" + for loader, ktype in ( + (paramiko.Ed25519Key, KEY_TYPE_ED25519), + (paramiko.RSAKey, KEY_TYPE_RSA), + ): + try: + pkey = loader.from_private_key_file(str(priv_path)) + pub_text = f"{pkey.get_name()} {pkey.get_base64()}" + return ktype, pub_text + except paramiko.PasswordRequiredException: + return ktype, "" + except (paramiko.SSHException, OSError): + continue + return "unknown", "" + + # ---------- export ---------- + def export_public_key(self, entry: KeyEntry, destination: str) -> None: + Path(destination).write_text(entry.public_key_text + "\n", encoding="utf-8") + + def export_private_key(self, entry: KeyEntry, destination: str) -> None: + shutil.copyfile(entry.private_path, destination) + + @staticmethod + def _fingerprint(pub_bytes: bytes) -> str: + import base64 + import hashlib + + parts = pub_bytes.split() + if len(parts) < 2: + return "" + raw = base64.b64decode(parts[1]) + digest = hashlib.sha256(raw).digest() + return "SHA256:" + base64.b64encode(digest).decode("utf-8").rstrip("=") diff --git a/app/known_hosts_manager.py b/app/known_hosts_manager.py new file mode 100644 index 0000000..d265fad --- /dev/null +++ b/app/known_hosts_manager.py @@ -0,0 +1,72 @@ +"""Gestion du fichier known_hosts local (cles d'hotes approuvees via TOFU). + +Avec des serveurs en IP dynamique (DHCP), une IP reattribuee a une autre machine +fait apparaitre une "cle d'hote modifiee" lors de la connexion : ce module permet +de retirer facilement l'ancienne entree, en masse ou au cas par cas, sans avoir +a editer le fichier a la main. +""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import paramiko + +from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +KNOWN_HOSTS_FILE = DATA_DIR / "known_hosts" + + +@dataclass +class KnownHostEntry: + hostname: str + key_type: str + fingerprint: str + + +class KnownHostsStore: + def __init__(self, path: Path = KNOWN_HOSTS_FILE): + self.path = path + + def _load(self) -> paramiko.HostKeys: + hk = paramiko.HostKeys() + if self.path.exists(): + try: + hk.load(str(self.path)) + except Exception: + pass + return hk + + def list_entries(self) -> list[KnownHostEntry]: + hk = self._load() + entries: list[KnownHostEntry] = [] + for hostname in hk.keys(): + for key_type, key in hk[hostname].items(): + entries.append(KnownHostEntry( + hostname=hostname, + key_type=key_type, + fingerprint=key.get_fingerprint().hex(":"), + )) + entries.sort(key=lambda e: e.hostname) + return entries + + def remove_host(self, hostname: str) -> None: + hk = self._load() + if hostname in hk: + del hk[hostname] + self._save(hk) + + def replace_host_key(self, hostname: str, key: paramiko.PKey) -> None: + """Retire l'ancienne cle d'un hote et enregistre la nouvelle (cas d'une IP DHCP reattribuee).""" + hk = self._load() + if hostname in hk: + del hk[hostname] + hk.add(hostname, key.get_name(), key) + self._save(hk) + + def clear_all(self) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + self.path.write_text("", encoding="utf-8") + + def _save(self, hk: paramiko.HostKeys) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + hk.save(str(self.path)) diff --git a/app/local_terminal.py b/app/local_terminal.py new file mode 100644 index 0000000..791d521 --- /dev/null +++ b/app/local_terminal.py @@ -0,0 +1,151 @@ +"""Terminal local (PowerShell / Invite de commandes) via pseudo-console Windows (pywinpty). + +Independant de toute session enregistree : permet d'ouvrir un terminal Windows +fonctionnel en un clic, comme le fait MobaXterm avec ses "terminaux locaux". +""" +from __future__ import annotations + +import queue +import threading + +import winpty +from PySide6.QtCore import QThread, Signal + +SHELL_POWERSHELL = "powershell" +SHELL_CMD = "cmd" + +SHELL_LABELS = { + SHELL_POWERSHELL: "PowerShell", + SHELL_CMD: "Invite de commandes (cmd)", +} +SHELL_COMMANDS = { + SHELL_POWERSHELL: "powershell.exe -NoLogo", + SHELL_CMD: "cmd.exe", +} + + +class LocalTerminalConnection(QThread): + """Meme interface de signaux que SSHConnection, pour reutiliser TerminalWidget tel quel.""" + + connected = Signal() + data_received = Signal(bytes) + disconnected = Signal(str) + error = Signal(str) + + def __init__(self, shell: str = SHELL_POWERSHELL, cwd: str | None = None, + cols: int = 80, rows: int = 24, parent=None): + super().__init__(parent) + self.shell = shell + self.cwd = cwd + self.cols = cols + self.rows = rows + self.pty: winpty.PtyProcess | None = None + self._stop = threading.Event() + + # File d'envoi + thread dedie : send() est appele directement depuis le thread GUI + # (une frappe clavier). pty.write() peut bloquer si la console Windows n'est pas + # assez rapide a consommer son pipe d'entree ; sans decouplage, ce blocage gele + # toute l'interface le temps de l'ecriture (pty.read() n'a pas de timeout pour + # entrelacer l'ecriture dans la meme boucle, contrairement au canal SSH). + self._out_queue: queue.Queue[bytes] = queue.Queue() + self._writer_thread: threading.Thread | None = None + + def run(self) -> None: + try: + command = SHELL_COMMANDS.get(self.shell, SHELL_COMMANDS[SHELL_POWERSHELL]) + self.pty = self._spawn(command) + + # La taille reelle du widget est connue apres la mise en page, souvent *apres* + # la creation de la connexion : on reapplique ici la derniere taille demandee, + # sinon la console reste en 80x24 alors que l'affichage est bien plus grand + # (desynchronisation = retours a la ligne et redraws incoherents). + try: + self.pty.setwinsize(self.rows, self.cols) + except Exception: + pass + + self._writer_thread = threading.Thread(target=self._writer_loop, daemon=True) + self._writer_thread.start() + self.connected.emit() + + while not self._stop.is_set(): + try: + data = self.pty.read(4096) + except EOFError: + break + if not data: + break + self.data_received.emit(data.encode("utf-8", errors="replace")) + + self.disconnected.emit("Terminal ferme") + except Exception as exc: # noqa: BLE001 + self.error.emit(str(exc)) + finally: + self._cleanup() + + def _spawn(self, command: str) -> winpty.PtyProcess: + """Ouvre la pseudo-console, en privilegiant ConPTY. + + ConPTY est l'API pseudo-console native de Windows (10 1809+) : c'est un vrai + passe-plat. L'ancien backend WinPTY, lui, *scrute* le tampon d'ecran de la console + et resynthetise l'affichage : mesure faite, il perd des caracteres a la frappe + (16 echos sur 23) et affiche ~32 ms de latence contre ~2,5 ms pour ConPTY, avec des + artefacts de curseur qui saute et de saisie reecrite. On ne retombe dessus que si + ConPTY est indisponible (Windows trop ancien). + """ + try: + return winpty.PtyProcess.spawn( + command, + dimensions=(self.rows, self.cols), + cwd=self.cwd, + backend=winpty.Backend.ConPTY, + ) + except Exception: + return winpty.PtyProcess.spawn( + command, + dimensions=(self.rows, self.cols), + cwd=self.cwd, + backend=winpty.Backend.WinPTY, + ) + + def send(self, data: bytes) -> None: + """Appele depuis le thread GUI (frappe clavier) : ne doit jamais bloquer.""" + self._out_queue.put(data) + + def _writer_loop(self) -> None: + """Tourne dans son propre thread : effectue les ecritures potentiellement bloquantes. + + Une erreur d'ecriture ponctuelle ne doit pas tuer le thread : sinon, toute frappe + suivante serait silencieusement perdue pour le reste de la session (le process + local restant lui-meme actif). On n'arrete la boucle que si le process est mort.""" + while not self._stop.is_set(): + try: + data = self._out_queue.get(timeout=0.2) + except queue.Empty: + continue + if not (self.pty and self.pty.isalive()): + continue + try: + self.pty.write(data.decode("utf-8", errors="ignore")) + except Exception: + if not (self.pty and self.pty.isalive()): + break + + def resize(self, cols: int, rows: int) -> None: + self.cols, self.rows = cols, rows + if self.pty and self.pty.isalive(): + try: + self.pty.setwinsize(rows, cols) + except Exception: + pass + + def close_connection(self) -> None: + self._stop.set() + self._cleanup() + + def _cleanup(self) -> None: + try: + if self.pty and self.pty.isalive(): + self.pty.terminate(force=True) + except Exception: + pass diff --git a/app/main_window.py b/app/main_window.py new file mode 100644 index 0000000..29bdde4 --- /dev/null +++ b/app/main_window.py @@ -0,0 +1,720 @@ +"""Fenetre principale : arborescence des sessions, onglets terminal/SFTP.""" +from __future__ import annotations + +import copy +import re +import uuid + +from PySide6.QtCore import Qt, Signal +from PySide6.QtGui import QAction, QActionGroup, QKeySequence +from PySide6.QtWidgets import ( + QMainWindow, QDockWidget, QTreeWidget, QTreeWidgetItem, QTabWidget, QWidget, + QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMenu, QMessageBox, QSplitter, + QStatusBar, QFrame, QApplication, +) + +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 +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.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.\-]*$") + + +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) + 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) + + self.connection = None # assigne par les sous-classes + + def _wire_connection(self) -> None: + self.connection.connected.connect(self._on_connected) + self.connection.data_received.connect(self.terminal.feed) + 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 _on_connected(self) -> None: + self.terminal.setFocus() + + 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", message) + + def close_connection(self) -> None: + 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.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 "", + # comme MobaXterm : le navigateur de fichiers est la des l'ouverture, + # pas cache derriere une action supplementaire + open_sftp_on_connect=True, + ) + self._wire_connection() + self.connection.sftp_ready.connect(self._on_sftp_ready) + self.connection.password_prompt.connect(self._on_password_prompt) + 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.sftp_toggle.toggled.connect(self._toggle_sftp) + + self.connection.start() + + def _on_connected(self) -> None: + self.status_label.setText(f"โ— Connecte : {self.session.username}@{self.session.host}:{self.session.port}") + # 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 _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.connection = LocalTerminalConnection(shell=shell) + self._wire_connection() + self.connection.start() + + 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(SHELL_POWERSHELL) + + # ---------- 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() + + new_ps_action = QAction("Nouveau terminal local (PowerShell)", self) + new_ps_action.setShortcut(QKeySequence("Ctrl+Shift+T")) + new_ps_action.triggered.connect(lambda: self.open_local_terminal(SHELL_POWERSHELL)) + file_menu.addAction(new_ps_action) + + new_cmd_action = QAction("Nouveau terminal local (Invite de commandes)", self) + new_cmd_action.triggered.connect(lambda: self.open_local_terminal(SHELL_CMD)) + file_menu.addAction(new_cmd_action) + + 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) + term_menu.addAction("PowerShell", lambda: self.open_local_terminal(SHELL_POWERSHELL)) + term_menu.addAction("Invite de commandes (cmd)", lambda: self.open_local_terminal(SHELL_CMD)) + term_btn.setMenu(term_menu) + toolbar.addWidget(term_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) + 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 = SHELL_POWERSHELL) -> None: + tab = LocalTerminalTabWidget(shell) + tab.sftp_requested.connect(self.quick_sftp) + label = "PowerShell" if shell == SHELL_POWERSHELL else "cmd" + index = self.tabs.addTab(tab, f"๐Ÿ’ป {label}") + self.tabs.setCurrentIndex(index) + + 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/paths.py b/app/paths.py new file mode 100644 index 0000000..fadce98 --- /dev/null +++ b/app/paths.py @@ -0,0 +1,49 @@ +"""Emplacements des donnees et des ressources, valables en mode script comme en .exe. + +Point d'attention : une fois l'application figee par PyInstaller, `__file__` pointe vers +un dossier temporaire (`sys._MEIPASS`) que Windows supprime a la fermeture. Y ecrire les +sessions, les cles ou known_hosts reviendrait a les perdre a chaque sortie. Les donnees +utilisateur vont donc a cote de l'executable (mode portable), avec repli sur %APPDATA% +si cet emplacement n'est pas accessible en ecriture (installation dans Program Files, +lancement depuis un support en lecture seule...). +""" +from __future__ import annotations + +import os +import sys +from pathlib import Path + +FROZEN = getattr(sys, "frozen", False) + + +def _app_dir() -> Path: + """Dossier de l'executable (mode figรฉ) ou racine du projet (mode script).""" + if FROZEN: + return Path(sys.executable).resolve().parent + return Path(__file__).resolve().parent.parent + + +def _resolve_resources() -> Path: + if FROZEN: + bundled = getattr(sys, "_MEIPASS", None) + if bundled: # onefile : ressources extraites dans un dossier temporaire + return Path(bundled) / "resources" + return _app_dir() / "resources" + + +def _resolve_data() -> Path: + candidate = _app_dir() / "data" + try: + candidate.mkdir(parents=True, exist_ok=True) + probe = candidate / ".write_test" + probe.write_text("", encoding="utf-8") + probe.unlink() + return candidate + except OSError: + fallback = Path(os.environ.get("APPDATA") or Path.home()) / "PySSHClient" + fallback.mkdir(parents=True, exist_ok=True) + return fallback + + +RESOURCES_DIR = _resolve_resources() +DATA_DIR = _resolve_data() diff --git a/app/session_manager.py b/app/session_manager.py new file mode 100644 index 0000000..e1f0fbc --- /dev/null +++ b/app/session_manager.py @@ -0,0 +1,83 @@ +"""Stockage des sessions SSH (serveurs enregistres) dans un fichier JSON local.""" +from __future__ import annotations + +import json +import uuid +from dataclasses import dataclass, field, asdict +from pathlib import Path + +from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +SESSIONS_FILE = DATA_DIR / "sessions.json" + +AUTH_AUTO = "auto" # comme la commande ssh : agent + cles ~/.ssh, mot de passe en secours +AUTH_KEY = "key" # cle privee explicitement choisie +AUTH_PASSWORD = "password" +AUTH_AGENT = "agent" + + +@dataclass +class Session: + id: str = field(default_factory=lambda: str(uuid.uuid4())) + name: str = "" + group: str = "" # ex: "Prod/Web" -- sert a construire l'arborescence de dossiers + host: str = "" + port: int = 22 + username: str = "" + auth_type: str = AUTH_AUTO # auto | key | password | agent + key_path: str = "" + notes: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + @staticmethod + def from_dict(data: dict) -> "Session": + known = {f: data.get(f) for f in Session.__dataclass_fields__ if f in data} + s = Session() + for k, v in known.items(): + setattr(s, k, v) + return s + + +class SessionStore: + """Charge/sauvegarde la liste des sessions depuis data/sessions.json.""" + + def __init__(self, path: Path = SESSIONS_FILE): + self.path = path + self.sessions: list[Session] = [] + self.load() + + def load(self) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + if not self.path.exists(): + self.sessions = [] + self.save() + return + try: + raw = json.loads(self.path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + raw = {"sessions": []} + self.sessions = [Session.from_dict(s) for s in raw.get("sessions", [])] + + def save(self) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + payload = {"sessions": [s.to_dict() for s in self.sessions]} + self.path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + + def add(self, session: Session) -> None: + self.sessions.append(session) + self.save() + + def update(self, session: Session) -> None: + for i, s in enumerate(self.sessions): + if s.id == session.id: + self.sessions[i] = session + break + self.save() + + def delete(self, session_id: str) -> None: + self.sessions = [s for s in self.sessions if s.id != session_id] + self.save() + + def groups(self) -> list[str]: + return sorted({s.group for s in self.sessions if s.group}) diff --git a/app/sftp_panel.py b/app/sftp_panel.py new file mode 100644 index 0000000..e8c0309 --- /dev/null +++ b/app/sftp_panel.py @@ -0,0 +1,673 @@ +"""Client SFTP integre facon WinSCP : double panneau, colonnes triables, glisser-deposer, +renommage, permissions, et file de transferts avec progression globale.""" +from __future__ import annotations + +import json +import os +import posixpath +import stat +from datetime import datetime +from pathlib import Path + +import paramiko +from PySide6.QtCore import Qt, QThread, Signal, QMimeData, QPoint +from PySide6.QtGui import QDrag, QKeySequence, QShortcut +from PySide6.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLineEdit, QToolButton, QTreeWidget, + QTreeWidgetItem, QSplitter, QLabel, QMessageBox, QInputDialog, QProgressDialog, + QAbstractItemView, QMenu, QHeaderView, +) + +MIME_TYPE = "application/x-pyssh-files" + +NAME_ROLE = Qt.UserRole +ISDIR_ROLE = Qt.UserRole + 1 +SIZE_ROLE = Qt.UserRole + 2 +MTIME_ROLE = Qt.UserRole + 3 +PARENT_ROLE = Qt.UserRole + 4 # marque l'entree ".." + + +def human_size(n: int) -> str: + value = float(n) + for unit in ("o", "Ko", "Mo", "Go", "To"): + if value < 1024: + return f"{value:.0f} {unit}" if unit == "o" else f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} Po" + + +def format_time(ts: float) -> str: + try: + return datetime.fromtimestamp(ts).strftime("%d/%m/%Y %H:%M") + except (OSError, ValueError, OverflowError): + return "" + + +# ---------------------------------------------------------------------------- +# Transferts +# ---------------------------------------------------------------------------- +class TransferJob: + def __init__(self, direction: str, local_path: str, remote_path: str, is_dir: bool): + self.direction = direction # "upload" | "download" + self.local_path = local_path + self.remote_path = remote_path + self.is_dir = is_dir + + +class TransferWorker(QThread): + """Execute une file de transferts sequentiellement, dans son propre thread.""" + + file_started = Signal(str, int, int) # nom, index, total + file_progress = Signal(int, int) # octets transferes, taille totale + finished_all = Signal(int, list) # nb reussis, liste d'erreurs + + def __init__(self, sftp: paramiko.SFTPClient, jobs: list[TransferJob]): + super().__init__() + self.sftp = sftp + self.jobs = jobs + self._cancel = False + self._done = 0 + self._errors: list[str] = [] + + def cancel(self) -> None: + self._cancel = True + + def run(self) -> None: + total = len(self.jobs) + for index, job in enumerate(self.jobs, start=1): + if self._cancel: + break + label = os.path.basename(job.local_path) or job.local_path + self.file_started.emit(label, index, total) + try: + if job.direction == "upload": + if job.is_dir: + self._upload_dir(job.local_path, job.remote_path) + else: + self.sftp.put(job.local_path, job.remote_path, callback=self._cb) + else: + if job.is_dir: + self._download_dir(job.remote_path, job.local_path) + else: + self.sftp.get(job.remote_path, job.local_path, callback=self._cb) + self._done += 1 + except Exception as exc: # noqa: BLE001 + self._errors.append(f"{label} : {exc}") + self.finished_all.emit(self._done, self._errors) + + def _cb(self, transferred: int, total: int) -> None: + self.file_progress.emit(transferred, total) + + def _upload_dir(self, local_dir: str, remote_dir: str) -> None: + try: + self.sftp.mkdir(remote_dir) + except IOError: + pass + for entry in os.listdir(local_dir): + if self._cancel: + return + lp = os.path.join(local_dir, entry) + rp = posixpath.join(remote_dir, entry) + if os.path.isdir(lp): + self._upload_dir(lp, rp) + else: + self.sftp.put(lp, rp, callback=self._cb) + + def _download_dir(self, remote_dir: str, local_dir: str) -> None: + Path(local_dir).mkdir(parents=True, exist_ok=True) + for attr in self.sftp.listdir_attr(remote_dir): + if self._cancel: + return + rp = posixpath.join(remote_dir, attr.filename) + lp = os.path.join(local_dir, attr.filename) + if stat.S_ISDIR(attr.st_mode): + self._download_dir(rp, lp) + else: + self.sftp.get(rp, lp, callback=self._cb) + + +# ---------------------------------------------------------------------------- +# Panneaux de fichiers +# ---------------------------------------------------------------------------- +class FileItem(QTreeWidgetItem): + """Ligne de fichier avec le tri attendu d'un gestionnaire de fichiers : + ".." toujours en tete, puis les dossiers, puis les fichiers ; taille et date + comparees numeriquement (et non comme du texte).""" + + SORT_VALUE_ROLE = Qt.UserRole + 10 + + def __lt__(self, other: QTreeWidgetItem) -> bool: + tree = self.treeWidget() + column = tree.sortColumn() if tree else 0 + descending = bool( + tree and tree.header().sortIndicatorOrder() == Qt.DescendingOrder + ) + + def pinned(result: bool) -> bool: + # Qt inverse le resultat en tri descendant : on le compense pour que + # ".." et les dossiers gardent leur place quel que soit le sens du tri. + return not result if descending else result + + self_parent = bool(self.data(0, PARENT_ROLE)) + other_parent = bool(other.data(0, PARENT_ROLE)) + if self_parent != other_parent: + return pinned(self_parent) + + self_dir = bool(self.data(0, ISDIR_ROLE)) + other_dir = bool(other.data(0, ISDIR_ROLE)) + if self_dir != other_dir: + return pinned(self_dir) + + self_value = self.data(column, self.SORT_VALUE_ROLE) + other_value = other.data(column, self.SORT_VALUE_ROLE) + if self_value is not None and other_value is not None: + return self_value < other_value + return self.text(column).lower() < other.text(column).lower() + + +class FileTree(QTreeWidget): + """Arbre de fichiers avec glisser-deposer entre panneaux.""" + + files_dropped = Signal(dict) + + def __init__(self, side: str, parent=None): + super().__init__(parent) + self.side = side + self.setSelectionMode(QAbstractItemView.ExtendedSelection) + self.setRootIsDecorated(False) + self.setSortingEnabled(True) + self.setDragEnabled(True) + self.setAcceptDrops(True) + self.setDropIndicatorShown(True) + self.setDragDropMode(QAbstractItemView.DragDrop) + self.setHeaderLabels(["Nom", "Taille", "Modifie", "Droits"]) + # sans tri initial explicite, les lignes resteraient dans l'ordre d'insertion + self.sortByColumn(0, Qt.AscendingOrder) + self.header().setSectionResizeMode(0, QHeaderView.Stretch) + self.setColumnWidth(1, 90) + self.setColumnWidth(2, 130) + self.setColumnWidth(3, 100) + + def startDrag(self, actions) -> None: + entries = [] + for item in self.selectedItems(): + if item.data(0, PARENT_ROLE): + continue + entries.append({"name": item.data(0, NAME_ROLE), "is_dir": bool(item.data(0, ISDIR_ROLE))}) + if not entries: + return + payload = {"side": self.side, "entries": entries} + mime = QMimeData() + mime.setData(MIME_TYPE, json.dumps(payload).encode("utf-8")) + drag = QDrag(self) + drag.setMimeData(mime) + drag.exec(Qt.CopyAction) + + def dragEnterEvent(self, event) -> None: + if event.mimeData().hasFormat(MIME_TYPE): + event.acceptProposedAction() + + def dragMoveEvent(self, event) -> None: + if event.mimeData().hasFormat(MIME_TYPE): + event.acceptProposedAction() + + def dropEvent(self, event) -> None: + if not event.mimeData().hasFormat(MIME_TYPE): + return + payload = json.loads(bytes(event.mimeData().data(MIME_TYPE)).decode("utf-8")) + if payload.get("side") == self.side: + return # depose sur lui-meme : rien a faire + event.acceptProposedAction() + self.files_dropped.emit(payload) + + +class FilePanel(QWidget): + """Panneau generique (decline en local et distant).""" + + path_changed = Signal(str) + transfer_here = Signal(dict) + + def __init__(self, title: str, side: str, parent=None): + super().__init__(parent) + self.side = side + layout = QVBoxLayout(self) + layout.setContentsMargins(4, 4, 4, 4) + layout.setSpacing(4) + + header = QHBoxLayout() + self.title_label = QLabel(f"{title}") + header.addWidget(self.title_label) + header.addStretch() + layout.addLayout(header) + + nav = QHBoxLayout() + self.up_btn = QToolButton(); self.up_btn.setText("โฌ†"); self.up_btn.setToolTip("Dossier parent") + self.refresh_btn = QToolButton(); self.refresh_btn.setText("โŸณ"); self.refresh_btn.setToolTip("Actualiser (F5)") + self.mkdir_btn = QToolButton(); self.mkdir_btn.setText("๐Ÿ“+"); self.mkdir_btn.setToolTip("Nouveau dossier") + self.path_edit = QLineEdit() + nav.addWidget(self.up_btn) + nav.addWidget(self.refresh_btn) + nav.addWidget(self.mkdir_btn) + nav.addWidget(self.path_edit, 1) + layout.addLayout(nav) + + self.tree = FileTree(side) + layout.addWidget(self.tree) + + self.status = QLabel("") + self.status.setObjectName("statusLabel") + layout.addWidget(self.status) + + self.up_btn.clicked.connect(self.go_up) + self.refresh_btn.clicked.connect(self.refresh) + self.mkdir_btn.clicked.connect(self.make_dir) + self.path_edit.returnPressed.connect(lambda: self.navigate(self.path_edit.text())) + self.tree.itemDoubleClicked.connect(self._on_double_click) + self.tree.files_dropped.connect(self.transfer_here.emit) + self.tree.setContextMenuPolicy(Qt.CustomContextMenu) + self.tree.customContextMenuRequested.connect(self._context_menu) + + QShortcut(QKeySequence("F2"), self.tree, activated=self.rename_selected) + QShortcut(QKeySequence("F5"), self.tree, activated=self.refresh) + QShortcut(QKeySequence(Qt.Key_Delete), self.tree, activated=self.delete_selected) + + self.current_path = "" + + # ---------- helpers d'affichage ---------- + def _add_row(self, name: str, is_dir: bool, size: int, mtime: float, perms: str) -> None: + item = FileItem([ + ("๐Ÿ“ " if is_dir else "๐Ÿ“„ ") + name, + "" if is_dir else human_size(size), + format_time(mtime) if mtime else "", + perms, + ]) + item.setData(0, NAME_ROLE, name) + item.setData(0, ISDIR_ROLE, is_dir) + # tri numerique correct sur la taille et la date + item.setData(1, FileItem.SORT_VALUE_ROLE, size) + item.setData(2, FileItem.SORT_VALUE_ROLE, mtime) + self.tree.addTopLevelItem(item) + + def _add_parent_row(self) -> None: + item = FileItem(["๐Ÿ“ ..", "", "", ""]) + item.setData(0, PARENT_ROLE, True) + item.setData(0, ISDIR_ROLE, True) + self.tree.addTopLevelItem(item) + + def _on_double_click(self, item: QTreeWidgetItem) -> None: + if item.data(0, PARENT_ROLE): + self.go_up() + return + if item.data(0, ISDIR_ROLE): + self.navigate(self.join(self.current_path, item.data(0, NAME_ROLE))) + + def go_up(self) -> None: + parent = self.parent_of(self.current_path) + if parent and parent != self.current_path: + self.navigate(parent) + + def selected_entries(self) -> list[tuple[str, bool]]: + out = [] + for item in self.tree.selectedItems(): + if item.data(0, PARENT_ROLE): + continue + out.append((item.data(0, NAME_ROLE), bool(item.data(0, ISDIR_ROLE)))) + return out + + def _context_menu(self, pos: QPoint) -> None: + menu = QMenu(self) + has_selection = bool(self.selected_entries()) + act_transfer = menu.addAction("โฌ† Envoyer" if self.side == "local" else "โฌ‡ Recevoir") + act_transfer.setEnabled(has_selection) + menu.addSeparator() + act_rename = menu.addAction("Renommer (F2)") + act_rename.setEnabled(has_selection) + act_delete = menu.addAction("Supprimer (Suppr)") + act_delete.setEnabled(has_selection) + act_perms = None + if self.side == "remote": + menu.addSeparator() + act_perms = menu.addAction("Permissions...") + act_perms.setEnabled(has_selection) + menu.addSeparator() + act_mkdir = menu.addAction("Nouveau dossier") + act_refresh = menu.addAction("Actualiser (F5)") + + action = menu.exec(self.tree.viewport().mapToGlobal(pos)) + if action == act_transfer: + self.transfer_requested() + elif action == act_rename: + self.rename_selected() + elif action == act_delete: + self.delete_selected() + elif act_perms is not None and action == act_perms: + self.change_permissions() + elif action == act_mkdir: + self.make_dir() + elif action == act_refresh: + self.refresh() + + def transfer_requested(self) -> None: + entries = [{"name": n, "is_dir": d} for n, d in self.selected_entries()] + if entries: + self.transfer_here.emit({"side": self.side, "entries": entries, "_from_button": True}) + + # ---------- a surcharger ---------- + def navigate(self, path: str) -> None: ... + def refresh(self) -> None: ... + def join(self, a: str, b: str) -> str: ... + def parent_of(self, path: str) -> str: ... + def make_dir(self) -> None: ... + def rename_selected(self) -> None: ... + def delete_selected(self) -> None: ... + def change_permissions(self) -> None: ... + + +class LocalFilePanel(FilePanel): + def __init__(self, parent=None): + super().__init__("๐Ÿ’ป Local", "local", parent) + self.navigate(str(Path.home())) + + def join(self, a: str, b: str) -> str: + return str(Path(a) / b) + + def parent_of(self, path: str) -> str: + return str(Path(path).parent) + + def navigate(self, path: str) -> None: + p = Path(path) + if not p.exists() or not p.is_dir(): + QMessageBox.warning(self, "Chemin invalide", f"Dossier introuvable : {path}") + return + self.current_path = str(p) + self.path_edit.setText(self.current_path) + self.refresh() + self.path_changed.emit(self.current_path) + + def refresh(self) -> None: + self.tree.setSortingEnabled(False) + self.tree.clear() + self._add_parent_row() + count = 0 + try: + for entry in Path(self.current_path).iterdir(): + try: + st = entry.stat() + is_dir = entry.is_dir() + self._add_row(entry.name, is_dir, st.st_size, st.st_mtime, stat.filemode(st.st_mode)) + count += 1 + except OSError: + continue + except OSError as exc: + QMessageBox.warning(self, "Erreur", str(exc)) + self.tree.setSortingEnabled(True) + self.status.setText(f"{count} element(s)") + + def make_dir(self) -> None: + name, ok = QInputDialog.getText(self, "Nouveau dossier", "Nom du dossier :") + if not ok or not name: + return + try: + Path(self.current_path, name).mkdir() + self.refresh() + except OSError as exc: + QMessageBox.critical(self, "Erreur", str(exc)) + + def rename_selected(self) -> None: + entries = self.selected_entries() + if not entries: + return + old_name = entries[0][0] + new_name, ok = QInputDialog.getText(self, "Renommer", "Nouveau nom :", text=old_name) + if not ok or not new_name or new_name == old_name: + return + try: + Path(self.current_path, old_name).rename(Path(self.current_path, new_name)) + self.refresh() + except OSError as exc: + QMessageBox.critical(self, "Erreur", str(exc)) + + def delete_selected(self) -> None: + entries = self.selected_entries() + if not entries: + return + if QMessageBox.question( + self, "Supprimer", + f"Supprimer definitivement {len(entries)} element(s) en local ?", + ) != QMessageBox.Yes: + return + import shutil + for name, is_dir in entries: + target = Path(self.current_path, name) + try: + if is_dir: + shutil.rmtree(target) + else: + target.unlink() + except OSError as exc: + QMessageBox.critical(self, "Erreur", f"{name} : {exc}") + self.refresh() + + +class RemoteFilePanel(FilePanel): + def __init__(self, sftp: paramiko.SFTPClient, parent=None): + super().__init__("๐ŸŒ Distant", "remote", parent) + self.sftp = sftp + try: + home = self.sftp.normalize(".") + except Exception: + home = "/" + self.navigate(home) + + def join(self, a: str, b: str) -> str: + return posixpath.join(a, b) + + def parent_of(self, path: str) -> str: + return posixpath.dirname(path.rstrip("/")) or "/" + + def navigate(self, path: str) -> None: + try: + attrs = self.sftp.listdir_attr(path) + except IOError as exc: + QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}") + return + self.current_path = path + self.path_edit.setText(path) + self._populate(attrs) + self.path_changed.emit(path) + + def refresh(self) -> None: + if self.current_path: + self.navigate(self.current_path) + + def _populate(self, attrs) -> None: + self.tree.setSortingEnabled(False) + self.tree.clear() + self._add_parent_row() + for attr in attrs: + is_dir = stat.S_ISDIR(attr.st_mode) + self._add_row( + attr.filename, is_dir, attr.st_size or 0, attr.st_mtime or 0, + stat.filemode(attr.st_mode), + ) + self.tree.setSortingEnabled(True) + self.status.setText(f"{len(attrs)} element(s)") + + def make_dir(self) -> None: + name, ok = QInputDialog.getText(self, "Nouveau dossier", "Nom du dossier :") + if not ok or not name: + return + try: + self.sftp.mkdir(self.join(self.current_path, name)) + self.refresh() + except IOError as exc: + QMessageBox.critical(self, "Erreur", str(exc)) + + def rename_selected(self) -> None: + entries = self.selected_entries() + if not entries: + return + old_name = entries[0][0] + new_name, ok = QInputDialog.getText(self, "Renommer", "Nouveau nom :", text=old_name) + if not ok or not new_name or new_name == old_name: + return + try: + self.sftp.rename(self.join(self.current_path, old_name), self.join(self.current_path, new_name)) + self.refresh() + except IOError as exc: + QMessageBox.critical(self, "Erreur", str(exc)) + + def delete_selected(self) -> None: + entries = self.selected_entries() + if not entries: + return + if QMessageBox.question( + self, "Supprimer", + f"Supprimer definitivement {len(entries)} element(s) sur le serveur ?", + ) != QMessageBox.Yes: + return + for name, is_dir in entries: + path = self.join(self.current_path, name) + try: + if is_dir: + self._rmdir_recursive(path) + else: + self.sftp.remove(path) + except IOError as exc: + QMessageBox.critical(self, "Erreur", f"{name} : {exc}") + self.refresh() + + def _rmdir_recursive(self, path: str) -> None: + for attr in self.sftp.listdir_attr(path): + sub = posixpath.join(path, attr.filename) + if stat.S_ISDIR(attr.st_mode): + self._rmdir_recursive(sub) + else: + self.sftp.remove(sub) + self.sftp.rmdir(path) + + def change_permissions(self) -> None: + entries = self.selected_entries() + if not entries: + return + name = entries[0][0] + path = self.join(self.current_path, name) + try: + current = stat.S_IMODE(self.sftp.stat(path).st_mode) + except IOError as exc: + QMessageBox.critical(self, "Erreur", str(exc)) + return + text, ok = QInputDialog.getText( + self, "Permissions", f"Droits octaux pour {name} :", text=f"{current:03o}", + ) + if not ok or not text: + return + try: + mode = int(text.strip(), 8) + except ValueError: + QMessageBox.warning(self, "Valeur invalide", "Entrez une valeur octale, par exemple 644 ou 755.") + return + errors = [] + for entry_name, _ in entries: + try: + self.sftp.chmod(self.join(self.current_path, entry_name), mode) + except IOError as exc: + errors.append(f"{entry_name} : {exc}") + if errors: + QMessageBox.critical(self, "Erreur", "\n".join(errors)) + self.refresh() + + +# ---------------------------------------------------------------------------- +# Conteneur double panneau +# ---------------------------------------------------------------------------- +class SFTPPanel(QWidget): + def __init__(self, sftp: paramiko.SFTPClient, parent=None): + super().__init__(parent) + self.sftp = sftp + self._worker: TransferWorker | None = None + + layout = QVBoxLayout(self) + layout.setContentsMargins(4, 4, 4, 4) + + toolbar = QHBoxLayout() + self.upload_btn = QToolButton(); self.upload_btn.setText("โฌ† Envoyer") + self.download_btn = QToolButton(); self.download_btn.setText("โฌ‡ Recevoir") + hint = QLabel(" Glissez-deposez entre les panneaux, ou utilisez F2 (renommer), Suppr, F5.") + hint.setObjectName("statusLabel") + toolbar.addWidget(self.upload_btn) + toolbar.addWidget(self.download_btn) + toolbar.addWidget(hint) + toolbar.addStretch() + layout.addLayout(toolbar) + + splitter = QSplitter(Qt.Horizontal) + self.local = LocalFilePanel() + self.remote = RemoteFilePanel(sftp) + splitter.addWidget(self.local) + splitter.addWidget(self.remote) + splitter.setSizes([500, 500]) + layout.addWidget(splitter, 1) + + self.upload_btn.clicked.connect(self.local.transfer_requested) + self.download_btn.clicked.connect(self.remote.transfer_requested) + # un depot venant du panneau local signifie "envoyer", et inversement + self.local.transfer_here.connect(self._on_transfer_payload) + self.remote.transfer_here.connect(self._on_transfer_payload) + + def _on_transfer_payload(self, payload: dict) -> None: + source_side = payload.get("side") + entries = payload.get("entries", []) + if not entries: + return + jobs = [] + for entry in entries: + name, is_dir = entry["name"], entry["is_dir"] + if source_side == "local": + jobs.append(TransferJob( + "upload", self.local.join(self.local.current_path, name), + self.remote.join(self.remote.current_path, name), is_dir, + )) + else: + jobs.append(TransferJob( + "download", self.local.join(self.local.current_path, name), + self.remote.join(self.remote.current_path, name), is_dir, + )) + self._run_jobs(jobs) + + def _run_jobs(self, jobs: list[TransferJob]) -> None: + if not jobs: + return + if self._worker and self._worker.isRunning(): + QMessageBox.information(self, "Transfert en cours", "Attendez la fin du transfert en cours.") + return + + dialog = QProgressDialog("Preparation...", "Annuler", 0, 100, self) + dialog.setWindowTitle("Transfert SFTP") + dialog.setWindowModality(Qt.WindowModal) + dialog.setMinimumDuration(0) + dialog.setAutoClose(False) + dialog.setAutoReset(False) + + worker = TransferWorker(self.sftp, jobs) + worker.file_started.connect( + lambda name, i, total: dialog.setLabelText(f"({i}/{total}) {name}") + ) + worker.file_progress.connect( + lambda done, total: dialog.setValue(int(done * 100 / total) if total else 0) + ) + + def on_finished(done: int, errors: list) -> None: + dialog.close() + self.local.refresh() + self.remote.refresh() + if errors: + QMessageBox.warning( + self, "Transfert termine avec des erreurs", + f"{done} element(s) transfere(s).\n\nErreurs :\n" + "\n".join(errors[:10]), + ) + + worker.finished_all.connect(on_finished) + dialog.canceled.connect(worker.cancel) + self._worker = worker + worker.start() diff --git a/app/ssh_connection.py b/app/ssh_connection.py new file mode 100644 index 0000000..8ced4e5 --- /dev/null +++ b/app/ssh_connection.py @@ -0,0 +1,336 @@ +"""Connexion SSH en arriere-plan (thread Qt) : shell interactif + verification des cles d'hote.""" +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.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +KNOWN_HOSTS_FILE = DATA_DIR / "known_hosts" + + +class TrustPrompt(paramiko.MissingHostKeyPolicy): + """Politique qui demande confirmation (via signal Qt) avant de faire confiance a une nouvelle cle d'hote.""" + + def __init__(self, connection: "SSHConnection"): + self.connection = connection + + def missing_host_key(self, client, hostname, key): + accepted = self.connection._ask_host_key_trust(hostname, key) + if not accepted: + raise paramiko.SSHException(f"Cle d'hote refusee pour {hostname}") + client.get_host_keys().add(hostname, key.get_name(), key) + DATA_DIR.mkdir(parents=True, exist_ok=True) + client.save_host_keys(str(KNOWN_HOSTS_FILE)) + + +class SSHConnection(QThread): + """Gere une connexion SSH + un canal shell interactif dans un thread dedie.""" + + connected = Signal() + data_received = Signal(bytes) + disconnected = Signal(str) + error = Signal(str) + host_key_prompt = Signal(str, str, str) # hostname, key_type, fingerprint -> attend confirm_host_key() + host_key_changed_prompt = Signal(str, str, str, str) # hostname, key_type, old_fp, new_fp -> attend confirm_host_key_change() + passphrase_prompt = Signal(str) # key_path -> attend provide_passphrase() + password_prompt = Signal(str) # "user@host" -> attend provide_password() + sftp_ready = Signal(object) # SFTPClient ouvert dans le thread, pret a l'emploi + + def __init__( + self, + host: str, + port: int, + username: str, + auth_type: str, + key_path: str = "", + password: str = "", + passphrase: str = "", + cols: int = 80, + rows: int = 24, + open_shell: bool = True, + open_sftp_on_connect: bool = False, + parent=None, + ): + super().__init__(parent) + self.open_shell = open_shell + self.open_sftp_on_connect = open_sftp_on_connect + self.sftp: paramiko.SFTPClient | None = None + self.host = host + self.port = port + self.username = username + self.auth_type = auth_type + self.key_path = key_path + self.password = password + self.passphrase = passphrase + self.cols = cols + self.rows = rows + + self.client: paramiko.SSHClient | None = None + self.channel: paramiko.Channel | None = None + self._stop = threading.Event() + + self._trust_event = threading.Event() + self._trust_result = False + + self._passphrase_event = threading.Event() + self._passphrase_result: str | None = None + + self._host_key_change_event = threading.Event() + self._host_key_change_result = False + + self._password_event = threading.Event() + self._password_result: str | None = None + self.known_hosts = KnownHostsStore(KNOWN_HOSTS_FILE) + + # File d'envoi + thread dedie : send() est appele directement depuis le thread GUI + # (une frappe clavier). channel.send() peut bloquer (reseau congestionne, fenetre TCP + # pleine) ; sans decouplage, ce blocage gele toute l'interface le temps de l'ecriture. + self._out_queue: queue.Queue[bytes] = queue.Queue() + self._writer_thread: threading.Thread | None = None + + # ---------- confiance cle d'hote (appele depuis le thread SSH) ---------- + def _ask_host_key_trust(self, hostname: str, key: paramiko.PKey) -> bool: + self._trust_event.clear() + fingerprint = key.get_fingerprint().hex(":") + self.host_key_prompt.emit(hostname, key.get_name(), fingerprint) + self._trust_event.wait() # debloque via confirm_host_key() depuis le thread GUI + return self._trust_result + + def confirm_host_key(self, accepted: bool) -> None: + self._trust_result = accepted + self._trust_event.set() + + # ---------- cle d'hote modifiee, ex. IP DHCP reattribuee (appele depuis le thread SSH) ---------- + def _ask_host_key_change(self, hostname: str, got_key: paramiko.PKey, expected_key: paramiko.PKey) -> bool: + self._host_key_change_event.clear() + old_fp = expected_key.get_fingerprint().hex(":") + new_fp = got_key.get_fingerprint().hex(":") + self.host_key_changed_prompt.emit(hostname, got_key.get_name(), old_fp, new_fp) + self._host_key_change_event.wait() # debloque via confirm_host_key_change() depuis le thread GUI + return self._host_key_change_result + + def confirm_host_key_change(self, accepted: bool) -> None: + self._host_key_change_result = accepted + self._host_key_change_event.set() + + # ---------- passphrase de cle privee (appele depuis le thread SSH) ---------- + def _ask_passphrase(self, key_path: str) -> str | None: + self._passphrase_event.clear() + self._passphrase_result = None + self.passphrase_prompt.emit(key_path) + self._passphrase_event.wait() # debloque via provide_passphrase() depuis le thread GUI + return self._passphrase_result + + def provide_passphrase(self, text: str | None) -> None: + self._passphrase_result = text or None + self._passphrase_event.set() + + # ---------- mot de passe demande en secours (appele depuis le thread SSH) ---------- + def _ask_password(self) -> str | None: + self._password_event.clear() + self._password_result = None + self.password_prompt.emit(f"{self.username}@{self.host}") + self._password_event.wait() # debloque via provide_password() depuis le thread GUI + return self._password_result + + def provide_password(self, text: str | None) -> None: + self._password_result = text or None + self._password_event.set() + + # ---------- cycle de vie du thread ---------- + def run(self) -> None: + try: + self.client = paramiko.SSHClient() + DATA_DIR.mkdir(parents=True, exist_ok=True) + if KNOWN_HOSTS_FILE.exists(): + self.client.load_host_keys(str(KNOWN_HOSTS_FILE)) + self.client.set_missing_host_key_policy(TrustPrompt(self)) + + 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, + ) + 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. + if self.open_sftp_on_connect: + try: + self.sftp = self.client.open_sftp() + self.sftp_ready.emit(self.sftp) + except Exception: + # Sans shell, le SFTP est la raison d'etre de la connexion : c'est fatal. + # Avec un shell, le terminal doit rester utilisable meme si le serveur + # n'expose pas le sous-systeme SFTP. + if not self.open_shell: + raise + + if not self.open_shell: + # Connexion dediee au SFTP : on garde la session vivante sans shell. + self.connected.emit() + while not self._stop.is_set(): + self._stop.wait(0.2) + self.disconnected.emit("Connexion fermee") + return + + self.channel = self.client.invoke_shell(term="xterm-256color", width=self.cols, height=self.rows) + # La taille reelle du widget arrive souvent apres la creation de la connexion : + # on reapplique la derniere taille demandee, sinon le shell distant reste en + # 80x24 alors que l'affichage est plus grand (retours a la ligne incoherents). + try: + self.channel.resize_pty(width=self.cols, height=self.rows) + except Exception: + pass + self.channel.settimeout(0.5) + self._writer_thread = threading.Thread(target=self._writer_loop, daemon=True) + self._writer_thread.start() + self.connected.emit() + + while not self._stop.is_set(): + try: + data = self.channel.recv(4096) + if not data: + break + self.data_received.emit(data) + except TimeoutError: + continue + except Exception: + break + + self.disconnected.emit("Connexion fermee") + except paramiko.AuthenticationException: + self.error.emit("Authentification refusee (identifiants ou cle invalides).") + except Exception as exc: # noqa: BLE001 + self.error.emit(str(exc)) + finally: + self._cleanup() + + def _load_private_key(self, key_path: str, passphrase: str) -> paramiko.PKey: + last_exc = None + for loader in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey): + try: + return loader.from_private_key_file(key_path, password=passphrase or None) + except paramiko.PasswordRequiredException: + pwd = passphrase or self._ask_passphrase(key_path) + if not pwd: + raise paramiko.SSHException("Passphrase requise pour deverrouiller la cle") + try: + return loader.from_private_key_file(key_path, password=pwd) + except paramiko.SSHException as exc: + last_exc = exc + continue + except paramiko.SSHException as exc: + last_exc = exc + continue + raise last_exc or paramiko.SSHException("Format de cle non reconnu") + + def send(self, data: bytes) -> None: + """Appele depuis le thread GUI (frappe clavier) : ne doit jamais bloquer.""" + self._out_queue.put(data) + + def _writer_loop(self) -> None: + """Tourne dans son propre thread : effectue les ecritures reseau potentiellement bloquantes. + + Une erreur d'ecriture ponctuelle ne doit pas tuer le thread : sinon, toute frappe + suivante serait silencieusement perdue pour le reste de la session (le canal + restant lui-meme ouvert). On n'arrete la boucle que si le canal est ferme.""" + while not self._stop.is_set(): + try: + data = self._out_queue.get(timeout=0.2) + except queue.Empty: + continue + if not (self.channel and not self.channel.closed): + continue + try: + self.channel.send(data) + except OSError: + if not self.channel or self.channel.closed: + break + + def resize(self, cols: int, rows: int) -> None: + self.cols, self.rows = cols, rows + if self.channel and not self.channel.closed: + try: + self.channel.resize_pty(width=cols, height=rows) + except Exception: + pass + + def open_sftp(self) -> paramiko.SFTPClient | None: + if self.client: + return self.client.open_sftp() + return None + + def close_connection(self) -> None: + self._stop.set() + self._trust_event.set() # au cas ou on serait bloque sur une confirmation + self._passphrase_event.set() + self._host_key_change_event.set() + self._password_event.set() + self._cleanup() + + def _cleanup(self) -> None: + try: + if self.channel: + self.channel.close() + except Exception: + pass + try: + if self.client: + self.client.close() + except Exception: + pass diff --git a/app/terminal_widget.py b/app/terminal_widget.py new file mode 100644 index 0000000..0a8491f --- /dev/null +++ b/app/terminal_widget.py @@ -0,0 +1,266 @@ +"""Widget terminal : emulation ANSI/VT100 (pyte) rendue avec QPainter. + +Optimisation cle : on ne redessine que les lignes reellement modifiees depuis le +dernier rendu (via `screen.dirty`, fourni par pyte), au lieu de repeindre tout +l'ecran a chaque octet recu. Sans ca, chaque frappe clavier (et son echo distant) +declenche un repaint de la totalite de la grille, ce qui devient tres sensible +des que le terminal contient beaucoup de lignes ou que la fenetre est grande. +""" +from __future__ import annotations + +import pyte +from PySide6.QtCore import QRect, Qt, Signal +from PySide6.QtGui import QColor, QFont, QFontMetrics, QPainter, QKeyEvent, QWheelEvent +from PySide6.QtWidgets import QWidget + +from app import theme as app_theme + +# Sequences envoyees pour les touches speciales +KEY_SEQUENCES = { + Qt.Key_Up: b"\x1b[A", Qt.Key_Down: b"\x1b[B", Qt.Key_Right: b"\x1b[C", Qt.Key_Left: b"\x1b[D", + Qt.Key_Home: b"\x1b[H", Qt.Key_End: b"\x1b[F", Qt.Key_Insert: b"\x1b[2~", Qt.Key_Delete: b"\x1b[3~", + Qt.Key_PageUp: b"\x1b[5~", Qt.Key_PageDown: b"\x1b[6~", + Qt.Key_F1: b"\x1bOP", Qt.Key_F2: b"\x1bOQ", Qt.Key_F3: b"\x1bOR", Qt.Key_F4: b"\x1bOS", + Qt.Key_F5: b"\x1b[15~", Qt.Key_F6: b"\x1b[17~", Qt.Key_F7: b"\x1b[18~", Qt.Key_F8: b"\x1b[19~", + Qt.Key_F9: b"\x1b[20~", Qt.Key_F10: b"\x1b[21~", Qt.Key_F11: b"\x1b[23~", Qt.Key_F12: b"\x1b[24~", + Qt.Key_Backspace: b"\x7f", Qt.Key_Tab: b"\t", Qt.Key_Return: b"\r", Qt.Key_Enter: b"\r", + Qt.Key_Escape: b"\x1b", +} + + +class ReplyingScreen(pyte.HistoryScreen): + """Ecran pyte capable de repondre aux requetes du terminal. + + Certaines pseudo-consoles (ConPTY notamment) interrogent le terminal au demarrage + ("Device Attributes" \x1b[c, "Cursor Position Report" \x1b[6n) et **attendent la + reponse** avant de poursuivre. pyte sait construire ces reponses, mais les envoie + via write_process_input(), qui ne fait rien par defaut : sans ce branchement, la + console reste bloquee en attente. + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.reply_callback = None + + def write_process_input(self, data: str) -> None: + if self.reply_callback: + self.reply_callback(data.encode("utf-8", errors="ignore")) + + +class TerminalWidget(QWidget): + data_to_send = Signal(bytes) + resized = Signal(int, int) # cols, rows + + def __init__(self, parent=None): + super().__init__(parent) + self.setFocusPolicy(Qt.StrongFocus) + self.setAttribute(Qt.WA_OpaquePaintEvent, True) + + self.font_ = QFont("Cascadia Mono", 10) + if not self.font_.exactMatch(): + self.font_ = QFont("Consolas", 10) + self.font_.setStyleHint(QFont.Monospace) + self.setFont(self.font_) + metrics = QFontMetrics(self.font_) + self.char_w = metrics.horizontalAdvance("M") + self.char_h = metrics.height() + + self.cols = 80 + self.rows = 24 + self.screen = ReplyingScreen(self.cols, self.rows, history=5000) + # les reponses aux requetes du terminal repartent par le meme canal que le clavier + self.screen.reply_callback = self.data_to_send.emit + self.stream = pyte.ByteStream(self.screen) + self.scroll_offset = 0 + self._prev_cursor_y = 0 + + self._load_palette() + self.setMinimumSize(200, 100) + + def _load_palette(self) -> None: + """Construit les couleurs du terminal a partir du theme courant. + + Le cache evite de re-parser les memes couleurs hexa a chaque caractere + redessine (c'etait un point chaud du rendu).""" + palette = app_theme.terminal_palette() + self.default_fg = QColor(palette["fg"]) + self.default_bg = QColor(palette["bg"]) + self.cursor_color = QColor(*palette["cursor"]) + self._color_cache: dict[str, QColor] = { + name: QColor(value) for name, value in palette["ansi"].items() + } + self._bold_fg = QColor(palette["ansi"]["brightwhite"]) + + def refresh_theme(self) -> None: + """Recharge les couleurs apres un changement de theme, et redessine.""" + self._load_palette() + self.update() + + def buffer_text(self) -> str: + """Texte visible + historique proche, pour retrouver ce qui a ete tape + (utilise pour deduire l'hote d'une commande ssh saisie a la main). + + Les lignes remplies jusqu'a la derniere colonne sont recollees avec la + suivante : une commande longue (prompt Windows + ssh user@ip) deborde sur + deux lignes, et la couper en plein milieu casserait la detection. + """ + lines = list(self.screen.display) + try: + for entry in self.screen.history.top: + lines.insert(0, "".join(char.data for char in entry.values())) + except Exception: + pass + + joined: list[str] = [] + current = "" + for line in lines: + stripped = line.rstrip() + current += stripped + if len(stripped) >= self.cols: # ligne pleine : la suite continue en dessous + continue + joined.append(current) + current = "" + if current: + joined.append(current) + return "\n".join(joined) + + # ---------- alimentation depuis la connexion SSH ---------- + def feed(self, data: bytes) -> None: + was_scrolled = self.scroll_offset != 0 + self.stream.feed(data) + self.scroll_offset = 0 + + if was_scrolled: + # on revient du scrollback a la vue live : tout le contenu affiche change, + # un repaint partiel base sur les seules lignes "dirty" serait incomplet + self.screen.dirty.clear() + self.update() + return + + dirty_rows = self.screen.dirty + cursor_y = self.screen.cursor.y + if cursor_y != self._prev_cursor_y: + dirty_rows = dirty_rows | {self._prev_cursor_y, cursor_y} + else: + dirty_rows = dirty_rows | {cursor_y} + self._prev_cursor_y = cursor_y + + if dirty_rows: + min_y = max(0, min(dirty_rows)) + max_y = min(self.rows - 1, max(dirty_rows)) + self.update(QRect(0, min_y * self.char_h, self.cols * self.char_w, (max_y - min_y + 1) * self.char_h)) + self.screen.dirty.clear() + + # ---------- rendu ---------- + def paintEvent(self, event) -> None: + painter = QPainter(self) + rect = event.rect() + painter.fillRect(rect, self.default_bg) + painter.setFont(self.font_) + + y0 = max(0, rect.top() // self.char_h) + y1 = min(self.rows - 1, rect.bottom() // self.char_h) + + cursor = self.screen.cursor + for y in range(y0, y1 + 1): + line = self.screen.buffer[y] + x = 0 + while x < self.cols: + char = line[x] + run_text = char.data + start_x = x + fg, bg = self._resolve_colors(char) + # regrouper les caracteres consecutifs de meme style pour limiter les appels de dessin + x += 1 + while x < self.cols: + nxt = line[x] + nfg, nbg = self._resolve_colors(nxt) + if nfg == fg and nbg == bg: + run_text += nxt.data + x += 1 + else: + break + + px = start_x * self.char_w + py = y * self.char_h + run_w = (x - start_x) * self.char_w + if bg != self.default_bg: + painter.fillRect(px, py, run_w, self.char_h, bg) + painter.setPen(fg) + painter.drawText(px, py + self.char_h - 4, run_text) + + # curseur (bloc plein), non affiche pendant un scroll dans l'historique + if self.scroll_offset == 0 and not self.screen.cursor.hidden and y0 <= cursor.y <= y1: + cx, cy = cursor.x, cursor.y + painter.fillRect(cx * self.char_w, cy * self.char_h, self.char_w, self.char_h, self.cursor_color) + + painter.end() + + def _resolve_colors(self, char) -> tuple[QColor, QColor]: + fg_color = self._color_for(char.fg, self.default_fg) + bg_color = self._color_for(char.bg, self.default_bg) + if char.reverse: + fg_color, bg_color = bg_color, fg_color + if char.bold and fg_color is self.default_fg: + fg_color = self._bold_fg + return fg_color, bg_color + + def _color_for(self, value: str, default: QColor) -> QColor: + cached = self._color_cache.get(value) + if cached is not None: + return cached + if isinstance(value, str) and len(value) == 6: + candidate = QColor(f"#{value}") + if candidate.isValid(): + self._color_cache[value] = candidate + return candidate + return default + + # ---------- redimensionnement ---------- + def resizeEvent(self, event) -> None: + new_cols = max(10, self.width() // self.char_w) + new_rows = max(3, self.height() // self.char_h) + if new_cols != self.cols or new_rows != self.rows: + self.cols, self.rows = new_cols, new_rows + self.screen.resize(self.rows, self.cols) + self.resized.emit(self.cols, self.rows) + super().resizeEvent(event) + + # ---------- entree clavier ---------- + def keyPressEvent(self, event: QKeyEvent) -> None: + key = event.key() + modifiers = event.modifiers() + + # Ctrl+lettre = raccourci (Ctrl+C, Ctrl+D...), mais PAS quand Alt est aussi enfonce : + # sur les claviers europeens (AZERTY compris), AltGr - indispensable pour taper + # @ # { } | ~ ... - est rapporte par Windows comme Ctrl+Alt enfonces ensemble. + # Sans cette exclusion, taper "@" (et les lettres voisines pendant le chevauchement + # naturel de la frappe) etait intercepte comme un raccourci Ctrl+lettre parasite, + # envoyant un octet de controle au lieu du vrai caractere. + if (modifiers & Qt.ControlModifier) and not (modifiers & Qt.AltModifier) and Qt.Key_A <= key <= Qt.Key_Z: + self.data_to_send.emit(bytes([key - Qt.Key_A + 1])) + return + + if key in KEY_SEQUENCES: + self.data_to_send.emit(KEY_SEQUENCES[key]) + return + + text = event.text() + if text: + self.data_to_send.emit(text.encode("utf-8", errors="ignore")) + return + super().keyPressEvent(event) + + def wheelEvent(self, event: QWheelEvent) -> None: + steps = event.angleDelta().y() // 120 + if steps > 0: + self.screen.prev_page() + self.scroll_offset += steps + elif steps < 0: + self.screen.next_page() + self.scroll_offset = max(0, self.scroll_offset + steps) + self.update() + + def mousePressEvent(self, event) -> None: + self.setFocus() + super().mousePressEvent(event) diff --git a/app/theme.py b/app/theme.py new file mode 100644 index 0000000..4f4bec5 --- /dev/null +++ b/app/theme.py @@ -0,0 +1,181 @@ +"""Themes de l'application : palette d'interface (QSS) + palette du terminal. + +Trois themes sont fournis (clair par defaut, bleu nuit, noir). Ils partagent le meme +gabarit `resources/theme.qss`, dont les $variables sont remplacees par la palette +choisie : une seule feuille de style a maintenir au lieu de trois quasi identiques. +""" +from __future__ import annotations + +import json +from pathlib import Path +from string import Template + +from PySide6.QtCore import Qt +from PySide6.QtGui import QIcon, QPixmap, QPainter, QColor, QFont, QLinearGradient +from PySide6.QtWidgets import QApplication + +from app.paths import RESOURCES_DIR, DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +STYLESHEET_PATH = RESOURCES_DIR / "theme.qss" +SETTINGS_FILE = DATA_DIR / "settings.json" + +THEME_LIGHT = "light" +THEME_MIDNIGHT = "midnight" +THEME_BLACK = "black" +DEFAULT_THEME = THEME_LIGHT + +# --------------------------------------------------------------------------- +# Palettes d'interface +# --------------------------------------------------------------------------- +UI_PALETTES: dict[str, dict[str, str]] = { + THEME_LIGHT: { + "window": "#f4f6f9", "base": "#ffffff", "panel": "#eef1f6", "elevated": "#e4e9f0", + "input": "#ffffff", "hover": "#dfe6f2", "pressed": "#d2dae8", + "border": "#d3dae4", "border_strong": "#b9c3d1", + "text": "#1c2430", "text_dim": "#5c6875", "text_disabled": "#a3adba", + "accent": "#2f6fed", "accent_hover": "#1f5ed6", "on_accent": "#ffffff", + "scroll": "#c3ccd9", "scroll_hover": "#a7b3c4", + }, + THEME_MIDNIGHT: { + "window": "#0a1220", "base": "#0c1526", "panel": "#0e1a30", "elevated": "#1c2c47", + "input": "#101c30", "hover": "#172542", "pressed": "#16233b", + "border": "#1c2c47", "border_strong": "#2a3c5c", + "text": "#e6ecf6", "text_dim": "#93a4c2", "text_disabled": "#55627c", + "accent": "#2f6fed", "accent_hover": "#4f8cff", "on_accent": "#ffffff", + "scroll": "#2a3c5c", "scroll_hover": "#3a527d", + }, + THEME_BLACK: { + "window": "#000000", "base": "#0a0a0a", "panel": "#141414", "elevated": "#1f1f1f", + "input": "#141414", "hover": "#262626", "pressed": "#0f0f0f", + "border": "#2b2b2b", "border_strong": "#3d3d3d", + "text": "#e8e8e8", "text_dim": "#9a9a9a", "text_disabled": "#5a5a5a", + "accent": "#4a9eff", "accent_hover": "#6fb3ff", "on_accent": "#000000", + "scroll": "#3a3a3a", "scroll_hover": "#525252", + }, +} + +# --------------------------------------------------------------------------- +# Palettes du terminal (16 couleurs ANSI + defauts + curseur) +# --------------------------------------------------------------------------- +TERMINAL_PALETTES: dict[str, dict] = { + THEME_LIGHT: { + "fg": "#1c2430", "bg": "#ffffff", "cursor": (47, 111, 237, 110), + "ansi": { + "black": "#1c2430", "red": "#c0392b", "green": "#2e7d32", "brown": "#a06800", + "blue": "#1f5ed6", "magenta": "#8e24aa", "cyan": "#00838f", "white": "#5c6875", + "brightblack": "#8a94a3", "brightred": "#e04b3a", "brightgreen": "#3ca142", + "brightbrown": "#c98a00", "brightblue": "#2f6fed", "brightmagenta": "#a944c7", + "brightcyan": "#0aa2b0", "brightwhite": "#1c2430", + }, + }, + THEME_MIDNIGHT: { + "fg": "#e6ecf6", "bg": "#0b1220", "cursor": (79, 140, 255, 140), + "ansi": { + "black": "#0b1220", "red": "#ff6b81", "green": "#8ce99a", "brown": "#f2c94c", + "blue": "#4f8cff", "magenta": "#c792ea", "cyan": "#63d4e0", "white": "#dbe4f5", + "brightblack": "#3a4a68", "brightred": "#ff8a9b", "brightgreen": "#a9e6b5", + "brightbrown": "#f5d778", "brightblue": "#7aa9ff", "brightmagenta": "#dcb3f0", + "brightcyan": "#8fe3ec", "brightwhite": "#ffffff", + }, + }, + THEME_BLACK: { + "fg": "#e8e8e8", "bg": "#000000", "cursor": (74, 158, 255, 140), + "ansi": { + "black": "#000000", "red": "#e05561", "green": "#8cc265", "brown": "#d18f52", + "blue": "#4aa5f0", "magenta": "#c162de", "cyan": "#42b3c2", "white": "#d0d0d0", + "brightblack": "#5a5a5a", "brightred": "#ff6b78", "brightgreen": "#a5e075", + "brightbrown": "#e5a663", "brightblue": "#6fb3ff", "brightmagenta": "#d178ee", + "brightcyan": "#5fc9d8", "brightwhite": "#ffffff", + }, + }, +} + +THEME_LABELS = { + THEME_LIGHT: "Clair", + THEME_MIDNIGHT: "Bleu nuit", + THEME_BLACK: "Noir", +} + +_current_theme = DEFAULT_THEME + + +# --------------------------------------------------------------------------- +# Preference persistante +# --------------------------------------------------------------------------- +def load_saved_theme() -> str: + try: + data = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) + theme = data.get("theme") + if theme in UI_PALETTES: + return theme + except (OSError, json.JSONDecodeError): + pass + return DEFAULT_THEME + + +def save_theme(theme: str) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + data = {} + try: + data = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + pass + data["theme"] = theme + try: + SETTINGS_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") + except OSError: + pass + + +def current_theme() -> str: + return _current_theme + + +def terminal_palette(theme: str | None = None) -> dict: + return TERMINAL_PALETTES.get(theme or _current_theme, TERMINAL_PALETTES[DEFAULT_THEME]) + + +# --------------------------------------------------------------------------- +# Application du theme +# --------------------------------------------------------------------------- +def apply_theme(app: QApplication, theme: str = DEFAULT_THEME) -> None: + """Bascule sur le style Fusion (necessaire pour que la QSS s'applique + uniformement, y compris aux widgets stylises nativement par Windows) puis + construit la feuille de style a partir de la palette demandee.""" + global _current_theme + if theme not in UI_PALETTES: + theme = DEFAULT_THEME + _current_theme = theme + + app.setStyle("Fusion") + try: + template = Template(STYLESHEET_PATH.read_text(encoding="utf-8")) + app.setStyleSheet(template.safe_substitute(UI_PALETTES[theme])) + except OSError: + pass + + +def build_app_icon() -> QIcon: + """Genere une icone simple (pas de fichier externe requis) : carre + prompt '>_'""" + size = 64 + pixmap = QPixmap(size, size) + pixmap.fill(Qt.transparent) + + painter = QPainter(pixmap) + painter.setRenderHint(QPainter.Antialiasing) + + gradient = QLinearGradient(0, 0, size, size) + gradient.setColorAt(0, QColor("#182a4a")) + gradient.setColorAt(1, QColor("#0a1220")) + painter.setBrush(gradient) + painter.setPen(QColor("#2a3c5c")) + painter.drawRoundedRect(2, 2, size - 4, size - 4, 14, 14) + + font = QFont("Consolas") + font.setBold(True) + font.setPointSize(24) + painter.setFont(font) + painter.setPen(QColor("#4f8cff")) + painter.drawText(pixmap.rect(), Qt.AlignCenter, ">_") + + painter.end() + return QIcon(pixmap) diff --git a/build_exe.bat b/build_exe.bat new file mode 100644 index 0000000..3e41b99 --- /dev/null +++ b/build_exe.bat @@ -0,0 +1,32 @@ +@echo off +setlocal +cd /d "%~dp0" + +if not exist venv ( + echo L'environnement virtuel n'existe pas. Lancez d'abord setup.bat + pause + exit /b 1 +) + +call venv\Scripts\activate.bat + +echo Installation de PyInstaller si necessaire... +python -m pip install pyinstaller --quiet + +echo Generation de l'icone... +python tools\make_icon.py + +echo. +echo Construction de l'application portable (plusieurs minutes)... +python -m PyInstaller --noconfirm --clean PySSHClient.spec +if errorlevel 1 ( + echo. + echo La construction a echoue. + pause + exit /b 1 +) + +echo. +echo Termine : le dossier "dist\PySSH Client" est autonome. +echo Copiez-le ou vous voulez, il fonctionne sans Python installe. +pause diff --git a/data/keys/.gitkeep b/data/keys/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/main.py b/main.py new file mode 100644 index 0000000..ca8981c --- /dev/null +++ b/main.py @@ -0,0 +1,25 @@ +"""Point d'entree de l'application PySSH Client.""" +import sys + +from PySide6.QtWidgets import QApplication + +from app.main_window import MainWindow +from app.theme import apply_theme, build_app_icon, load_saved_theme + + +def main() -> None: + app = QApplication(sys.argv) + app.setApplicationName("PySSH Client") + # theme clair par defaut au premier lancement, sinon celui choisi precedemment + apply_theme(app, load_saved_theme()) + icon = build_app_icon() + app.setWindowIcon(icon) + + window = MainWindow() + window.setWindowIcon(icon) + window.show() + sys.exit(app.exec()) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..724028b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +PySide6>=6.7 +paramiko>=3.4 +pyte>=0.8.2 +cryptography>=42.0 +pywinpty>=2.0 diff --git a/resources/theme.qss b/resources/theme.qss new file mode 100644 index 0000000..8358fa4 --- /dev/null +++ b/resources/theme.qss @@ -0,0 +1,342 @@ +/* ============================================================ + PySSH Client - gabarit de theme + Les jetons prefixes du signe dollar sont remplaces par la palette choisie + (voir app/theme.py). + Applique sur le style de base Fusion. + ============================================================ */ + +* { + outline: none; + font-family: "Segoe UI", "Cascadia Mono", sans-serif; + font-size: 13px; +} + +QWidget { + background-color: $base; + color: $text; + selection-background-color: $accent; + selection-color: $on_accent; +} + +QMainWindow, QDialog { + background-color: $window; +} + +/* ---------- Barre de menu ---------- */ +QMenuBar { + background-color: $panel; + color: $text; + border-bottom: 1px solid $border; + padding: 2px 4px; +} +QMenuBar::item { + background: transparent; + padding: 6px 12px; + border-radius: 4px; +} +QMenuBar::item:selected { + background-color: $hover; +} +QMenu { + background-color: $panel; + border: 1px solid $border; + padding: 4px; + border-radius: 6px; +} +QMenu::item { + padding: 6px 24px 6px 12px; + border-radius: 4px; + color: $text; +} +QMenu::item:selected { + background-color: $accent; + color: $on_accent; +} +QMenu::separator { + height: 1px; + background: $border; + margin: 4px 8px; +} + +/* ---------- Barre de statut ---------- */ +QStatusBar { + background-color: $panel; + color: $text_dim; + border-top: 1px solid $border; +} + +/* ---------- Dock / sidebar ---------- */ +QDockWidget { + color: $text; + titlebar-close-icon: none; + border: none; +} +QDockWidget::title { + background-color: $panel; + padding: 8px 10px; + font-weight: 600; + border-bottom: 1px solid $border; +} + +/* ---------- Arborescence des sessions ---------- */ +QTreeWidget { + background-color: $panel; + border: none; + border-right: 1px solid $border; + padding: 4px; + show-decoration-selected: 1; +} +QTreeWidget::item { + padding: 6px 4px; + border-radius: 5px; + margin: 1px 2px; +} +QTreeWidget::item:hover { + background-color: $hover; +} +QTreeWidget::item:selected { + background-color: $accent; + color: $on_accent; +} +QTreeWidget::branch { + background: transparent; +} +QHeaderView::section { + background-color: $elevated; + color: $text_dim; + border: none; + border-right: 1px solid $border; + border-bottom: 1px solid $border; + padding: 5px 6px; +} + +/* ---------- Onglets ---------- */ +QTabWidget::pane { + border: none; + background-color: $window; + top: -1px; +} +QTabBar { + background-color: $panel; +} +QTabBar::tab { + background-color: $panel; + color: $text_dim; + padding: 9px 18px; + margin-right: 2px; + border-top-left-radius: 8px; + border-top-right-radius: 8px; + border: 1px solid $border; + border-bottom: none; +} +QTabBar::tab:selected { + background-color: $window; + color: $text; + border-bottom: 2px solid $accent; +} +QTabBar::tab:hover:!selected { + background-color: $hover; + color: $text; +} +QTabBar::close-button { + subcontrol-position: right; +} + +/* ---------- Boutons ---------- */ +QPushButton { + background-color: $elevated; + color: $text; + border: 1px solid $border_strong; + border-radius: 6px; + padding: 7px 16px; +} +QPushButton:hover { + background-color: $hover; + border-color: $accent; +} +QPushButton:pressed { + background-color: $pressed; +} +QPushButton:disabled { + color: $text_disabled; + background-color: $base; +} +QPushButton#primary, QDialogButtonBox QPushButton[default="true"] { + background-color: $accent; + border-color: $accent_hover; + color: $on_accent; + font-weight: 600; +} +QPushButton#primary:hover, QDialogButtonBox QPushButton[default="true"]:hover { + background-color: $accent_hover; +} + +QToolButton { + background-color: transparent; + color: $text; + border: 1px solid transparent; + border-radius: 6px; + padding: 6px 10px; +} +QToolButton:hover { + background-color: $hover; + border-color: $border_strong; +} +QToolButton:pressed, QToolButton:checked { + background-color: $accent; + color: $on_accent; +} +QToolButton::menu-indicator { + subcontrol-position: right center; + subcontrol-origin: padding; + right: 2px; +} +QToolButton#primary { + background-color: $accent; + border: 1px solid $accent_hover; + color: $on_accent; + font-weight: 600; +} +QToolButton#primary:hover { + background-color: $accent_hover; +} + +/* ---------- Barre superieure des onglets terminal ---------- */ +QFrame#tabBar { + background-color: $panel; + border-bottom: 1px solid $border; +} +QLabel#statusLabel { + color: $text_dim; + font-weight: 500; +} + +/* ---------- Champs de saisie ---------- */ +QLineEdit, QTextEdit, QPlainTextEdit, QSpinBox, QComboBox { + background-color: $input; + color: $text; + border: 1px solid $border_strong; + border-radius: 5px; + padding: 5px 8px; + selection-background-color: $accent; + selection-color: $on_accent; +} +QLineEdit:focus, QTextEdit:focus, QSpinBox:focus, QComboBox:focus { + border: 1px solid $accent; +} +QComboBox::drop-down { + border: none; + width: 22px; +} +QComboBox QAbstractItemView { + background-color: $input; + color: $text; + border: 1px solid $border_strong; + selection-background-color: $accent; + selection-color: $on_accent; + outline: none; +} +QSpinBox::up-button, QSpinBox::down-button { + background-color: $elevated; + width: 16px; + border-left: 1px solid $border; +} +QCheckBox { + color: $text; + spacing: 6px; +} + +/* ---------- Listes ---------- */ +QListWidget { + background-color: $panel; + border: 1px solid $border; + border-radius: 6px; + padding: 2px; +} +QListWidget::item { + padding: 5px 6px; + border-radius: 4px; +} +QListWidget::item:hover { + background-color: $hover; +} +QListWidget::item:selected { + background-color: $accent; + color: $on_accent; +} + +/* ---------- Splitter ---------- */ +QSplitter::handle { + background-color: $border; +} +QSplitter::handle:horizontal { + width: 2px; +} +QSplitter::handle:vertical { + height: 2px; +} +QSplitter::handle:hover { + background-color: $accent; +} + +/* ---------- Barres de defilement ---------- */ +QScrollBar:vertical { + background: $base; + width: 12px; + margin: 0; +} +QScrollBar::handle:vertical { + background: $scroll; + min-height: 24px; + border-radius: 5px; + margin: 2px; +} +QScrollBar::handle:vertical:hover { + background: $scroll_hover; +} +QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { + height: 0; +} +QScrollBar:horizontal { + background: $base; + height: 12px; +} +QScrollBar::handle:horizontal { + background: $scroll; + min-width: 24px; + border-radius: 5px; + margin: 2px; +} +QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal { + width: 0; +} + +/* ---------- Barres de progression ---------- */ +QProgressBar { + background-color: $input; + border: 1px solid $border_strong; + border-radius: 5px; + text-align: center; + color: $text; +} +QProgressBar::chunk { + background-color: $accent; + border-radius: 5px; +} + +/* ---------- Etiquettes ---------- */ +QLabel { + background: transparent; +} +QLabel[heading="true"] { + font-size: 15px; + font-weight: 600; + color: $text; +} + +QToolTip { + background-color: $elevated; + color: $text; + border: 1px solid $border_strong; + padding: 4px 6px; +} diff --git a/run.bat b/run.bat new file mode 100644 index 0000000..2902356 --- /dev/null +++ b/run.bat @@ -0,0 +1,12 @@ +@echo off +cd /d "%~dp0" + +if not exist venv ( + echo L'environnement virtuel n'existe pas. Lancez d'abord setup.bat + pause + exit /b 1 +) + +call venv\Scripts\activate.bat +python main.py +if errorlevel 1 pause diff --git a/setup.bat b/setup.bat new file mode 100644 index 0000000..0cc87e2 --- /dev/null +++ b/setup.bat @@ -0,0 +1,23 @@ +@echo off +setlocal +cd /d "%~dp0" + +if not exist venv ( + echo Creation de l'environnement virtuel... + py -3 -m venv venv + if errorlevel 1 ( + echo Erreur: impossible de creer le venv. Verifiez que Python est installe et accessible via "py -3". + pause + exit /b 1 + ) +) + +call venv\Scripts\activate.bat + +echo Installation des dependances... +python -m pip install --upgrade pip +pip install -r requirements.txt + +echo. +echo Installation terminee. Lancez run.bat pour demarrer l'application. +pause diff --git a/tools/create_shortcut.ps1 b/tools/create_shortcut.ps1 new file mode 100644 index 0000000..1d1f2bc --- /dev/null +++ b/tools/create_shortcut.ps1 @@ -0,0 +1,50 @@ +# Cree un raccourci Windows pour lancer l'application sans fenetre de console. +# +# Usage : clic droit sur ce fichier > "Executer avec PowerShell" +# ou : powershell -ExecutionPolicy Bypass -File tools\create_shortcut.ps1 +# +# Options : +# -Desktop place aussi une copie sur le Bureau +# -StartMenu place aussi une copie dans le menu Demarrer + +param( + [switch]$Desktop, + [switch]$StartMenu +) + +$root = Split-Path -Parent $PSScriptRoot +$target = Join-Path $root "venv\Scripts\pythonw.exe" # pythonw = pas de console noire +$script = Join-Path $root "main.py" +$icon = Join-Path $root "resources\app.ico" + +if (-not (Test-Path $target)) { + Write-Host "Environnement virtuel introuvable. Lancez d'abord setup.bat." -ForegroundColor Red + exit 1 +} +if (-not (Test-Path $icon)) { + # l'icone se regenere depuis l'application elle-meme + & (Join-Path $root "venv\Scripts\python.exe") (Join-Path $root "tools\make_icon.py") +} + +$shell = New-Object -ComObject WScript.Shell + +function New-AppShortcut([string]$path) { + $lnk = $shell.CreateShortcut($path) + $lnk.TargetPath = $target + $lnk.Arguments = '"' + $script + '"' + $lnk.WorkingDirectory = $root + $lnk.IconLocation = $icon + $lnk.Description = "Client SSH/SFTP personnel" + $lnk.Save() + Write-Host "Raccourci cree : $path" -ForegroundColor Green +} + +New-AppShortcut (Join-Path $root "PySSH Client.lnk") + +if ($Desktop) { + New-AppShortcut (Join-Path ([Environment]::GetFolderPath("Desktop")) "PySSH Client.lnk") +} +if ($StartMenu) { + $programs = Join-Path ([Environment]::GetFolderPath("StartMenu")) "Programs" + New-AppShortcut (Join-Path $programs "PySSH Client.lnk") +} diff --git a/tools/make_icon.py b/tools/make_icon.py new file mode 100644 index 0000000..1ca9402 --- /dev/null +++ b/tools/make_icon.py @@ -0,0 +1,34 @@ +"""Genere resources/app.ico a partir de l'icone dessinee par l'application. + +A relancer seulement si l'icone change : le fichier .ico est necessaire au +raccourci Windows (et a un futur packaging), alors que l'application dessine +la sienne en memoire au demarrage. + +Usage : venv\\Scripts\\python.exe tools\\make_icon.py +""" +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from PySide6.QtCore import QSize # noqa: E402 +from PySide6.QtGui import QIcon # noqa: E402 +from PySide6.QtWidgets import QApplication # noqa: E402 + +from app.theme import build_app_icon # noqa: E402 + + +def main() -> None: + app = QApplication(sys.argv) # noqa: F841 - requis pour manipuler des QPixmap + icon: QIcon = build_app_icon() + target = ROOT / "resources" / "app.ico" + # plusieurs tailles pour un rendu net dans la barre des taches et l'explorateur + pixmap = icon.pixmap(QSize(256, 256)) + if not pixmap.save(str(target), "ICO"): + raise SystemExit(f"Echec de l'ecriture de {target}") + print(f"Icone generee : {target}") + + +if __name__ == "__main__": + main()