Mathieu Benoit
4b60dcba30
L'audit de npm interroge un service qu'aucun cache ne rejoue : amont coupé, il échoue à chaque installation sans rien vérifier. Une VM déployée hors ligne reçoit donc NPM_CONFIG_AUDIT=false, écrit dans /etc/environment et relu après cloud-init ; une VM en ligne garde son audit. Le retrait d'une exception du détournement parle désormais la langue du menu, son message allant à l'erreur standard tandis que la sortie standard garde les règles que nft lit, qui ne se traduisent jamais. --- EN --- The npm audit queries a service no cache can replay: upstream cut, it fails every install without checking anything. A VM deployed offline therefore gets NPM_CONFIG_AUDIT=false, written in /etc/environment and read back after cloud-init; an online VM keeps its audit. Removing a redirection exception now speaks the menu's language, its message going to standard error while standard output keeps the rules nft reads, which are never translated. Assisted-by: Claude Opus 5
320 lines
12 KiB
Python
320 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# © 2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
"""Soustraire UNE VM au cache : l'exception par adresse MAC.
|
|
|
|
Le détournement est transparent et vaut pour tout le pont. Une VM ne peut pas
|
|
s'y soustraire de l'intérieur, et lui retirer l'autorité ne la dispense de
|
|
rien : elle est interceptée quand même et échoue sur un certificat qu'elle ne
|
|
reconnaît pas. L'exception se pose donc sur l'HÔTE, et elle a besoin d'un
|
|
identifiant que la VM porte de façon stable.
|
|
|
|
D'où ce que ces tests tiennent. L'adresse MAC est choisie AVANT la création :
|
|
la lire après laisserait la fenêtre où cloud-init télécharge déjà. Une MAC
|
|
posée à la main l'emporte sur le tirage. Et l'exception l'emporte sur
|
|
l'autorité, qu'il serait absurde de faire approuver à une VM qui ne
|
|
rencontrera jamais le cache.
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
RACINE = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(RACINE))
|
|
|
|
from script.qemu import deploy_qemu # noqa: E402
|
|
|
|
|
|
class FauxRunner:
|
|
"""Retient les commandes. Rend ce qu'on lui a dit de rendre."""
|
|
|
|
def __init__(self, reponses=None):
|
|
self.commandes = []
|
|
self.reponses = reponses or {}
|
|
self.use_sudo = True
|
|
self.dry_run = False
|
|
|
|
def run(self, cmd, *, privileged=False, check=True, capture=False):
|
|
self.commandes.append(list(cmd))
|
|
for motif, reponse in self.reponses.items():
|
|
if motif in " ".join(cmd):
|
|
return reponse
|
|
return (0, "") if capture else None
|
|
|
|
|
|
def args_neufs(**kw):
|
|
base = dict(
|
|
name="vm-essai",
|
|
network="network=default,model=virtio",
|
|
cache_bypass=True,
|
|
cache_ca="",
|
|
distro="arch",
|
|
dry_run=False,
|
|
)
|
|
base.update(kw)
|
|
return argparse.Namespace(**base)
|
|
|
|
|
|
class TestLaMACChoisie(unittest.TestCase):
|
|
def test_une_mac_deja_demandee_lemporte(self):
|
|
"""Une MAC posée à la main sert une réservation DHCP : la remplacer
|
|
casserait ce que l'appelant a réglé ailleurs."""
|
|
reseau = "network=default,model=virtio,mac=52:54:00:de:ad:be"
|
|
self.assertEqual(
|
|
deploy_qemu.mac_du_network(reseau), "52:54:00:de:ad:be"
|
|
)
|
|
self.assertEqual(
|
|
deploy_qemu.network_avec_mac(reseau, "52:54:00:11:11:11"), reseau
|
|
)
|
|
|
|
def test_sans_mac_elle_est_ajoutee(self):
|
|
reseau = deploy_qemu.network_avec_mac(
|
|
"network=default,model=virtio", "52:54:00:11:22:33"
|
|
)
|
|
self.assertEqual(
|
|
reseau, "network=default,model=virtio,mac=52:54:00:11:22:33"
|
|
)
|
|
self.assertEqual(
|
|
deploy_qemu.mac_du_network(reseau), "52:54:00:11:22:33"
|
|
)
|
|
|
|
def test_la_mac_tiree_porte_le_prefixe_qemu(self):
|
|
mac = deploy_qemu.mac_neuve(set())
|
|
self.assertTrue(
|
|
mac.startswith(deploy_qemu.MAC_PREFIXE),
|
|
f"{mac} n'est pas une adresse QEMU",
|
|
)
|
|
self.assertEqual(len(mac.split(":")), 6)
|
|
|
|
def test_une_mac_prise_nest_pas_retiree(self):
|
|
"""Libvirt refuse un doublon, mais à la création — après le
|
|
téléchargement de l'image. Le refus doit venir avant."""
|
|
prises = set()
|
|
for _ in range(30):
|
|
mac = deploy_qemu.mac_neuve(prises)
|
|
self.assertNotIn(mac, prises)
|
|
prises.add(mac)
|
|
|
|
|
|
class TestLExceptionEstPoseeAvant(unittest.TestCase):
|
|
def test_la_mac_entre_dans_le_reseau_avant_la_creation(self):
|
|
args = args_neufs()
|
|
runner = FauxRunner({"is-active": (0, "")})
|
|
with mock.patch.object(
|
|
deploy_qemu.os.path, "isfile", return_value=True
|
|
), mock.patch.object(
|
|
deploy_qemu.shutil, "which", return_value="/usr/bin/nft"
|
|
), mock.patch.object(
|
|
deploy_qemu, "macs_deja_prises", return_value=set()
|
|
):
|
|
mac = deploy_qemu.cache_bypass_apply(args, runner)
|
|
self.assertTrue(mac, "aucune MAC retenue")
|
|
self.assertIn(f"mac={mac}", args.network)
|
|
pose = [c for c in runner.commandes if "--bypass-add" in " ".join(c)]
|
|
self.assertTrue(pose, "l'exception n'a pas été posée")
|
|
self.assertIn("nft -f -", " ".join(pose[0]))
|
|
|
|
def test_le_message_du_binaire_suit_la_langue(self):
|
|
"""La commande passe par sudo, qui retire EL_LANG : sans l'option, le
|
|
binaire répondrait en français à qui déploie en anglais."""
|
|
args = args_neufs()
|
|
args.lang = "en"
|
|
runner = FauxRunner({"is-active": (0, "")})
|
|
with mock.patch.object(
|
|
deploy_qemu.os.path, "isfile", return_value=True
|
|
), mock.patch.object(
|
|
deploy_qemu.shutil, "which", return_value="/usr/bin/nft"
|
|
), mock.patch.object(
|
|
deploy_qemu, "macs_deja_prises", return_value=set()
|
|
):
|
|
deploy_qemu.cache_bypass_apply(args, runner)
|
|
pose = [c for c in runner.commandes if "--bypass-add" in " ".join(c)]
|
|
self.assertIn("--lang en", " ".join(pose[0]))
|
|
|
|
def test_sans_cache_installe_rien_nest_pose(self):
|
|
"""Exiger un cache pour pouvoir s'en passer n'aurait aucun sens :
|
|
sans lui, rien n'intercepte."""
|
|
args = args_neufs()
|
|
runner = FauxRunner()
|
|
with mock.patch.object(
|
|
deploy_qemu.os.path, "isfile", return_value=False
|
|
):
|
|
self.assertEqual(deploy_qemu.cache_bypass_apply(args, runner), "")
|
|
self.assertEqual(runner.commandes, [])
|
|
self.assertNotIn("mac=", args.network)
|
|
|
|
def test_service_arrete_rien_nest_pose(self):
|
|
args = args_neufs()
|
|
runner = FauxRunner({"is-active": (3, "")})
|
|
with mock.patch.object(
|
|
deploy_qemu.os.path, "isfile", return_value=True
|
|
):
|
|
self.assertEqual(deploy_qemu.cache_bypass_apply(args, runner), "")
|
|
self.assertNotIn("mac=", args.network)
|
|
|
|
def test_sans_la_demande_rien_ne_bouge(self):
|
|
args = args_neufs(cache_bypass=False)
|
|
runner = FauxRunner()
|
|
self.assertEqual(deploy_qemu.cache_bypass_apply(args, runner), "")
|
|
self.assertEqual(runner.commandes, [])
|
|
|
|
def test_sans_nft_le_service_est_repose(self):
|
|
"""iptables n'a pas d'ensemble nommé : l'exception n'entre qu'en
|
|
reposant la chaîne, et seul un redémarrage la repose."""
|
|
args = args_neufs()
|
|
runner = FauxRunner({"is-active": (0, "")})
|
|
with mock.patch.object(
|
|
deploy_qemu.os.path, "isfile", return_value=True
|
|
), mock.patch.object(
|
|
deploy_qemu.shutil, "which", return_value=None
|
|
), mock.patch.object(
|
|
deploy_qemu, "macs_deja_prises", return_value=set()
|
|
):
|
|
deploy_qemu.cache_bypass_apply(args, runner)
|
|
dit = [" ".join(c) for c in runner.commandes]
|
|
self.assertTrue(
|
|
any("systemctl" in c and "restart" in c for c in dit),
|
|
f"le service n'est pas reposé : {dit}",
|
|
)
|
|
|
|
|
|
class TestLExceptionLemporteSurLAutorite(unittest.TestCase):
|
|
def test_pas_dautorite_dans_une_vm_exceptee(self):
|
|
"""Elle ne rencontrera jamais le cache : poser cette signature dans
|
|
son magasin n'aurait aucun objet."""
|
|
args = args_neufs(cache_ca="/var/lib/erplibre_go_qemu_cache/ca.crt")
|
|
self.assertEqual(deploy_qemu.cache_files(args), [])
|
|
|
|
def test_la_ligne_de_commande_ne_porte_pas_les_deux(self):
|
|
from script.todo.qemu_deploy import QemuDeployMixin
|
|
|
|
parts = QemuDeployMixin._qemu_build_deploy_parts(
|
|
mock.MagicMock(_qemu_script_path=lambda: "deploy_qemu.py"),
|
|
"arch",
|
|
"latest",
|
|
"x86_64",
|
|
"vm",
|
|
4096,
|
|
2,
|
|
"20G",
|
|
"",
|
|
"master",
|
|
True,
|
|
cache_ca="/var/lib/erplibre_go_qemu_cache/ca.crt",
|
|
cache_bypass=True,
|
|
)
|
|
self.assertIn("--cache-bypass", parts)
|
|
self.assertNotIn("--cache-ca", parts)
|
|
|
|
def test_sans_exception_lautorite_passe(self):
|
|
"""Le cas courant : la VM approuve l'autorité et traverse le cache."""
|
|
from script.todo.qemu_deploy import QemuDeployMixin
|
|
|
|
parts = QemuDeployMixin._qemu_build_deploy_parts(
|
|
mock.MagicMock(_qemu_script_path=lambda: "deploy_qemu.py"),
|
|
"arch",
|
|
"latest",
|
|
"x86_64",
|
|
"vm",
|
|
4096,
|
|
2,
|
|
"20G",
|
|
"",
|
|
"master",
|
|
True,
|
|
cache_ca="/var/lib/erplibre_go_qemu_cache/ca.crt",
|
|
)
|
|
self.assertIn("--cache-ca", parts)
|
|
self.assertNotIn("--cache-bypass", parts)
|
|
|
|
|
|
class TestLeMenageDesExceptions(unittest.TestCase):
|
|
"""Une exception qui survit à sa VM est le danger de cette liste.
|
|
|
|
L'adresse MAC est libérée par la destruction et se réattribue : l'entrée
|
|
restée derrière soustrairait au cache une machine neuve que personne n'a
|
|
exceptée. Ni la VM ni le cache ne le diraient — la VM télécharge
|
|
normalement, le journal du cache reste seulement muet à son sujet.
|
|
"""
|
|
|
|
def test_une_vm_disparue_rend_son_entree_orpheline(self):
|
|
from script.todo.qemu_cache_menu import QemuCacheMenuMixin as M
|
|
|
|
entrees = [
|
|
("52:54:00:00:00:01", "vm-vivante"),
|
|
("52:54:00:00:00:02", "vm-detruite"),
|
|
]
|
|
with mock.patch.object(
|
|
M, "_cache_domaines", return_value={"vm-vivante"}
|
|
):
|
|
orphelines = M._cache_bypass_orphelines(entrees)
|
|
self.assertEqual(orphelines, [("52:54:00:00:00:02", "vm-detruite")])
|
|
|
|
def test_une_entree_sans_nom_nest_jamais_orpheline(self):
|
|
"""Posée à la main : rien ne dit à quelle VM elle se rapporte, et la
|
|
retirer d'office déferait le travail de quelqu'un."""
|
|
from script.todo.qemu_cache_menu import QemuCacheMenuMixin as M
|
|
|
|
with mock.patch.object(M, "_cache_domaines", return_value=set()):
|
|
self.assertEqual(
|
|
M._cache_bypass_orphelines([("52:54:00:00:00:03", "")]), []
|
|
)
|
|
|
|
def test_le_menage_retire_du_fichier_et_du_noyau(self):
|
|
"""Les deux : le fichier est ce que le service reposera, l'ensemble
|
|
est ce qui s'applique en ce moment."""
|
|
from script.todo import qemu_cache_menu as menu
|
|
|
|
execute = mock.MagicMock()
|
|
with mock.patch.object(
|
|
menu.os.path, "isfile", return_value=True
|
|
), mock.patch.object(
|
|
menu.QemuCacheMenuMixin,
|
|
"_cache_bypass_orphelines",
|
|
return_value=[("52:54:00:00:00:04", "partie")],
|
|
):
|
|
self.assertEqual(menu.bypass_menage(execute), 1)
|
|
cmd = execute.exec_command_live.call_args.args[0]
|
|
self.assertIn("--bypass-del 52:54:00:00:00:04", cmd)
|
|
self.assertIn("nft -f -", cmd)
|
|
|
|
def test_sans_cache_pose_le_menage_ne_fait_rien(self):
|
|
from script.todo import qemu_cache_menu as menu
|
|
|
|
execute = mock.MagicMock()
|
|
with mock.patch.object(menu.os.path, "isfile", return_value=False):
|
|
self.assertEqual(menu.bypass_menage(execute), 0)
|
|
execute.exec_command_live.assert_not_called()
|
|
|
|
def test_la_liste_est_lue_du_binaire(self):
|
|
"""Lui seul normalise une adresse et saute une ligne fautive : une
|
|
seconde lecture écrite ici dériverait de la sienne."""
|
|
from script.todo.qemu_cache_menu import QemuCacheMenuMixin as M
|
|
|
|
sortie = "52:54:00:aa:bb:cc vm-une\n52:54:00:11:22:33\n\n"
|
|
with mock.patch.object(
|
|
M, "_cache_lire", return_value=sortie
|
|
), mock.patch("os.path.isfile", return_value=True):
|
|
self.assertEqual(
|
|
M._cache_bypass_lire(),
|
|
[("52:54:00:aa:bb:cc", "vm-une"), ("52:54:00:11:22:33", "")],
|
|
)
|
|
|
|
|
|
class TestLaSuppressionFaitLeMenage(unittest.TestCase):
|
|
def test_la_suppression_de_vm_appelle_le_menage(self):
|
|
"""Le seul endroit qui SAIT que la VM vient de disparaître."""
|
|
src = (RACINE / "script" / "todo" / "qemu_manage.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
bloc = src[src.index("def _qemu_delete_vm(") :]
|
|
bloc = bloc[: bloc.index("\n def ")]
|
|
self.assertIn("bypass_menage(self.execute)", bloc)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|