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 <noreply@anthropic.com>
This commit is contained in:
snoopy
2026-09-04 06:50:10 +02:00
commit b977842186
28 changed files with 3946 additions and 0 deletions

0
app/__init__.py Normal file
View File

0
app/dialogs/__init__.py Normal file
View File

View File

@@ -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

198
app/dialogs/key_dialog.py Normal file
View File

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

View File

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

View File

@@ -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

201
app/key_manager.py Normal file
View File

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

View File

@@ -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))

151
app/local_terminal.py Normal file
View File

@@ -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

720
app/main_window.py Normal file
View File

@@ -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<args>[^\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)

49
app/paths.py Normal file
View File

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

83
app/session_manager.py Normal file
View File

@@ -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})

673
app/sftp_panel.py Normal file
View File

@@ -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"<b>{title}</b>")
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()

336
app/ssh_connection.py Normal file
View File

@@ -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

266
app/terminal_widget.py Normal file
View File

@@ -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)

181
app/theme.py Normal file
View File

@@ -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)