diff --git a/app/known_hosts_manager.py b/app/known_hosts_manager.py index d265fad..ed38717 100644 --- a/app/known_hosts_manager.py +++ b/app/known_hosts_manager.py @@ -7,12 +7,15 @@ a editer le fichier a la main. """ from __future__ import annotations +import os from dataclasses import dataclass from pathlib import Path import paramiko from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +from app.storage.atomic_file import atomic_write_text + KNOWN_HOSTS_FILE = DATA_DIR / "known_hosts" @@ -64,9 +67,13 @@ class KnownHostsStore: self._save(hk) def clear_all(self) -> None: - DATA_DIR.mkdir(parents=True, exist_ok=True) - self.path.write_text("", encoding="utf-8") + atomic_write_text(self.path, "") def _save(self, hk: paramiko.HostKeys) -> None: + # paramiko.HostKeys.save() ecrit directement sur le chemin donne ; on le fait + # ecrire dans un fichier temporaire puis on bascule dessus atomiquement, pour + # ne jamais laisser known_hosts tronque en cas de coupure pendant l'ecriture. DATA_DIR.mkdir(parents=True, exist_ok=True) - hk.save(str(self.path)) + tmp_path = self.path.with_name(self.path.name + ".tmp") + hk.save(str(tmp_path)) + os.replace(tmp_path, self.path) diff --git a/app/session_manager.py b/app/session_manager.py index e1f0fbc..796e7a7 100644 --- a/app/session_manager.py +++ b/app/session_manager.py @@ -1,13 +1,15 @@ """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) +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 @@ -45,24 +47,24 @@ class SessionStore: 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: - DATA_DIR.mkdir(parents=True, exist_ok=True) - if not self.path.exists(): - self.sessions = [] + 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() - 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") + 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) diff --git a/app/storage/__init__.py b/app/storage/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/storage/atomic_file.py b/app/storage/atomic_file.py new file mode 100644 index 0000000..838945a --- /dev/null +++ b/app/storage/atomic_file.py @@ -0,0 +1,24 @@ +"""Ecriture atomique de fichiers : un crash ou une coupure de courant pendant +l'ecriture ne doit jamais laisser un fichier tronque ou corrompu a la place de +l'ancien, valide.""" +from __future__ import annotations + +import os +from pathlib import Path + + +def atomic_write_bytes(path: Path, content: bytes) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_name(path.name + ".tmp") + with open(tmp_path, "wb") as f: + f.write(content) + f.flush() + os.fsync(f.fileno()) + # os.replace() est atomique y compris sous Windows (contrairement a os.rename(), + # qui y echoue si la destination existe deja). + os.replace(tmp_path, path) + + +def atomic_write_text(path: Path, content: str, encoding: str = "utf-8") -> None: + atomic_write_bytes(path, content.encode(encoding)) diff --git a/app/storage/json_store.py b/app/storage/json_store.py new file mode 100644 index 0000000..43c6382 --- /dev/null +++ b/app/storage/json_store.py @@ -0,0 +1,81 @@ +"""Chargement/sauvegarde de documents JSON versionnes (sessions, cles, ...). + +Chaque document porte un `schema_version`. `migrations` associe une version de +depart a une fonction qui transforme le document vers la version suivante ; le +chargement applique la chaine jusqu'a `current_version`, ce qui permet de faire +evoluer le format plus tard sans invalider les fichiers existants. + +La sauvegarde est atomique (voir atomic_file) et conserve la version precedente +dans un `.bak` : si l'ecriture suivante produit un fichier corrompu (crash, +disque plein), le chargement suivant retombe automatiquement sur cette sauvegarde +plutot que de perdre les donnees de l'utilisateur. +""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Callable + +from app.storage.atomic_file import atomic_write_bytes + +MigrationMap = dict[int, Callable[[dict], dict]] + + +def _bak_path(path: Path) -> Path: + return path.with_name(path.name + ".bak") + + +def _migrate(document: dict, current_version: int, migrations: MigrationMap) -> dict: + version = int(document.get("schema_version", 0) or 0) + while version < current_version: + migrate = migrations.get(version) + if migrate is None: + break # pas de chemin de migration connu : le document est garde tel quel + document = migrate(document) + version = int(document.get("schema_version", version + 1)) + document["schema_version"] = current_version + return document + + +def load_json_document( + path: Path, + *, + default: dict, + current_version: int, + migrations: MigrationMap | None = None, +) -> tuple[dict, str | None]: + """Renvoie (document, avertissement). `avertissement` est None si le fichier + principal a ete lu normalement ; sinon il decrit la recuperation effectuee + (restauration depuis .bak, ou repli sur les valeurs par defaut).""" + migrations = migrations or {} + warning: str | None = None + + for candidate, is_backup in ((Path(path), False), (_bak_path(Path(path)), True)): + if not candidate.exists(): + continue + try: + raw = json.loads(candidate.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError) as exc: + warning = f"{candidate.name} illisible ({exc})" + continue + if not isinstance(raw, dict): + warning = f"{candidate.name} invalide (contenu inattendu)" + continue + if is_backup: + warning = f"{Path(path).name} illisible, restaure depuis {candidate.name}" + return _migrate(raw, current_version, migrations), warning + + if warning is None and not Path(path).exists(): + pass # premier lancement : pas d'avertissement, juste les valeurs par defaut + return _migrate(dict(default), current_version, migrations), warning + + +def save_json_document(path: Path, document: dict) -> None: + path = Path(path) + if path.exists(): + try: + atomic_write_bytes(_bak_path(path), path.read_bytes()) + except OSError: + pass + payload = json.dumps(document, indent=2, ensure_ascii=False) + atomic_write_bytes(path, payload.encode("utf-8")) diff --git a/app/theme.py b/app/theme.py index 4e66e3f..9176098 100644 --- a/app/theme.py +++ b/app/theme.py @@ -15,6 +15,8 @@ from PySide6.QtGui import QIcon, QPixmap, QPainter, QColor, QFont, QLinearGradie from PySide6.QtWidgets import QApplication from app.paths import RESOURCES_DIR, DATA_DIR # noqa: E402 (chemins valables aussi en .exe) +from app.storage.atomic_file import atomic_write_text + STYLESHEET_PATH = RESOURCES_DIR / "theme.qss" SETTINGS_FILE = DATA_DIR / "settings.json" @@ -116,17 +118,7 @@ def load_saved_theme() -> str: 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 + _update_settings("theme", theme) DEFAULT_FONT_SIZE = 10 @@ -146,7 +138,6 @@ def save_font_size(size: int) -> None: def _update_settings(key: str, value) -> None: - DATA_DIR.mkdir(parents=True, exist_ok=True) data = {} try: data = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) @@ -154,7 +145,7 @@ def _update_settings(key: str, value) -> None: pass data[key] = value try: - SETTINGS_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") + atomic_write_text(SETTINGS_FILE, json.dumps(data, indent=2)) except OSError: pass diff --git a/tests/test_atomic_storage.py b/tests/test_atomic_storage.py new file mode 100644 index 0000000..abe5228 --- /dev/null +++ b/tests/test_atomic_storage.py @@ -0,0 +1,116 @@ +"""Stockage JSON : ecriture atomique, schema_version, migration, recuperation +automatique depuis .bak quand le fichier principal est corrompu.""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +import json +import tempfile +from pathlib import Path + +from app.storage.atomic_file import atomic_write_text +from app.storage.json_store import load_json_document, save_json_document +from app.session_manager import SessionStore, Session +from app.key_manager import KeyManager, compute_fingerprint + +tmp = Path(tempfile.mkdtemp()) + +print("=== 1. Ecriture atomique : pas de fichier .tmp residuel, contenu exact ===") +target = tmp / "fichier.txt" +atomic_write_text(target, "contenu") +assert target.read_text(encoding="utf-8") == "contenu" +assert not target.with_name("fichier.txt.tmp").exists() +print(" OK") + +print() +print("=== 2. load_json_document : valeurs par defaut si le fichier n'existe pas ===") +missing = tmp / "absent.json" +doc, warning = load_json_document(missing, default={"schema_version": 1, "items": []}, current_version=1) +assert doc == {"schema_version": 1, "items": []}, doc +assert warning is None +print(" OK") + +print() +print("=== 3. Migration : un fichier sans schema_version est promu a current_version ===") +legacy = tmp / "legacy.json" +legacy.write_text(json.dumps({"items": ["a", "b"]}), encoding="utf-8") + + +def _v0_to_v1(document: dict) -> dict: + document["items"] = [x.upper() for x in document.get("items", [])] + document["schema_version"] = 1 + return document + + +doc, warning = load_json_document( + legacy, default={"schema_version": 1, "items": []}, current_version=1, + migrations={0: _v0_to_v1}, +) +print(" document migre:", doc, "| warning:", warning) +assert doc["schema_version"] == 1 +assert doc["items"] == ["A", "B"], doc +assert warning is None +print(" OK") + +print() +print("=== 4. save_json_document : cree un .bak de la version precedente ===") +target2 = tmp / "doc.json" +save_json_document(target2, {"schema_version": 1, "n": 1}) +save_json_document(target2, {"schema_version": 1, "n": 2}) +bak = target2.with_name("doc.json.bak") +assert bak.exists() +assert json.loads(bak.read_text(encoding="utf-8"))["n"] == 1 +assert json.loads(target2.read_text(encoding="utf-8"))["n"] == 2 +print(" .bak contient la version precedente, le fichier principal la derniere : OK") + +print() +print("=== 5. Recuperation automatique si le fichier principal est corrompu ===") +target2.write_text("{ceci n'est pas du json", encoding="utf-8") +doc, warning = load_json_document(target2, default={"schema_version": 1, "n": 0}, current_version=1) +print(" document recupere:", doc, "| warning:", warning) +assert doc["n"] == 1, doc # revient a l'avant-derniere version connue (le .bak) +assert warning is not None and "bak" in warning +print(" OK") + +print() +print("=== 6. SessionStore : fichier legacy sans schema_version reste chargeable ===") +sessions_path = tmp / "sessions.json" +sessions_path.write_text( + json.dumps({"sessions": [{"id": "s1", "name": "srv1", "host": "10.0.0.1"}]}), encoding="utf-8", +) +store = SessionStore(path=sessions_path) +assert [s.name for s in store.sessions] == ["srv1"] +store.add(Session(name="srv2", host="10.0.0.2")) +on_disk = json.loads(sessions_path.read_text(encoding="utf-8")) +print(" schema_version apres sauvegarde:", on_disk.get("schema_version")) +assert on_disk.get("schema_version") == 1 +assert sessions_path.with_name("sessions.json.bak").exists() +print(" OK") + +print() +print("=== 7. KeyManager : une empreinte au mauvais format est corrigee au chargement ===") +keys_path = tmp / "keys.json" +# cle publique valide obtenue via une vraie generation, pour ne pas fabriquer un +# base64 a la main +_seed_km = KeyManager(registry_path=tmp / "seed_keys.json") +pub_line = _seed_km.generate_key("seed", save_dir=tmp / "seed").public_key_text +bogus_fp = pub_line.split()[1][:16] # l'ancien "fingerprint" errone : un bout de base64 +keys_path.write_text(json.dumps({ + "keys": [{ + "id": "k1", "name": "importee", "key_type": "ed25519", + "private_path": "", "public_path": "", + "fingerprint": bogus_fp, "public_key_text": pub_line, + }], +}), encoding="utf-8") + +km = KeyManager(registry_path=keys_path) +entry = km.keys[0] +expected = compute_fingerprint(pub_line) +print(" ancienne empreinte (tronquee):", bogus_fp) +print(" empreinte corrigee: ", entry.fingerprint) +assert entry.fingerprint == expected +assert entry.fingerprint.startswith("SHA256:") +assert entry.fingerprint != bogus_fp +print(" OK") + +print() +print("STOCKAGE ATOMIQUE / SCHEMA / FINGERPRINT : TOUS LES TESTS OK")