diff --git a/script/todo/qemu_manage.py b/script/todo/qemu_manage.py index e565838..9deafa5 100644 --- a/script/todo/qemu_manage.py +++ b/script/todo/qemu_manage.py @@ -22,6 +22,78 @@ from script.todo.qemu_privilege import ( from script.todo.todo_i18n import t +# Les fichiers d'état de dnsmasq, un par réseau libvirt. Sur une installation +# standard ils sont en 0644 dans un répertoire en 0755, donc lisibles sans +# privilège — le « .conf » posé à côté est en 0600, et c'est lui qui donne +# l'impression que le répertoire est fermé. +DNSMASQ_STATUS = "/var/lib/libvirt/dnsmasq/*.status" + + +def lease_status_text(*, paths=None, read=None, run=None, euid=None) -> str: + """Le contenu des fichiers d'état de dnsmasq, ou la chaîne vide. + + La lecture DIRECTE passe d'abord, et elle suffit sur une installation + standard, où ces fichiers sont en lecture pour tous. Ce n'est qu'ensuite + que « sudo -n » est tenté, pour l'installation durcie — jamais « sudo » + tout court. L'attente d'une VM appelle ce chemin + toutes les trois secondes pendant dix minutes, et l'affichage d'une liste + l'appelle une fois par VM : un sudo interactif y demande donc un mot de + passe root en boucle, au milieu d'un écran, ce qui entraîne précisément le + réflexe de le taper dans ce qui le demande. + + `needs_sudo()` ne tranche pas ici : il dit si libvirt est joignable, pas + si un fichier de root est lisible. Les deux questions n'ont ni la même + réponse ni la même cause. + + Rend la chaîne vide quand rien n'est lisible — un répertoire interdit rend + un glob VIDE, indistinguable de « aucun réseau », d'où l'essai privilégié + même sans chemin trouvé. Les appelants se replient déjà sur une autre + source. + """ + if paths is None: + paths = glob.glob + if read is None: + + def read(chemin): + with open(chemin, encoding="utf-8") as fh: + return fh.read() + + if run is None: + run = subprocess.run + if euid is None: + euid = os.geteuid + + morceaux = [] + for chemin in sorted(paths(DNSMASQ_STATUS)): + try: + morceaux.append(read(chemin)) + except OSError: + morceaux = [] + break + if morceaux: + return "".join(morceaux) + if euid() == 0: + # Root a déjà tout vu : il n'y a rien à lire, et sudo n'y changerait + # rien. + return "" + try: + res = run( + [ + "sudo", + "-n", + "sh", + "-c", + f"cat {DNSMASQ_STATUS} 2>/dev/null", + ], + capture_output=True, + text=True, + timeout=10, + ) + except (OSError, subprocess.SubprocessError): + return "" + return res.stdout or "" + + def parse_ssh_blocks(content) -> dict: """{nom: {"hostname": …, "proxyjump": …}} pour CHAQUE nom déclaré. @@ -3022,23 +3094,14 @@ class QemuManageMixin: @staticmethod def _qemu_lease_ip_for_host(name, candidates): """Parmi `candidates`, l'IP dont le bail dnsmasq porte le hostname de la - VM (le bail DÉFINITIF, pas le bail précoce « ubuntu »). None sinon.""" - try: - res = subprocess.run( - [ - "sudo", - "sh", - "-c", - "cat /var/lib/libvirt/dnsmasq/*.status 2>/dev/null", - ], - capture_output=True, - text=True, - timeout=10, - ) - except (OSError, subprocess.SubprocessError): - return None + VM (le bail DÉFINITIF, pas le bail précoce « ubuntu »). None sinon. + + La lecture ne demande JAMAIS de mot de passe : `lease_status_text` lit + en direct puis tente « sudo -n », et rend la chaîne vide plutôt que + d'ouvrir une invite. Les deux appelants se replient sur une autre + source quand ceci rend None.""" # Plusieurs tableaux JSON concaténés : on parse chaque objet {...}. - for obj in re.findall(r"\{[^{}]*\}", res.stdout or ""): + for obj in re.findall(r"\{[^{}]*\}", lease_status_text()): if re.search(rf'"hostname":\s*"{re.escape(name)}"', obj): m = re.search(r'"ip-address":\s*"([\d.]+)"', obj) if m and m.group(1) in candidates: diff --git a/test/test_qemu_lease_read.py b/test/test_qemu_lease_read.py new file mode 100644 index 0000000..789a15b --- /dev/null +++ b/test/test_qemu_lease_read.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""La lecture des baux dnsmasq : jamais une invite de mot de passe. + +Ce que ces tests défendent n'est pas une fonctionnalité, c'est une +ABSENCE. Le chemin lisait les baux par « sudo sh -c cat », sans jamais +demander s'il en avait besoin. Or il est appelé une fois par VM pour +afficher une liste, et toutes les trois secondes pendant dix minutes par +l'attente d'une VM : l'invite de mot de passe root tombait donc au milieu +d'un écran, en boucle, et entraînait à taper un mot de passe root dans ce +qui le demande. C'est l'habitude qui coûte, pas l'appel. + +Trois règles en sortent, et chacune a son test. + +La lecture directe passe d'abord, et elle suffit sur une installation +standard : ces fichiers d'état sont en lecture pour tous, contrairement au +« .conf » posé à côté. Le privilège n'est tenté qu'ensuite, et seulement +avec « -n », qui échoue au lieu de demander. Et un répertoire interdit rend +un glob VIDE, indistinguable de « aucun réseau » — d'où l'essai privilégié +même sans chemin trouvé. + +Les adresses sont INVENTÉES, comme l'exige la règle du dépôt pour tout +exemple qui illustre un interdit : 192.168.199.x ne paraît nulle part +ailleurs. +""" + +import unittest +from unittest import mock + +from script.todo.qemu_manage import DNSMASQ_STATUS, lease_status_text + +BAIL = ( + '{"ip-address":"192.168.199.24","mac-address":"52:54:00:aa:bb:cc",' + '"hostname":"machine-un","expiry-time":1788000000}' +) +BAIL_AUTRE = ( + '{"ip-address":"192.168.199.31","mac-address":"52:54:00:dd:ee:ff",' + '"hostname":"machine-deux","expiry-time":1788000001}' +) + + +class Appels: + """Un exécuteur qui note ce qu'on lui a demandé de lancer.""" + + def __init__(self, stdout="", leve=None): + self.stdout = stdout + self.leve = leve + self.argvs = [] + + def __call__(self, argv, **kwargs): + self.argvs.append(argv) + if self.leve: + raise self.leve + return mock.Mock(stdout=self.stdout, returncode=0) + + +class TestLectureDirecte(unittest.TestCase): + def test_direct_read_wins(self): + run = Appels(stdout="ne doit pas servir") + texte = lease_status_text( + paths=lambda motif: ["/a.status", "/b.status"], + read=lambda chemin: BAIL if chemin == "/a.status" else BAIL_AUTRE, + run=run, + euid=lambda: 1000, + ) + self.assertIn("192.168.199.24", texte) + self.assertIn("192.168.199.31", texte) + self.assertEqual(run.argvs, [], "aucun privilège n'était nécessaire") + + def test_readable_but_empty_does_not_escalate(self): + """Un réseau sans bail rend un fichier VIDE, pas une interdiction. + + Escalader ici lancerait un sudo par VM sur toute machine dont le + réseau libvirt n'a encore servi aucun bail.""" + run = Appels(stdout=BAIL) + texte = lease_status_text( + paths=lambda motif: ["/virbr0.status"], + read=lambda chemin: "", + run=run, + euid=lambda: 1000, + ) + self.assertEqual(texte, "") + self.assertEqual(run.argvs, []) + + def test_paths_are_sorted(self): + """L'ordre de lecture ne dépend pas de celui du système de fichiers.""" + lus = [] + + def read(chemin): + lus.append(chemin) + return "" + + lease_status_text( + paths=lambda motif: ["/z.status", "/a.status"], + read=read, + run=Appels(), + euid=lambda: 1000, + ) + self.assertEqual(lus, ["/a.status", "/z.status"]) + + +class TestReplisPrivilegies(unittest.TestCase): + def test_unreadable_file_escalates(self): + run = Appels(stdout=BAIL) + texte = lease_status_text( + paths=lambda motif: ["/a.status"], + read=mock.Mock(side_effect=PermissionError(13, "refusé")), + run=run, + euid=lambda: 1000, + ) + self.assertIn("192.168.199.24", texte) + self.assertEqual(len(run.argvs), 1) + + def test_empty_glob_escalates(self): + """Un répertoire interdit rend un glob vide, non une erreur.""" + run = Appels(stdout=BAIL) + texte = lease_status_text( + paths=lambda motif: [], + read=lambda chemin: "", + run=run, + euid=lambda: 1000, + ) + self.assertIn("192.168.199.24", texte) + + def test_sudo_is_never_interactive(self): + """« -n » suit « sudo » immédiatement : sudo échoue au lieu de demander. + + C'est la seule assertion de ce fichier qui porte sur la FORME de + l'argv, et elle le fait parce que l'ordre compte : « sudo sh -c … -n » + passerait le drapeau au shell, pas à sudo.""" + run = Appels(stdout="") + lease_status_text( + paths=lambda motif: [], + read=lambda chemin: "", + run=run, + euid=lambda: 1000, + ) + argv = run.argvs[0] + self.assertEqual(argv[0], "sudo") + self.assertEqual(argv[1], "-n") + self.assertIn(DNSMASQ_STATUS, " ".join(argv)) + + def test_root_does_not_escalate(self): + """Root a déjà tout vu : sudo n'y changerait rien.""" + run = Appels(stdout=BAIL) + texte = lease_status_text( + paths=lambda motif: [], + read=lambda chemin: "", + run=run, + euid=lambda: 0, + ) + self.assertEqual(texte, "") + self.assertEqual(run.argvs, []) + + def test_sudo_absent_returns_empty(self): + texte = lease_status_text( + paths=lambda motif: [], + read=lambda chemin: "", + run=Appels(leve=FileNotFoundError(2, "sudo")), + euid=lambda: 1000, + ) + self.assertEqual(texte, "") + + def test_sudo_refusal_returns_empty(self): + """« sudo -n » sans droit rend un code non nul et un stdout vide.""" + texte = lease_status_text( + paths=lambda motif: [], + read=lambda chemin: "", + run=lambda argv, **kw: mock.Mock(stdout="", returncode=1), + euid=lambda: 1000, + ) + self.assertEqual(texte, "") + + +class TestBailParHostname(unittest.TestCase): + """Le parcours qui consomme le texte, pour que le repli reste vrai.""" + + def _chercher(self, texte, nom, candidates): + from script.todo import qemu_manage + + with mock.patch.object( + qemu_manage, "lease_status_text", return_value=texte + ): + return qemu_manage.QemuManageMixin._qemu_lease_ip_for_host( + nom, candidates + ) + + def test_the_lease_naming_the_vm_wins(self): + trouve = self._chercher( + BAIL + BAIL_AUTRE, + "machine-deux", + ["192.168.199.24", "192.168.199.31"], + ) + self.assertEqual(trouve, "192.168.199.31") + + def test_a_lease_outside_the_candidates_is_ignored(self): + """Un bail périmé nomme la VM sans être une candidate joignable.""" + trouve = self._chercher(BAIL, "machine-un", ["192.168.199.99"]) + self.assertIsNone(trouve) + + def test_no_readable_lease_returns_none(self): + """Le cas du privilège refusé : None, et l'appelant se replie.""" + self.assertIsNone(self._chercher("", "machine-un", ["192.168.199.24"])) + + +if __name__ == "__main__": + unittest.main()