[FIX] qemu menu : passer par le groupe libvirt plutôt que par sudo
Appartenir au groupe libvirt suffit à joindre qemu:///system : le sudo écrit en dur n'y ajoutait aucun droit et réclamait un mot de passe à chaque entrée de menu. La question se tranche en ESSAYANT, jamais en lisant /etc/group : les groupes d'un processus sont figés à l'ouverture de session, donc la table dit le déclaré, l'essai le faisable. C'est la distinction que porte aussi l'avertissement d'avant-installation. Un hyperviseur distant garde sudo, ses droits ne se sondant pas d'ici. Vérifié : 10 tests, rougis par deux mutations — lire /etc/group, et conclure « pas de sudo » sur un sondage mort. --- EN --- Membership of the libvirt group is enough to reach qemu:///system: the hardcoded sudo added no right there and asked for a password at every menu entry. The question is settled by TRYING, never by reading /etc/group: a process's groups are frozen at session start, so the table states what is declared, the attempt what is doable. The pre-install warning carries that same distinction. A remote hypervisor keeps sudo, its rights not being probeable from here. Checked: 10 tests, turned red by two mutations — reading /etc/group, and concluding « no sudo » from a dead probe. Assisted-by: Claude Opus 5
This commit is contained in:
parent
d3e8d50953
commit
feb6cd7b12
9 changed files with 366 additions and 29 deletions
|
|
@ -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 '<graphics'"
|
||||
f" {pre}{su}virsh dumpxml {domain} | grep -A2 '<graphics'"
|
||||
)
|
||||
print(
|
||||
f" {t('To open it on the loopback (VM restart required):')}"
|
||||
)
|
||||
print(f" {pre}sudo virsh destroy {domain}")
|
||||
print(f" {pre}{su}virsh destroy {domain}")
|
||||
print(
|
||||
f" {pre}sudo virsh edit {domain} # <graphics type='vnc'"
|
||||
f" {pre}{su}virsh edit {domain} # <graphics type='vnc'"
|
||||
" port='-1' autoport='yes' listen='127.0.0.1'/>"
|
||||
)
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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():
|
||||
|
|
|
|||
|
|
@ -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 <id> »
|
||||
# é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)
|
||||
|
|
|
|||
|
|
@ -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): "))
|
||||
):
|
||||
|
|
|
|||
104
script/todo/qemu_privilege.py
Normal file
104
script/todo/qemu_privilege.py
Normal file
|
|
@ -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 ""
|
||||
|
|
@ -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:",
|
||||
|
|
|
|||
|
|
@ -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"})
|
||||
|
|
|
|||
139
test/test_qemu_privilege.py
Normal file
139
test/test_qemu_privilege.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue