Une session peut desormais passer par un jump host avant d'atteindre sa cible, avec 3 sources possibles (limite a un seul saut pour l'instant) : une session PySSH existante, un Host de ~/.ssh/config, ou une configuration manuelle (hote/port/utilisateur/authentification/cle). - session_manager.py : BastionConfig (mode + parametres), champ Session.bastion. Jamais de mot de passe stocke : un bastion en mode "mot de passe" le redemande a la connexion, comme le fait deja Session.auth_type. - app/bastion.py : resolve_bastion_target() convertit un BastionConfig en parametres de connexion concrets quelle que soit sa source, et leve une erreur claire si la reference est cassee (session supprimee, alias OpenSSH introuvable, config manuelle incomplete) plutot que d'echouer plus tard avec un message obscur. - ssh_connection.py : la logique de connexion (auth, cle d'hote modifiee, mot de passe de secours en mode auto) est factorisee dans _connect_host(), reutilisee a l'identique pour le bastion et pour la cible. _open_bastion_channel() connecte d'abord le bastion puis ouvre un canal direct-tcpip vers la cible, passe en sock= a la connexion finale (mecanisme ProxyJump standard de paramiko). Le bastion a son propre SSHClient, ferme proprement a la fin de la connexion. - SessionDialog : section "Bastion / Jump Host" (choix de la source + champs associes, caches/affiches via un QStackedWidget). - ConnectionController resout le bastion et demande son mot de passe (si nécessaire) avant d'ouvrir l'onglet ; erreur affichee immediatement si la reference est invalide plutot que de lancer une connexion vouee a l'echec. - app/openssh/config.py : un Host avec ProxyJump produit maintenant une Session dont le bastion reference cet alias (mode BASTION_OPENSSH) : le scenario proxmox-prod / bastion de la Phase 2 fonctionne de bout en bout. tests/test_bastion.py (resolution des 3 sources + erreurs, et ouverture du canal via un bastion entierement mocke, sans reseau reel) et tests/test_session_dialog_bastion.py (UI : construction, pre-remplissage a l'edition, resultat par mode). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Lecture de ~/.ssh/config (format OpenSSH), via paramiko.SSHConfig.
|
|
|
|
Support minimum couvert : Host, HostName, User, Port, IdentityFile, ProxyJump,
|
|
ForwardAgent, ServerAliveInterval, ServerAliveCountMax.
|
|
|
|
Le ProxyJump detecte ici est seulement affiche/reporte pour l'instant : la
|
|
connexion via un jump host est prise en charge par le module dedie (bastion),
|
|
pas par ce module qui ne fait que lire la configuration.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import paramiko
|
|
|
|
from app.session_manager import Session, BastionConfig, AUTH_AUTO, AUTH_KEY, BASTION_OPENSSH
|
|
|
|
|
|
@dataclass
|
|
class OpenSSHHost:
|
|
alias: str
|
|
hostname: str = ""
|
|
user: str = ""
|
|
port: int = 22
|
|
identity_files: list[str] = field(default_factory=list)
|
|
proxy_jump: str = ""
|
|
forward_agent: bool = False
|
|
server_alive_interval: int | None = None
|
|
server_alive_count_max: int | None = None
|
|
|
|
@property
|
|
def display_target(self) -> str:
|
|
target = self.hostname or self.alias
|
|
return f"{target}:{self.port}" if self.port != 22 else target
|
|
|
|
|
|
def default_config_path() -> Path:
|
|
return Path.home() / ".ssh" / "config"
|
|
|
|
|
|
def _as_int(value: str | None) -> int | None:
|
|
try:
|
|
return int(value) if value is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def load_openssh_hosts(path: Path | None = None) -> list[OpenSSHHost]:
|
|
"""Renvoie les Host detectes dans le fichier, tries par alias.
|
|
|
|
Un fichier absent (tres courant : rien ne force un utilisateur Windows a en
|
|
avoir un) n'est pas une erreur : la liste est simplement vide."""
|
|
path = path or default_config_path()
|
|
if not path.exists():
|
|
return []
|
|
|
|
parser = paramiko.SSHConfig()
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
parser.parse(f)
|
|
except OSError:
|
|
return []
|
|
|
|
hosts: list[OpenSSHHost] = []
|
|
for alias in parser.get_hostnames():
|
|
# "*" et les motifs (glob) ne designent pas un hote precis auquel se
|
|
# connecter, seulement des valeurs par defaut a fusionner dans les autres.
|
|
if alias == "*" or "*" in alias or "?" in alias:
|
|
continue
|
|
looked_up = parser.lookup(alias)
|
|
hosts.append(OpenSSHHost(
|
|
alias=alias,
|
|
hostname=looked_up.get("hostname", alias),
|
|
user=looked_up.get("user", ""),
|
|
port=_as_int(looked_up.get("port")) or 22,
|
|
identity_files=list(looked_up.get("identityfile", []) or []),
|
|
proxy_jump=looked_up.get("proxyjump", ""),
|
|
forward_agent=str(looked_up.get("forwardagent", "")).lower() == "yes",
|
|
server_alive_interval=_as_int(looked_up.get("serveraliveinterval")),
|
|
server_alive_count_max=_as_int(looked_up.get("serveralivecountmax")),
|
|
))
|
|
hosts.sort(key=lambda h: h.alias.lower())
|
|
return hosts
|
|
|
|
|
|
def session_from_openssh_host(host: OpenSSHHost) -> Session:
|
|
"""Convertit un Host OpenSSH en Session PySSH (utilisation directe ou import).
|
|
|
|
Si ProxyJump est present, le bastion resultant reference l'alias OpenSSH
|
|
correspondant (mode BASTION_OPENSSH) : il sera resolu et applique a la
|
|
connexion par app.bastion, tant que cet alias reste present dans
|
|
~/.ssh/config (voir app.bastion.resolve_bastion_target)."""
|
|
session = Session(
|
|
name=host.alias,
|
|
group="OpenSSH",
|
|
host=host.hostname or host.alias,
|
|
port=host.port,
|
|
username=host.user,
|
|
)
|
|
if host.identity_files:
|
|
session.auth_type = AUTH_KEY
|
|
session.key_path = str(Path(host.identity_files[0]).expanduser())
|
|
else:
|
|
session.auth_type = AUTH_AUTO
|
|
|
|
notes = [f"Importe depuis ~/.ssh/config (Host {host.alias})."]
|
|
if host.proxy_jump:
|
|
session.bastion = BastionConfig(mode=BASTION_OPENSSH, openssh_alias=host.proxy_jump)
|
|
notes.append(f"ProxyJump vers « {host.proxy_jump} » (voir section Bastion / Jump Host).")
|
|
session.notes = "\n".join(notes)
|
|
return session
|