[FIX] qemu : lire les baux dnsmasq sans invite de mot de passe
La lecture passait par « sudo sh -c cat » sans jamais demander si le privilège était nécessaire. Elle est appelée une fois par VM pour afficher une liste, et toutes les trois secondes pendant dix minutes par l'attente d'une VM : l'invite root tombait donc en boucle au milieu d'un écran, ce qui entraîne à taper un mot de passe root dans ce qui le demande. La lecture directe passe d'abord et suffit sur une installation standard, ces fichiers d'état étant en 0644 là où le « .conf » voisin est en 0600. Le privilège n'est tenté qu'ensuite, avec « -n », qui échoue au lieu de demander ; un répertoire interdit rend un glob vide, d'où l'essai même sans chemin trouvé. Les deux appelants se replient déjà sur une autre source. Vérifié : 12 tests. --- EN --- The read went through "sudo sh -c cat" without ever asking whether the privilege was needed. It is called once per VM to display a list, and every three seconds for ten minutes while waiting on a VM: the root prompt therefore landed in a loop in the middle of a screen, which trains someone to type a root password into whatever asks for it. The direct read comes first and suffices on a standard install, those status files being 0644 where the neighbouring ".conf" is 0600. The privilege is only tried afterwards, with "-n", which fails instead of asking; a forbidden directory returns an empty glob, hence the attempt even with no path found. Both callers already fall back to another source. Checked: 12 tests. Assisted-by: Claude Opus 5
This commit is contained in:
parent
eb28952d6c
commit
7c83dd1fc8
2 changed files with 287 additions and 16 deletions
|
|
@ -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:
|
||||
|
|
|
|||
208
test/test_qemu_lease_read.py
Normal file
208
test/test_qemu_lease_read.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue