pyssh-client/app/session_manager.py

"""Stockage des sessions SSH (serveurs enregistres) dans un fichier JSON local."""
from __future__ import annotations

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)
from app.storage.json_store import load_json_document, save_json_document

SESSIONS_FILE = DATA_DIR / "sessions.json"
SCHEMA_VERSION = 1

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"

# Sources possibles d'un jump host (ProxyJump / bastion) - voir BastionConfig.
BASTION_NONE = "none"        # connexion directe, sans bastion
BASTION_SESSION = "session"  # reutilise une session PySSH existante (par id)
BASTION_OPENSSH = "openssh"  # reutilise un Host de ~/.ssh/config (par alias)
BASTION_MANUAL = "manual"    # parametres saisis directement


@dataclass
class BastionConfig:
    """Jump host d'une session. Un seul saut pour l'instant (mode != "chain"),
    mais rien n'empeche d'ajouter plus tard une liste de sauts a la place d'un
    bastion unique : le mecanisme de resolution (app.bastion) traite deja ceci
    comme "le saut a franchir avant la cible", independamment du nombre total.

    Jamais de mot de passe stocke ici : comme pour Session.auth_type, un bastion
    en mode mot de passe le redemande a la connexion."""
    mode: str = BASTION_NONE
    session_id: str = ""      # mode == BASTION_SESSION
    openssh_alias: str = ""   # mode == BASTION_OPENSSH
    host: str = ""            # mode == BASTION_MANUAL
    port: int = 22
    username: str = ""
    auth_type: str = AUTH_AUTO
    key_path: str = ""

    def to_dict(self) -> dict:
        return asdict(self)

    @staticmethod
    def from_dict(data: dict) -> "BastionConfig":
        b = BastionConfig()
        for k in BastionConfig.__dataclass_fields__:
            if k in data:
                setattr(b, k, data[k])
        return b


@dataclass
class Session:
    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 = ""
    bastion: BastionConfig = field(default_factory=BastionConfig)

    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():
            if k == "bastion" and isinstance(v, dict):
                v = BastionConfig.from_dict(v)
            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_warning: str | None = None
        self.load()

    def load(self) -> None:
        was_missing = not self.path.exists()
        document, warning = load_json_document(
            self.path,
            default={"schema_version": SCHEMA_VERSION, "sessions": []},
            current_version=SCHEMA_VERSION,
        )
        self.load_warning = warning
        self.sessions = [Session.from_dict(s) for s in document.get("sessions", [])]
        if was_missing:
            self.save()

    def save(self) -> None:
        payload = {"schema_version": SCHEMA_VERSION, "sessions": [s.to_dict() for s in self.sessions]}
        save_json_document(self.path, payload)

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