f537868da3
Zoom de police : Ctrl+molette et Ctrl+ +/-, Ctrl+0 pour revenir a la taille par defaut. La taille choisie est memorisee entre deux lancements. Barre de defilement de l'historique. pyte ne sait defiler que par demi-pages : la barre est donc graduee en pages plutot qu'en lignes, ce qui donne un repere visuel de position et permet de faire glisser. Maj+PagePrec/Suiv au clavier, Ctrl+Fin pour revenir a la vue en direct. Recherche dans l'historique (Ctrl+F) : la barre apparait en bas de l'onglet et la recherche remonte l'historique page par page jusqu'a trouver, en selectionnant la correspondance — donc visible et directement copiable. Reconnexion en un clic : un bouton apparait dans la barre de l'onglet des que la connexion tombe. La fabrication de la connexion a ete extraite dans _create_connection(), ce qui permet de la recreer a l'identique sans ressaisir quoi que ce soit. Utile avec des serveurs de test qui redemarrent souvent. Les hotes deja approuves par la commande ssh (~/.ssh/known_hosts) sont desormais reconnus, ce qui evite de reconfirmer une empreinte que votre ssh connait deja. paramiko ne reecrit jamais ce fichier : les nouvelles cles continuent d'aller dans celui de l'application. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
344 lines
15 KiB
Python
344 lines
15 KiB
Python
"""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)
|
|
# Les hotes deja approuves par la commande ssh (~/.ssh/known_hosts) sont
|
|
# reconnus : inutile de reconfirmer une empreinte que votre ssh connait deja.
|
|
# paramiko ne reecrit jamais ce fichier ; les nouvelles cles vont dans le notre.
|
|
try:
|
|
self.client.load_system_host_keys()
|
|
except Exception:
|
|
pass
|
|
if KNOWN_HOSTS_FILE.exists():
|
|
self.client.load_host_keys(str(KNOWN_HOSTS_FILE))
|
|
self.client.set_missing_host_key_policy(TrustPrompt(self))
|
|
|
|
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
|