From f016eefce9b24e85dd4b023ed0f8886fadcf2d5b Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Thu, 3 Sep 2026 01:23:34 -0400 Subject: [PATCH] =?UTF-8?q?[ADD]=20qemu=20menu=20:=20r=C3=A9cup=C3=A9rer?= =?UTF-8?q?=20des=20fichiers=20dans=20le=20disque=20d'une=20VM?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Une VM qui ne démarre plus garde ses fichiers : libguestfs monte son qcow2 sans elle. Toute commande porte « --ro », et c'est ce qui change la manœuvre : ouvrir en écriture le disque d'une machine allumée corrompt son système de fichiers, la lire ne risque rien. Un arrêt reste proposé et non imposé — une lecture vivante voit un état peut-être incohérent, un fichier à moitié écrit, un journal non rejoué. Le nom du paquet suit la distribution, celui de Debian n'existant nulle part ailleurs. Vérifié : 16 tests, rougis par quatre mutations — perdre le « --ro », donner le nom Debian partout, dupliquer un numéro, retirer l'entrée. --- EN --- A VM that no longer boots still holds its files: libguestfs mounts its qcow2 without it. Every command carries « --ro », and that is what changes the operation: opening a running machine's disk for writing corrupts its filesystem, reading it risks nothing. A shutdown stays offered, not imposed — a live read sees a possibly torn state, a half-written file, an unreplayed journal. The package name follows the distribution, Debian's existing nowhere else. Checked: 16 tests, turned red by four mutations — losing « --ro », using Debian's name everywhere, duplicating a number, dropping the entry. Assisted-by: Claude Opus 5 --- script/todo/qemu_menu.py | 17 +- script/todo/qemu_recover.py | 318 ++++++++++++++++++++++++++++++++++++ script/todo/todo.py | 4 +- script/todo/todo_i18n.py | 109 ++++++++++++ test/test_qemu_recover.py | 248 ++++++++++++++++++++++++++++ 5 files changed, 690 insertions(+), 6 deletions(-) create mode 100644 script/todo/qemu_recover.py create mode 100644 test/test_qemu_recover.py diff --git a/script/todo/qemu_menu.py b/script/todo/qemu_menu.py index 1d488cc..80d4534 100644 --- a/script/todo/qemu_menu.py +++ b/script/todo/qemu_menu.py @@ -332,6 +332,11 @@ class QemuMenuMixin: {"prompt_description": t("Resize a VM disk")}, {"prompt_description": t("Delete VM(s)")}, {"prompt_description": t("Clean up QEMU (orphan files)")}, + { + "prompt_description": t( + "Recover files from a VM disk (libguestfs)" + ) + }, { "prompt_description": t( "Test a VM (open Odoo in a CLI browser)" @@ -387,16 +392,18 @@ class QemuMenuMixin: elif status == "10": self._qemu_cleanup() elif status == "11": - self._qemu_test_vm() + self._qemu_recover_files() elif status == "12": - self._qemu_stats() + self._qemu_test_vm() elif status == "13": - self._qemu_ssh_config_menu() + self._qemu_stats() elif status == "14": - self._qemu_tunnel_menu() + self._qemu_ssh_config_menu() elif status == "15": - self._qemu_emulator_menu() + self._qemu_tunnel_menu() elif status == "16": + self._qemu_emulator_menu() + elif status == "17": self._qemu_list_images() else: cmd_no_found = True diff --git a/script/todo/qemu_recover.py b/script/todo/qemu_recover.py new file mode 100644 index 0000000..92b52da --- /dev/null +++ b/script/todo/qemu_recover.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Menu QEMU/KVM : récupérer des fichiers dans le disque d'une VM. + +Une VM qui ne démarre plus garde ses fichiers : le qcow2 est là, et libguestfs +sait le monter SANS la machine. Le chemin complet : poser les outils selon la +distribution de l'hôte, lister les systèmes de fichiers du disque, en choisir +un, parcourir un répertoire, puis extraire vers l'hôte. + +Frontière claire : ici on LIT un disque, on n'y écrit jamais. Toute commande +porte « --ro » — c'est ce qui rend l'opération sûre même sur une VM allumée, +et c'est aussi pourquoi rien de ce fichier ne peut abîmer une machine. +""" + +import os +import shlex +import shutil +import subprocess + +from script.todo.todo_i18n import t + +# Le paquet qui apporte guestfish, par gestionnaire de paquets. Il ne vit pas +# dans la table du déploiement : ces outils ne servent QU'À la récupération, +# et les poser sur chaque hôte qui déploie une VM serait du poids pour rien. +# +# Les noms diffèrent d'une distribution à l'autre et le nom Debian ne marche +# nulle part ailleurs : « libguestfs-tools » sur apt, « guestfs-tools » sur +# dnf et zypper depuis que le paquet a été scindé, « libguestfs » sur pacman. +GUESTFS_PACKAGES = { + "apt": "libguestfs-tools", + "dnf": "guestfs-tools", + "pacman": "libguestfs", + "zypper": "guestfs-tools", +} + +# (gestionnaire, binaire à détecter, commande d'installation) +GUESTFS_INSTALL = ( + ("apt", "apt-get", "sudo apt-get install -y"), + ("dnf", "dnf", "sudo dnf install -y"), + ("pacman", "pacman", "sudo pacman -S --needed --noconfirm"), + ("zypper", "zypper", "sudo zypper --non-interactive install"), +) + +# guestfish suffit à tout faire ; les deux autres ne font que présenter mieux +# ce qu'il montre déjà. Les distinguer évite d'exiger un paquet complet là où +# l'essentiel est présent. +GUESTFS_BIN_ESSENTIEL = "guestfish" +GUESTFS_BIN_CONFORT = ("virt-filesystems", "virt-df") + + +def guestfs_install_cmd(): + """(commande d'installation, paquet) pour cet hôte, ou (None, None). + + Rend None sur un hôte dont le gestionnaire n'est pas connu — y compris + macOS, où libguestfs ne tourne pas : proposer une commande qui échouera + vaut moins que de dire qu'on ne sait pas. + """ + for cle, binaire, install in GUESTFS_INSTALL: + if shutil.which(binaire): + return f"{install} {GUESTFS_PACKAGES[cle]}", GUESTFS_PACKAGES[cle] + return None, None + + +class QemuRecoverMixin: + """Menu QEMU/KVM : récupérer des fichiers dans le disque d'une VM. + + Une VM qui ne démarre plus garde ses fichiers : le qcow2 est là, et + libguestfs sait le monter SANS la machine. Le chemin complet : poser les + outils selon la distribution de l'hôte, lister les systèmes de fichiers du + disque, en choisir un, parcourir un répertoire, puis extraire vers l'hôte. + + Frontière claire : ici on LIT un disque, on n'y écrit jamais. + """ + + def _qemu_guestfish_cmd(self, disk, *commandes): + """Commande guestfish en LECTURE SEULE sur ce disque. + + « --ro » n'est pas une précaution parmi d'autres : sans lui, ouvrir le + disque d'une VM allumée corrompt son système de fichiers. Avec lui, la + lecture est sûre à tout moment — au pire elle voit un instantané + incohérent, ce que le menu annonce. + """ + parties = " : ".join(commandes) + return ( + f"guestfish --ro -a {shlex.quote(str(disk))}" + f"{' ' + parties if parties else ''}" + ) + + def _qemu_guestfish_lines(self, disk, *commandes, timeout=180): + """Lignes rendues par guestfish, ou [] s'il échoue. + + L'appliance libguestfs démarre un noyau : compter en secondes, pas en + dixièmes. Un timeout généreux vaut mieux qu'un échec sur une machine + chargée. + """ + cmd = self._qemu_guestfish_cmd(disk, *commandes) + try: + res = subprocess.run( + cmd, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.SubprocessError): + return [] + if res.returncode != 0: + if res.stderr: + print(f" ⚠ {res.stderr.strip()[:300]}") + return [] + return [ln for ln in (res.stdout or "").splitlines() if ln.strip()] + + def _qemu_recover_ensure_tools(self): + """guestfish est-il là ? Sinon, proposer de l'installer. Rend un bool. + + L'installation est PROPOSÉE et la commande montrée avant : elle pose + un paquet de plusieurs centaines de mégaoctets, avec une image noyau. + """ + manque_essentiel = shutil.which(GUESTFS_BIN_ESSENTIEL) is None + confort = [b for b in GUESTFS_BIN_CONFORT if shutil.which(b) is None] + if not manque_essentiel and not confort: + return True + install, paquet = guestfs_install_cmd() + if manque_essentiel: + print(f"\n⚠ {t('guestfish is missing: libguestfs is not here.')}") + else: + print(f"\n⚠ {t('Some libguestfs helpers are missing:')}") + print(f" {' '.join(confort)}") + if not install: + print(f" {t('Unknown package manager: install libguestfs.')}") + return not manque_essentiel + print(f" {t('Will execute:')} {install}") + if not self._is_yes_default_yes( + input(t("Install libguestfs now? (Y/n): ")) + ): + return not manque_essentiel + self.execute.exec_command_live(install, source_erplibre=False) + return shutil.which(GUESTFS_BIN_ESSENTIEL) is not None + + def _qemu_recover_ready(self, name): + """La VM est-elle dans un état où la lecture a du sens ? Rend un bool. + + Une VM ALLUMÉE écrit pendant qu'on lit : la copie voit un instantané + qui peut être incohérent — un fichier à moitié écrit, un journal non + rejoué. Ce n'est pas dangereux en « --ro », mais il faut le dire, et + proposer l'arrêt propre qui rend la lecture fidèle. + """ + etat = self._qemu_domstate(name) + if etat != "running": + return True + print(f"\n⚠ {t('This VM is running.')}") + print(f" {t('Reading a live disk sees a possibly torn state:')}") + print(f" {t('a half-written file, an unreplayed journal.')}") + choix = self._qemu_pick( + t("What do you want to do?"), + ["read", "shutdown", "cancel"], + "shutdown", + [ + t("Read anyway (read-only, no risk for the VM)"), + t("Shut the VM down cleanly, then read"), + t("Cancel"), + ], + ) + if choix == "cancel": + return False + if choix == "read": + return True + return self._qemu_shutdown_wait(name) + + def _qemu_recover_pick_filesystem(self, disk): + """Système de fichiers choisi dans le disque, ou "" si aucun. + + « list-filesystems » les rend tous, y compris ceux qu'on ne peut pas + monter — swap, partitions vides. Les montrer quand même : leur absence + de la liste inquiéterait plus qu'elle n'aiderait. + """ + cmd = self._qemu_guestfish_cmd(disk, "run", "list-filesystems") + print(f"\n{t('Will execute:')} {cmd}") + lignes = self._qemu_guestfish_lines(disk, "run", "list-filesystems") + if not lignes: + print(f" {t('No filesystem found on this disk.')}") + return "" + parts, labels = [], [] + for ligne in lignes: + # « /dev/sda3: ext4 » — le nom, puis son type. + dev, _, typ = ligne.partition(":") + dev, typ = dev.strip(), typ.strip() + if not dev: + continue + parts.append(dev) + labels.append(f"{dev} {typ}") + if not parts: + return "" + # Le plus grand système de fichiers non-swap est presque toujours la + # racine : le proposer par défaut épargne un choix à qui ne connaît + # pas le partitionnement de sa VM. + defaut = next( + (p for p, lab in zip(parts, labels) if "swap" not in lab), + parts[0], + ) + return self._qemu_pick(t("Filesystem to mount"), parts, defaut, labels) + + def _qemu_recover_browse(self, disk, part, chemin): + """Liste un répertoire du système de fichiers monté. + + Rend les entrées trouvées.""" + cmd = self._qemu_guestfish_cmd( + disk, "run", f"mount {part} /", f"ls {shlex.quote(chemin)}" + ) + print(f"\n{t('Will execute:')} {cmd}") + entrees = self._qemu_guestfish_lines( + disk, "run", f"mount {part} /", f"ls {shlex.quote(chemin)}" + ) + if not entrees: + print(f" {t('Empty or unreadable directory.')}") + return [] + for entree in entrees: + print(f" {entree}") + print(f" {t('Total:')} {len(entrees)}") + return entrees + + def _qemu_recover_copy_out(self, disk, part, source, dest): + """Extrait un chemin de la VM vers l'hôte. Rend un bool. + + « copy-out » écrit DANS le répertoire de destination : il faut donc + qu'il existe, sinon guestfish s'arrête sur une erreur qui ne dit pas + laquelle des deux extrémités manque. + """ + try: + os.makedirs(dest, exist_ok=True) + except OSError as exc: + print(f" ⚠ {t('Cannot create the destination: ')}{exc}") + return False + cmd = self._qemu_guestfish_cmd( + disk, + "run", + f"mount {part} /", + f"copy-out {shlex.quote(source)} {shlex.quote(dest)}", + ) + print(f"\n{t('Will execute:')} {cmd}") + code = self.execute.exec_command_live(cmd, source_erplibre=False) + if code: + print(f" ⚠ {t('Extraction failed.')}") + return False + cible = os.path.join(dest, os.path.basename(source.rstrip("/"))) + print(f"\n✅ {t('Extracted to: ')}{cible}") + return True + + def _qemu_recover_diagnostics(self, disk): + """Commandes de diagnostic sur le disque, sans rien monter à la main. + + Chacune répond à une question précise qu'on se pose quand la lecture + ne donne pas ce qu'on attend : les partitions existent-elles, restent- + elles de la place, le système est-il reconnu, l'appliance + démarre-t-elle. + """ + q = shlex.quote(str(disk)) + sondes = [ + ( + t("Partitions and sizes"), + f"virt-filesystems -a {q} --long -h --all", + ), + (t("Free space per filesystem"), f"virt-df -a {q} -h"), + ( + t("Detected operating system"), + self._qemu_guestfish_cmd(disk, "run", "inspect-os"), + ), + ( + t("Does the libguestfs appliance boot?"), + "libguestfs-test-tool", + ), + ] + for titre, cmd in sondes: + print(f"\n── {titre} ──") + print(f"{t('Will execute:')} {cmd}") + self.execute.exec_command_live(cmd, source_erplibre=False) + + def _qemu_recover_files(self): + """Récupère des fichiers dans le disque d'une VM, sans la démarrer.""" + print(f"\n💾 {t('Recover files from a VM disk (libguestfs)')}") + if not self._qemu_recover_ensure_tools(): + return + self._qemu_list_vms() + print() + name = input(t("VM name or ID: ")).strip() + if not name: + print(t("VM name is required!")) + return + name = self._qemu_domname(name) + disk = self._qemu_main_disk(name) + if not disk: + print(f" ⚠ {t('No disk found for this VM.')}") + return + print(f" {t('Disk:')} {disk}") + if not self._qemu_recover_ready(name): + print(t("Cancelled.")) + return + if self._is_yes(input(t("Run the diagnostics first? (y/N): "))): + self._qemu_recover_diagnostics(disk) + part = self._qemu_recover_pick_filesystem(disk) + if not part: + return + chemin = input(t("Directory to list [/]: ")).strip() or "/" + while True: + self._qemu_recover_browse(disk, part, chemin) + suite = input( + t("Another directory, or Enter to extract: ") + ).strip() + if not suite: + break + chemin = suite + source = input(f"{t('Path to extract')} [{chemin}] : ").strip() + source = source or chemin + defaut_dest = f"/tmp/{name}-backup" + dest = input(f"{t('Destination on the host')} [{defaut_dest}] : ") + dest = dest.strip() or defaut_dest + self._qemu_recover_copy_out(disk, part, source, dest) diff --git a/script/todo/todo.py b/script/todo/todo.py index 8b5f2cf..580815a 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -28,6 +28,7 @@ from script.config import config_file from script.execute import execute from script.todo import todo_install, todo_prefs from script.todo.database_manager import DatabaseManager +from script.todo.kdbx_manager import KdbxManager from script.todo.longtest_menu import LongTestMenuMixin from script.todo.proxmox_menu import ProxmoxMenuMixin from script.todo.qemu_access import QemuAccessMixin @@ -35,7 +36,7 @@ from script.todo.qemu_deploy import QemuDeployMixin from script.todo.qemu_install import QemuInstallMixin from script.todo.qemu_manage import QemuManageMixin from script.todo.qemu_menu import QemuMenuMixin -from script.todo.kdbx_manager import KdbxManager +from script.todo.qemu_recover import QemuRecoverMixin from script.todo.todo_i18n import get_lang, lang_is_configured, set_lang, t from script.todo.version_manager import get_odoo_version @@ -95,6 +96,7 @@ class TODO( QemuDeployMixin, QemuInstallMixin, QemuManageMixin, + QemuRecoverMixin, QemuAccessMixin, ProxmoxMenuMixin, LongTestMenuMixin, diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 020cb17..8cc8d57 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -5004,6 +5004,115 @@ TRANSLATIONS = { " (y/N, default: no): " ), }, + # Récupération de fichiers dans le disque d'une VM (libguestfs) + "Recover files from a VM disk (libguestfs)": { + "fr": "💾 Récupérer des fichiers dans le disque d'une VM (libguestfs)", + "en": "Recover files from a VM disk (libguestfs)", + }, + "guestfish is missing: libguestfs is not here.": { + "fr": "guestfish est absent : libguestfs n'est pas installé ici.", + "en": "guestfish is missing: libguestfs is not here.", + }, + "Some libguestfs helpers are missing:": { + "fr": "Des utilitaires libguestfs manquent :", + "en": "Some libguestfs helpers are missing:", + }, + "Unknown package manager: install libguestfs.": { + "fr": "Gestionnaire de paquets inconnu : installez libguestfs.", + "en": "Unknown package manager: install libguestfs.", + }, + "Install libguestfs now? (Y/n): ": { + "fr": "Installer libguestfs maintenant ? (O/n) : ", + "en": "Install libguestfs now? (Y/n, default: yes): ", + }, + "This VM is running.": { + "fr": "Cette VM est allumée.", + "en": "This VM is running.", + }, + "Reading a live disk sees a possibly torn state:": { + "fr": "Lire un disque vivant montre un état peut-être incohérent :", + "en": "Reading a live disk sees a possibly torn state:", + }, + "a half-written file, an unreplayed journal.": { + "fr": "un fichier à moitié écrit, un journal non rejoué.", + "en": "a half-written file, an unreplayed journal.", + }, + "Read anyway (read-only, no risk for the VM)": { + "fr": "Lire quand même (lecture seule, sans risque pour la VM)", + "en": "Read anyway (read-only, no risk for the VM)", + }, + "Shut the VM down cleanly, then read": { + "fr": "Éteindre proprement la VM, puis lire", + "en": "Shut the VM down cleanly, then read", + }, + "No filesystem found on this disk.": { + "fr": "Aucun système de fichiers trouvé sur ce disque.", + "en": "No filesystem found on this disk.", + }, + "Filesystem to mount": { + "fr": "Système de fichiers à monter", + "en": "Filesystem to mount", + }, + "Empty or unreadable directory.": { + "fr": "Répertoire vide ou illisible.", + "en": "Empty or unreadable directory.", + }, + "Cannot create the destination: ": { + "fr": "Impossible de créer la destination : ", + "en": "Cannot create the destination: ", + }, + "Extraction failed.": { + "fr": "L'extraction a échoué.", + "en": "Extraction failed.", + }, + "Extracted to: ": { + "fr": "Extrait vers : ", + "en": "Extracted to: ", + }, + "Partitions and sizes": { + "fr": "Partitions et tailles", + "en": "Partitions and sizes", + }, + "Free space per filesystem": { + "fr": "Espace libre par système de fichiers", + "en": "Free space per filesystem", + }, + "Detected operating system": { + "fr": "Système d'exploitation détecté", + "en": "Detected operating system", + }, + "Does the libguestfs appliance boot?": { + "fr": "L'appliance libguestfs démarre-t-elle ?", + "en": "Does the libguestfs appliance boot?", + }, + "No disk found for this VM.": { + "fr": "Aucun disque trouvé pour cette VM.", + "en": "No disk found for this VM.", + }, + "Disk:": { + "fr": "Disque :", + "en": "Disk:", + }, + "Run the diagnostics first? (y/N): ": { + "fr": "Lancer d'abord les diagnostics ? (o/N) : ", + "en": "Run the diagnostics first? (y/N, default: no): ", + }, + "Directory to list [/]: ": { + "fr": "Répertoire à lister [/] : ", + "en": "Directory to list [/]: ", + }, + "Another directory, or Enter to extract: ": { + "fr": "Autre répertoire, ou Entrée pour extraire : ", + "en": "Another directory, or Enter to extract: ", + }, + "Path to extract": { + "fr": "Chemin à extraire", + "en": "Path to extract", + }, + "Destination on the host": { + "fr": "Destination sur l'hôte", + "en": "Destination on the host", + }, "Full output:": { "fr": "Sortie complète :", "en": "Full output:", diff --git a/test/test_qemu_recover.py b/test/test_qemu_recover.py new file mode 100644 index 0000000..7333da2 --- /dev/null +++ b/test/test_qemu_recover.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Récupérer des fichiers dans le disque d'une VM, sans la démarrer. + +Une VM qui ne démarre plus garde ses fichiers, et libguestfs sait monter son +qcow2 sans elle. Le risque n'est pas la lecture : c'est d'ouvrir en ÉCRITURE +le disque d'une machine allumée, ce qui corrompt son système de fichiers. + +Ce que ces tests gardent : + +- « --ro » sur CHAQUE commande, sans exception : c'est lui qui rend l'opération + sûre sur une VM en marche ; +- le nom du paquet suit la distribution — celui de Debian n'existe nulle part + ailleurs ; +- les chemins saisis par l'utilisateur sont échappés avant d'atteindre le + shell. +""" + +import shlex +import sys +import unittest +from unittest import mock + +sys.argv = ["todo.py"] +from script.todo import qemu_recover as qr # noqa: E402 +from script.todo.todo import TODO # noqa: E402 + +DISQUE = "/var/lib/libvirt/images/vm-a.qcow2" + + +class LesOutils(unittest.TestCase): + def test_each_package_manager_gets_its_own_name(self): + """« libguestfs-tools » est le nom Debian : il n'existe ni sur Arch, + ni sur Fedora, ni sur openSUSE.""" + attendu = { + "apt-get": "libguestfs-tools", + "dnf": "guestfs-tools", + "pacman": "libguestfs", + "zypper": "guestfs-tools", + } + for binaire, paquet in attendu.items(): + with self.subTest(binaire=binaire): + with mock.patch.object( + qr.shutil, + "which", + side_effect=lambda b, cible=binaire: ( + "/usr/bin/x" if b == cible else None + ), + ): + cmd, nom = qr.guestfs_install_cmd() + self.assertEqual(paquet, nom) + self.assertIn(paquet, cmd) + + def test_an_unknown_host_proposes_nothing(self): + """Proposer une commande qui échouera vaut moins que de dire qu'on ne + sait pas — sur macOS, libguestfs ne tourne pas.""" + with mock.patch.object(qr.shutil, "which", return_value=None): + self.assertEqual((None, None), qr.guestfs_install_cmd()) + + +class LaLectureEstToujoursSeule(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + + def test_every_command_carries_read_only(self): + """Sans « --ro », ouvrir le disque d'une VM allumée corrompt son + système de fichiers. Aucune commande n'a de raison de s'en passer.""" + cmds = [ + self.todo._qemu_guestfish_cmd(DISQUE), + self.todo._qemu_guestfish_cmd(DISQUE, "run", "list-filesystems"), + self.todo._qemu_guestfish_cmd( + DISQUE, "run", "mount /dev/sda3 /", "ls /home" + ), + ] + for cmd in cmds: + with self.subTest(cmd=cmd): + self.assertIn("--ro", cmd) + self.assertIn(shlex.quote(DISQUE), cmd) + + def test_the_commands_are_chained_the_guestfish_way(self): + cmd = self.todo._qemu_guestfish_cmd( + DISQUE, "run", "mount /dev/sda3 /", "ls /home" + ) + self.assertIn("run : mount /dev/sda3 / : ls /home", cmd) + + def test_a_hostile_path_cannot_escape_the_shell(self): + """Le chemin vient de l'utilisateur : sans échappement, « ; » ouvre + une seconde commande.""" + cmd = self.todo._qemu_guestfish_cmd("/tmp/x.qcow2; rm -rf /", "run") + # Le chemin entier tient dans UNE apostrophe : le « ; » y est du + # texte, pas un séparateur de commande. + self.assertIn("'/tmp/x.qcow2; rm -rf /'", cmd) + self.assertNotIn( + "qcow2; rm", cmd.replace("'/tmp/x.qcow2; rm -rf /'", "") + ) + + +class LaVmAllumee(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + + def test_a_stopped_vm_goes_straight_through(self): + self.todo._qemu_domstate = lambda n: "shut off" + self.assertTrue(self.todo._qemu_recover_ready("vm-a")) + + def test_a_running_vm_offers_three_ways(self): + self.todo._qemu_domstate = lambda n: "running" + vus = {} + + def faux_pick(titre, valeurs, defaut, labels=None): + vus["valeurs"] = valeurs + vus["defaut"] = defaut + return "read" + + self.todo._qemu_pick = faux_pick + with mock.patch("builtins.print"): + self.assertTrue(self.todo._qemu_recover_ready("vm-a")) + self.assertEqual(["read", "shutdown", "cancel"], vus["valeurs"]) + # L'arrêt propre est le défaut : c'est la seule voie qui donne une + # copie fidèle. + self.assertEqual("shutdown", vus["defaut"]) + + def test_cancelling_stops_everything(self): + self.todo._qemu_domstate = lambda n: "running" + self.todo._qemu_pick = lambda *a, **k: "cancel" + with mock.patch("builtins.print"): + self.assertFalse(self.todo._qemu_recover_ready("vm-a")) + + def test_choosing_shutdown_waits_for_it(self): + self.todo._qemu_domstate = lambda n: "running" + self.todo._qemu_pick = lambda *a, **k: "shutdown" + appels = [] + self.todo._qemu_shutdown_wait = lambda n: appels.append(n) or True + with mock.patch("builtins.print"): + self.assertTrue(self.todo._qemu_recover_ready("vm-a")) + self.assertEqual(["vm-a"], appels) + + +class LExtraction(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + self.lances = [] + self.todo.execute = mock.MagicMock() + self.todo.execute.exec_command_live.side_effect = ( + lambda cmd, **k: self.lances.append(cmd) or 0 + ) + + def test_copy_out_names_both_ends(self): + with mock.patch.object(qr.os, "makedirs"), mock.patch( + "builtins.print" + ): + ok = self.todo._qemu_recover_copy_out( + DISQUE, "/dev/sda3", "/home/erplibre", "/tmp/vm-a-backup" + ) + self.assertTrue(ok) + cmd = self.lances[0] + self.assertIn("copy-out /home/erplibre /tmp/vm-a-backup", cmd) + self.assertIn("mount /dev/sda3 /", cmd) + self.assertIn("--ro", cmd) + + def test_an_uncreatable_destination_stops_before_running(self): + """guestfish s'arrêterait sur une erreur qui ne dit pas laquelle des + deux extrémités manque.""" + with mock.patch.object( + qr.os, "makedirs", side_effect=OSError("lecture seule") + ), mock.patch("builtins.print"): + ok = self.todo._qemu_recover_copy_out( + DISQUE, "/dev/sda3", "/home", "/interdit" + ) + self.assertFalse(ok) + self.assertEqual([], self.lances) + + +class LesDiagnostics(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + self.lances = [] + self.todo.execute = mock.MagicMock() + self.todo.execute.exec_command_live.side_effect = ( + lambda cmd, **k: self.lances.append(cmd) or 0 + ) + + def test_the_four_probes_answer_four_questions(self): + with mock.patch("builtins.print"): + self.todo._qemu_recover_diagnostics(DISQUE) + joint = " ".join(self.lances) + for attendu in ( + "virt-filesystems", + "virt-df", + "inspect-os", + "libguestfs-test-tool", + ): + self.assertIn(attendu, joint) + + def test_no_diagnostic_writes_to_the_disk(self): + with mock.patch("builtins.print"): + self.todo._qemu_recover_diagnostics(DISQUE) + for cmd in self.lances: + if "guestfish" in cmd: + self.assertIn("--ro", cmd) + + +class LeMenu(unittest.TestCase): + def test_the_entry_is_wired_in_the_manage_section(self): + from pathlib import Path + + source = Path("script/todo/qemu_menu.py").read_text(encoding="utf-8") + self.assertIn("Recover files from a VM disk (libguestfs)", source) + self.assertIn("self._qemu_recover_files()", source) + + def test_the_config_entry_still_reaches_its_command(self): + """L'entrée du catalogue vient de todo.json : elle n'a pas de branche + « elif » et dépend du repli par indice. Insérer une entrée codée en + dur la décale, et un décalage manqué lancerait la mauvaise.""" + from unittest import mock as m + + from script.todo.todo_i18n import set_lang + + set_lang("fr") + todo = TODO() + todo._menu_header = lambda: "x" + lancees = [] + todo.execute_from_configuration = lambda e: lancees.append(e) + with m.patch("script.todo.qemu_menu.click") as click, m.patch.object( + todo, "_qemu_ensure_tools", return_value=True + ), m.patch("builtins.print"): + click.prompt.side_effect = ["18", "0"] + todo.prompt_execute_qemu() + self.assertEqual(1, len(lancees), lancees) + self.assertIn("dry-run", lancees[0].get("bash_command", "")) + + def test_the_branches_stay_in_order_and_unique(self): + """Insérer une entrée décale tout ce qui suit : un numéro en double + rendrait une commande inatteignable.""" + import re + from pathlib import Path + + source = Path("script/todo/qemu_menu.py").read_text(encoding="utf-8") + debut = source.index("def prompt_execute_qemu") + corps = source[debut : source.index("\n def ", debut + 10)] + nums = [int(n) for n in re.findall(r'status == "(\d+)"', corps)] + self.assertEqual(sorted(nums), nums, nums) + self.assertEqual(len(set(nums)), len(nums), nums) + + +if __name__ == "__main__": + unittest.main()