diff --git a/script/todo/qemu_access.py b/script/todo/qemu_access.py index e0782f9..08f21d4 100644 --- a/script/todo/qemu_access.py +++ b/script/todo/qemu_access.py @@ -11,6 +11,7 @@ import subprocess import time from script.todo import todo_install +from script.todo.qemu_privilege import sudo_prefix from script.todo.todo_i18n import t @@ -730,7 +731,10 @@ class QemuAccessMixin: port = self._qemu_vnc_port(domain, jump) # Les commandes de réparation se lancent SUR l'hyperviseur : le préfixe # évite de les copier sur la mauvaise machine, l'erreur naturelle ici. + # Sur l'hyperviseur DISTANT, on ne peut pas sonder ses droits d'ici : + # « sudo » y reste écrit. En local, le sondage tranche. pre = f"ssh {jump} " if jump else "" + su = "sudo " if jump else sudo_prefix() if not port and self._hypervisor_is_proxmox(jump): self._pve_console_hint(jump, domain) return @@ -738,17 +742,17 @@ class QemuAccessMixin: print(f"\n ⚠ {t('This VM exposes no VNC port.')}") print(f" {t('Its display is likely spice with listen=none:')}") print( - f" {pre}sudo virsh dumpxml {domain} | grep -A2 '" ) - print(f" {pre}sudo virsh start {domain}") + print(f" {pre}{su}virsh start {domain}") print(f"\n {t('New VMs get this by default; see deploy_qemu.')}") return if jump: diff --git a/script/todo/qemu_deploy.py b/script/todo/qemu_deploy.py index 6a86938..b6804ec 100644 --- a/script/todo/qemu_deploy.py +++ b/script/todo/qemu_deploy.py @@ -13,6 +13,7 @@ import subprocess import time from script.todo import todo_prefs +from script.todo.qemu_privilege import sudo_prefix from script.todo.todo_i18n import get_lang, t @@ -2121,4 +2122,4 @@ class QemuDeployMixin: print(f"{'═' * 60}") print(f"\n✅ {t('ERPLibre infra deployment done.')}") print(f" {t('Default login:')} erplibre / erplibre") - print(f" {t('Manage with:')} sudo virsh list --all") + print(f" {t('Manage with:')} {sudo_prefix()}virsh list --all") diff --git a/script/todo/qemu_install_monitor.py b/script/todo/qemu_install_monitor.py index 2261f10..d5eb71b 100644 --- a/script/todo/qemu_install_monitor.py +++ b/script/todo/qemu_install_monitor.py @@ -22,6 +22,8 @@ import shutil import socket import subprocess import time + +from script.todo.qemu_privilege import sudo_prefix from pathlib import Path try: @@ -1974,15 +1976,16 @@ def delete_vm_cmd(name: str, with_disks: bool, uuid: str = "") -> str: cmd = "" if uuid: cmd = ( - f"vu=$(sudo virsh domuuid {q} 2>/dev/null | tr -d '[:space:]'); " + f"vu=$({sudo_prefix()}virsh domuuid {q} 2>/dev/null" + " | tr -d '[:space:]'); " f'if [ "$vu" != {shlex.quote(uuid)} ]; then ' f'echo "REFUS : {name} n\'est plus le même domaine"' f' "($vu). Rien n\'a été effacé."; exit 1; fi; ' ) cmd += ( - f"sudo virsh destroy {q} 2>/dev/null; " - f"sudo virsh undefine {q} --nvram 2>/dev/null " - f"|| sudo virsh undefine {q}" + f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " + f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " + f"|| {sudo_prefix()}virsh undefine {q}" ) if with_disks: disk = shlex.quote(f"/var/lib/libvirt/images/{name}.qcow2") @@ -2917,7 +2920,7 @@ def run_monitor(manifest_path: str, run_app: bool = True): ) sortie = "Ctrl+O" else: - cmd = f"sudo virsh console {shlex.quote(vm['name'])}" + cmd = f"{sudo_prefix()}virsh console {shlex.quote(vm['name'])}" titre = f"virsh console {vm['name']}" sortie = "Ctrl+]" with self.suspend(): diff --git a/script/todo/qemu_manage.py b/script/todo/qemu_manage.py index 53d95a7..de55f10 100644 --- a/script/todo/qemu_manage.py +++ b/script/todo/qemu_manage.py @@ -13,6 +13,7 @@ import subprocess import time from script.todo import todo_install +from script.todo.qemu_privilege import sudo_prefix from script.todo.todo_i18n import t @@ -127,7 +128,7 @@ class QemuManageMixin: self.execute.exec_command_live(cmd, source_erplibre=False) def _qemu_list_vms(self, ask_advanced=False): - cmd = "sudo virsh list --all" + cmd = f"{sudo_prefix()}virsh list --all" print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) if not ask_advanced: @@ -211,7 +212,7 @@ class QemuManageMixin: print(t("Cancelled.")) return for real in resolved: - cmd = f"sudo virsh {action} {shlex.quote(real)}" + cmd = f"{sudo_prefix()}virsh {action} {shlex.quote(real)}" print(f"\n{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -391,7 +392,9 @@ class QemuManageMixin: print(t("Cancelled.")) return for entry in cmds: - cmd = "sudo " + " ".join(shlex.quote(c) for c in entry["cmd"]) + cmd = sudo_prefix() + " ".join( + shlex.quote(c) for c in entry["cmd"] + ) print(f"\n{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -674,7 +677,10 @@ class QemuManageMixin: else: targets = [name] for tgt in targets: - cmd = f"sudo virsh domifaddr {shlex.quote(tgt)} --source lease" + cmd = ( + f"{sudo_prefix()}virsh domifaddr {shlex.quote(tgt)}" + " --source lease" + ) print(f"\n{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -691,7 +697,7 @@ class QemuManageMixin: print( f"👤 {t('Default login (if set at deploy): erplibre / erplibre')}" ) - cmd = f"sudo virsh console {shlex.quote(name)}" + cmd = f"{sudo_prefix()}virsh console {shlex.quote(name)}" print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -1035,7 +1041,10 @@ class QemuManageMixin: return True # --mode acpi,agent : envoie le SIGNAL d'extinction (bouton ACPI) puis # tente l'agent invité si présent — plus fiable qu'un arrêt brutal. - cmd = f"sudo virsh shutdown {shlex.quote(name)} --mode acpi,agent" + cmd = ( + f"{sudo_prefix()}virsh shutdown {shlex.quote(name)}" + " --mode acpi,agent" + ) print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) print( @@ -1066,7 +1075,7 @@ class QemuManageMixin: ) ) ): - cmd = f"sudo virsh destroy {shlex.quote(name)}" + cmd = f"{sudo_prefix()}virsh destroy {shlex.quote(name)}" print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) time.sleep(2) @@ -1142,7 +1151,8 @@ class QemuManageMixin: print(f"\n{t('Current disk:')} {disk}") # -U : lecture sûre même VM allumée (sinon « shared write lock »). self.execute.exec_command_live( - f"sudo qemu-img info -U {shlex.quote(disk)}", source_erplibre=False + f"{sudo_prefix()}qemu-img info -U {shlex.quote(disk)}", + source_erplibre=False, ) cur_bytes = self._qemu_disk_virtual_bytes(disk) cur_gb = cur_bytes / (1 << 30) @@ -1259,11 +1269,14 @@ class QemuManageMixin: # Agrandissement À CHAUD : le disque virtuel grossit, le FS invité # devra être étendu ensuite. cmd = ( - f"sudo virsh blockresize {shlex.quote(name)} " + f"{sudo_prefix()}virsh blockresize {shlex.quote(name)} " f"{shlex.quote(disk)} {new_gb:g}G" ) else: - cmd = f"sudo qemu-img resize {shlex.quote(disk)} {new_gb:g}G" + cmd = ( + f"{sudo_prefix()}qemu-img resize" + f" {shlex.quote(disk)} {new_gb:g}G" + ) # 4) Agrandissement : exécuter la commande + proposer d'étendre le FS. if cmd is not None: @@ -1302,7 +1315,7 @@ class QemuManageMixin: if self._is_yes(input(t("Start the VM now? (y/N): "))): # `name` est déjà le nom canonique : « virsh start » # échouerait car l'ID disparaît quand la VM est éteinte. - cmd = f"sudo virsh start {shlex.quote(name)}" + cmd = f"{sudo_prefix()}virsh start {shlex.quote(name)}" print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -1862,7 +1875,7 @@ class QemuManageMixin: ) if not self._is_yes(input(t("Open the serial console now? (y/N): "))): return - cmd = f"sudo virsh console {shlex.quote(name)}" + cmd = f"{sudo_prefix()}virsh console {shlex.quote(name)}" print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) @@ -1923,9 +1936,9 @@ class QemuManageMixin: # Éteindre si en cours, puis retirer la définition (+ nvram si # UEFI ; repli sans l'option pour les vieilles versions de virsh). cmd = ( - f"sudo virsh destroy {q} 2>/dev/null; " - f"sudo virsh undefine {q} --nvram 2>/dev/null " - f"|| sudo virsh undefine {q}" + f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " + f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " + f"|| {sudo_prefix()}virsh undefine {q}" ) if del_disks and fichiers: cmd += "; sudo rm -f " + " ".join( @@ -2148,9 +2161,9 @@ class QemuManageMixin: for name in ghosts: q = shlex.quote(name) cmd = ( - f"sudo virsh destroy {q} 2>/dev/null; " - f"sudo virsh undefine {q} --nvram 2>/dev/null " - f"|| sudo virsh undefine {q}" + f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " + f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " + f"|| {sudo_prefix()}virsh undefine {q}" ) print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) diff --git a/script/todo/qemu_menu.py b/script/todo/qemu_menu.py index a09e04f..1d488cc 100644 --- a/script/todo/qemu_menu.py +++ b/script/todo/qemu_menu.py @@ -239,6 +239,38 @@ class QemuMenuMixin: print(f"{t('Will execute:')} {cmd}") self.execute.exec_command_live(cmd, source_erplibre=False) + def _qemu_warn_libvirt_access(self): + """Dit, AVANT d'installer, par quelle voie les commandes passeront. + + Deux voies mènent à qemu:///system : le groupe libvirt, qui ne demande + rien, et sudo, qui demande un mot de passe à chaque commande. Le suivi + d'installation, lui, tourne détaché et sans terminal : il ne peut + répondre à aucune invite. Le savoir avant l'installation vaut mieux que + de le découvrir devant la première invite de mot de passe. + + Ne dit rien quand l'accès est déjà là — un avertissement qui se répète + sans objet finit par ne plus se lire. + """ + from script.todo import qemu_privilege as qp + + if qp.libvirt_reachable(force=True): + return + declare, actif = qp.group_state() + if actif: + return + print(f"\n⚠ {t('Recommended before installing:')}") + if declare: + # Le groupe est acquis mais la session est plus ancienne que lui : + # aucun usermod à refaire, seulement une session à rouvrir. + print(f" {t('You are in the libvirt group, but this session')}") + print(f" {t('predates it. Log out and back in, or run:')}") + print(" newgrp libvirt") + else: + print(f" {t('Grant libvirt access, or every VM command will')}") + print(f" {t('ask for a sudo password:')}") + print(" sudo usermod -aG libvirt $USER") + print(f" {t('then log out and back in.')}") + def _qemu_ensure_tools(self): """virsh absent : proposer l'installation plutôt que de laisser chaque commande échouer sur « sudo: virsh: command not found ». @@ -249,6 +281,7 @@ class QemuMenuMixin: return True print(f"\n⚠ {t('virsh is missing: libvirt is not installed here.')}") print(f" {t('Every VM command will fail until it is.')}") + self._qemu_warn_libvirt_access() if not self._is_yes_default_yes( input(t("Install the QEMU/libvirt tools now? (Y/n): ")) ): diff --git a/script/todo/qemu_privilege.py b/script/todo/qemu_privilege.py new file mode 100644 index 0000000..e5a33fe --- /dev/null +++ b/script/todo/qemu_privilege.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Faut-il « sudo » pour parler à libvirt ? Une seule réponse pour tout TODO. + +Appartenir au groupe libvirt suffit à joindre qemu:///system : préfixer alors +les commandes de « sudo » ne donne aucun droit de plus et réclame un mot de +passe pour rien. La question se tranche en ESSAYANT, jamais en lisant +/etc/group : les groupes d'un processus sont figés à l'ouverture de session, +donc un utilisateur fraîchement ajouté y figure sans que le shell courant en +dispose. L'essai dit ce que le shell peut FAIRE, la table dit ce qui a été +DÉCLARÉ — et c'est l'écart entre les deux qui explique « je suis pourtant +dans le groupe ». +""" + +import grp +import os +import shutil +import subprocess + +# Réponse du sondage, gardée pour la session : chaque commande du menu la +# demande, et lancer un virsh par entrée de menu se verrait. +_CACHE: bool | None = None + +PROBE = ["virsh", "--connect", "qemu:///system", "list", "--name"] + + +def reset_cache() -> None: + """Oublie le sondage. À appeler après un changement de droits.""" + global _CACHE + _CACHE = None + + +def libvirt_reachable(force: bool = False) -> bool: + """qemu:///system répond-il SANS sudo ? + + Rend faux quand virsh est absent : il n'y a alors rien à joindre, et le + dire évite de conclure « il faut sudo » sur une machine sans libvirt. + """ + global _CACHE + if _CACHE is not None and not force: + return _CACHE + if shutil.which("virsh") is None: + _CACHE = False + return _CACHE + try: + probe = subprocess.run( + PROBE, capture_output=True, text=True, timeout=15 + ) + _CACHE = probe.returncode == 0 + except (OSError, subprocess.SubprocessError): + _CACHE = False + return _CACHE + + +def needs_sudo() -> bool: + """Faut-il préfixer les commandes libvirt de « sudo » ? + + Root n'en a jamais besoin. Sans virsh, il n'y a rien à préfixer : rendre + faux laisse la commande échouer sur « command not found » plutôt que sur + une invite de mot de passe qui ne mène nulle part. + """ + if os.geteuid() == 0: + return False + if shutil.which("virsh") is None: + return False + return not libvirt_reachable() + + +def sudo_prefix() -> str: + """« sudo » et son espace, ou la chaîne vide. À coller devant virsh.""" + return "sudo " if needs_sudo() else "" + + +def group_state(user: str = "") -> tuple[bool, bool]: + """(déclaré, actif) pour le groupe libvirt. + + « déclaré » : le nom figure dans le groupe, d'après la base système. + « actif » : le processus courant PORTE le groupe. Les deux diffèrent tant + que la session n'a pas été rouverte, et c'est le cas qui déroute le plus. + Les deux valent faux quand le groupe n'existe pas — libvirt pas encore + installé. + """ + user = user or _current_user() + try: + entry = grp.getgrnam("libvirt") + except KeyError: + return False, False + declared = user in entry.gr_mem + try: + declared = declared or grp.getgrgid(os.getgid()).gr_name == "libvirt" + except (KeyError, OSError): + pass + return declared, entry.gr_gid in os.getgroups() + + +def _current_user() -> str: + """Le nom de l'utilisateur courant, sans lever si l'environnement ment.""" + try: + import getpass + + return getpass.getuser() + except Exception: + return os.environ.get("USER") or "" diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index c5cb581..1a93945 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -4965,6 +4965,22 @@ TRANSLATIONS = { "fr": "La 3D était active ; réessayer sans elle :", "en": "3D was on; retry without it:", }, + "Recommended before installing:": { + "fr": "Recommandé avant d'installer :", + "en": "Recommended before installing:", + }, + "Grant libvirt access, or every VM command will": { + "fr": "Donner l'accès à libvirt, sinon chaque commande VM", + "en": "Grant libvirt access, or every VM command will", + }, + "ask for a sudo password:": { + "fr": "demandera un mot de passe sudo :", + "en": "ask for a sudo password:", + }, + "then log out and back in.": { + "fr": "puis se déconnecter et se reconnecter.", + "en": "then log out and back in.", + }, "Full output:": { "fr": "Sortie complète :", "en": "Full output:", diff --git a/test/test_qemu_gpu.py b/test/test_qemu_gpu.py index 071fddc..f20ba4a 100644 --- a/test/test_qemu_gpu.py +++ b/test/test_qemu_gpu.py @@ -446,7 +446,31 @@ class TestMenuGlue(unittest.TestCase): joined = " ".join(todo.launched) self.assertIn("--vcpus 4", joined) self.assertIn("accel3d=on", joined) - self.assertTrue(all(c.startswith("sudo ") for c in todo.launched)) + + def test_the_privilege_follows_the_probe(self): + """Appartenir au groupe libvirt suffit à joindre qemu:///system : + préfixer alors de « sudo » ne donne aucun droit de plus et réclame un + mot de passe pour rien. Les deux cas sont tenus ici, sans quoi la + réponse du sondage pourrait être ignorée sans que rien ne rougisse.""" + for joignable, attendu in ((True, ""), (False, "sudo ")): + with self.subTest(joignable=joignable): + todo = self._todo({"vm-a": "shut off"}) + todo._qemu_hw_form = lambda rows, node, nets=None: { + "vm-a": {"vcpus": 4, "ram": 8192, "gpu": True} + } + with mock.patch( + "script.todo.qemu_privilege.libvirt_reachable", + return_value=joignable, + ), mock.patch( + "script.todo.qemu_privilege.os.geteuid", return_value=1000 + ), mock.patch( + "script.todo.qemu_privilege.shutil.which", + return_value="/usr/bin/virsh", + ): + self._run(todo, ["vm-a"], ["o"]) + self.assertTrue(todo.launched) + for c in todo.launched: + self.assertEqual(c.startswith("sudo "), bool(attendu), c) def test_nothing_to_change_launches_nothing(self): todo = self._todo({"vm-a": "shut off"}) diff --git a/test/test_qemu_privilege.py b/test/test_qemu_privilege.py new file mode 100644 index 0000000..3385d58 --- /dev/null +++ b/test/test_qemu_privilege.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Le privilège se sonde, il ne se suppose pas. + +Appartenir au groupe libvirt suffit à joindre qemu:///system. Préfixer alors +chaque commande de « sudo » ne donne aucun droit de plus et réclame un mot de +passe à chaque entrée de menu. + +Ce que ces tests gardent : + +- la question se tranche en ESSAYANT ; /etc/group ne dit que ce qui est + DÉCLARÉ, et les groupes d'un processus sont figés à l'ouverture de session ; +- root ne demande jamais sudo, et une machine sans virsh non plus — une invite + de mot de passe pour une commande introuvable ne mène nulle part ; +- l'avertissement d'avant-installation se tait quand l'accès est déjà là. +""" + +import io +import sys +import unittest +from contextlib import redirect_stdout +from unittest import mock + +sys.argv = ["todo.py"] +from script.todo import qemu_privilege as qp # noqa: E402 +from script.todo.todo import TODO # noqa: E402 + + +class Sondage(unittest.TestCase): + def setUp(self): + qp.reset_cache() + + def tearDown(self): + qp.reset_cache() + + def _sonde(self, rc): + class Res: + returncode = rc + stdout = "" + stderr = "" + + return mock.patch.object(qp.subprocess, "run", return_value=Res()) + + def test_reachable_means_no_sudo(self): + with mock.patch.object( + qp.shutil, "which", return_value="/usr/bin/virsh" + ), mock.patch.object(qp.os, "geteuid", return_value=1000), self._sonde( + 0 + ): + self.assertFalse(qp.needs_sudo()) + self.assertEqual(qp.sudo_prefix(), "") + + def test_unreachable_means_sudo(self): + with mock.patch.object( + qp.shutil, "which", return_value="/usr/bin/virsh" + ), mock.patch.object(qp.os, "geteuid", return_value=1000), self._sonde( + 1 + ): + self.assertTrue(qp.needs_sudo()) + self.assertEqual(qp.sudo_prefix(), "sudo ") + + def test_root_never_needs_sudo(self): + with mock.patch.object( + qp.shutil, "which", return_value="/usr/bin/virsh" + ), mock.patch.object(qp.os, "geteuid", return_value=0), self._sonde(1): + self.assertFalse(qp.needs_sudo()) + + def test_without_virsh_no_password_prompt(self): + """Demander un mot de passe pour lancer une commande introuvable ne + mène nulle part : l'échec doit être « command not found ».""" + with mock.patch.object( + qp.shutil, "which", return_value=None + ), mock.patch.object(qp.os, "geteuid", return_value=1000): + self.assertFalse(qp.needs_sudo()) + + def test_the_probe_runs_once(self): + """Chaque entrée de menu la demande : un virsh par commande se + verrait.""" + with mock.patch.object( + qp.shutil, "which", return_value="/usr/bin/virsh" + ), mock.patch.object(qp.os, "geteuid", return_value=1000): + with self._sonde(0) as run: + for _ in range(5): + qp.needs_sudo() + self.assertEqual(run.call_count, 1) + + def test_a_dead_probe_falls_back_on_sudo(self): + """Un virsh qui n'arrive pas au bout ne prouve pas l'accès : mieux + vaut une invite de mot de passe qu'une commande refusée.""" + with mock.patch.object( + qp.shutil, "which", return_value="/usr/bin/virsh" + ), mock.patch.object( + qp.os, "geteuid", return_value=1000 + ), mock.patch.object( + qp.subprocess, "run", side_effect=OSError("boom") + ): + self.assertTrue(qp.needs_sudo()) + + +class AvertissementAvantInstallation(unittest.TestCase): + def setUp(self): + qp.reset_cache() + self.todo = TODO.__new__(TODO) + + def tearDown(self): + qp.reset_cache() + + def _rendu(self, joignable, declare, actif): + with mock.patch.object( + qp, "libvirt_reachable", return_value=joignable + ), mock.patch.object(qp, "group_state", return_value=(declare, actif)): + buf = io.StringIO() + with redirect_stdout(buf): + self.todo._qemu_warn_libvirt_access() + return buf.getvalue() + + def test_it_stays_quiet_when_access_is_there(self): + self.assertEqual("", self._rendu(True, True, True)) + + def test_it_names_usermod_when_the_group_is_missing(self): + rendu = self._rendu(False, False, False) + self.assertIn("usermod -aG libvirt", rendu) + + def test_a_declared_but_inactive_group_asks_for_a_new_session(self): + """Refaire un usermod déjà fait ne changerait rien : ce qui manque est + une session, pas une ligne dans /etc/group.""" + rendu = self._rendu(False, True, False) + self.assertIn("newgrp libvirt", rendu) + self.assertNotIn("usermod", rendu) + + def test_an_active_group_says_nothing_even_if_virsh_fails(self): + """Le groupe est porté par le processus : la cause est ailleurs + (démon arrêté, socket absente) et l'accuser tromperait.""" + self.assertEqual("", self._rendu(False, True, True)) + + +if __name__ == "__main__": + unittest.main()