From 4b60dcba308100bb910d5f51bf45d8b765629678 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Wed, 16 Sep 2026 03:00:17 -0400 Subject: [PATCH] =?UTF-8?q?[ADD]=20d=C3=A9ploiement=20qemu=20:=20couper=20?= =?UTF-8?q?l'audit=20npm,=20traduire=20l'exception?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- long_test/qemu_cache.py | 6 +++++ script/qemu/deploy_qemu.py | 40 +++++++++++++++++++++++++++++- script/todo/proxmox_menu.py | 14 ++++++++--- script/todo/qemu_deploy.py | 6 +++++ test/test_proxmox_form.py | 45 ++++++++++++++++++++++++++++++++-- test/test_qemu_cache_bypass.py | 17 +++++++++++++ test/test_qemu_cache_charge.py | 32 ++++++++++++++++++++++++ test/test_qemu_cache_form.py | 19 ++++++++++++++ test/test_qemu_cache_guest.py | 45 ++++++++++++++++++++++++++++++++++ 9 files changed, 218 insertions(+), 6 deletions(-) diff --git a/long_test/qemu_cache.py b/long_test/qemu_cache.py index 18af5c5..b128060 100755 --- a/long_test/qemu_cache.py +++ b/long_test/qemu_cache.py @@ -622,9 +622,13 @@ def deployer( distro=DISTRO, version=VERSION, charge="minimum", + hors_ligne=False, ): """Une VM Arch, branchée sur le cache ou non. Rend son adresse, ou ''. + `hors_ligne` : la VM naît l'amont du cache coupé. Le déploiement y + désactive ce qu'aucun cache ne rejoue, comme le ferait le formulaire. + Sans le cache, la VM télécharge en direct : c'est le TÉMOIN, la mesure de ce que coûte une installation quand rien n'est gardé. Un gain ne veut rien dire sans lui. @@ -645,6 +649,7 @@ def deployer( f" --ssh-key {shlex.quote(cle_publique())}" + ( f" --cache-ca {shlex.quote(CA)}" + + (" --offline" if hors_ligne else "") if avec_cache else " --cache-bypass" ) @@ -949,6 +954,7 @@ def contre_epreuve( distro=distro, version=version, charge=charge, + hors_ligne=True, ) if not adresse: return False diff --git a/script/qemu/deploy_qemu.py b/script/qemu/deploy_qemu.py index 01a1447..a70cd54 100755 --- a/script/qemu/deploy_qemu.py +++ b/script/qemu/deploy_qemu.py @@ -2523,6 +2523,14 @@ CACHE_TRUST = { # disparaît alors que la VM garde sa variable. CACHE_ENV_VARS = ("PIP_CERT", "REQUESTS_CA_BUNDLE", "NODE_EXTRA_CA_CERTS") +# Ce qu'une VM déployée l'amont du cache coupé reçoit en plus. L'audit de npm +# interroge un service distant qu'aucun cache ne peut rejouer : hors ligne il +# échoue à chaque installation sans rien vérifier. La variable reste dans la +# VM après son installation ; un « npm config set audit true » la contredit. +# Séparée de CACHE_ENV_VARS, que le binaire du cache recopie et que le test +# d'accord avec le Go compare. +OFFLINE_ENV_VARS = (("NPM_CONFIG_AUDIT", "false"),) + CACHE_CERT_NAME = "erplibre-cache.crt" @@ -2667,6 +2675,9 @@ def cache_bypass_apply(args: argparse.Namespace, runner: Runner) -> str: geste = ( f"{shlex.quote(CACHE_BIN)} --bypass-add {shlex.quote(mac)}" f" --bypass-name {shlex.quote(args.name)}" + # La commande passe par sudo, qui retire EL_LANG : la langue du + # message du binaire va donc en option. + f" --lang {shlex.quote(langue_des_messages(args))}" ) if shutil.which("nft"): # Le binaire écrit le fichier et rend sur sa sortie le geste à chaud. @@ -2681,6 +2692,20 @@ def cache_bypass_apply(args: argparse.Namespace, runner: Runner) -> str: return mac +def langue_des_messages(args: argparse.Namespace) -> str: + """La langue des messages : --lang du déploiement, sinon celle de todo.py, + sinon le français. Rend toujours « fr » ou « en ».""" + langue = (getattr(args, "lang", "") or "").strip().lower() + if not langue: + try: + from script.todo.todo_i18n import get_lang + + langue = get_lang() + except Exception: + langue = "fr" + return "en" if langue.startswith("en") else "fr" + + def cache_files(args: argparse.Namespace) -> list[tuple[str, str, str, str]]: """L'autorité du cache, posée par cloud-init à l'étape init. @@ -2742,6 +2767,12 @@ def cache_commands(args: argparse.Namespace) -> list[str]: f"sh -c 'grep -q ^{var}= /etc/environment" f" || echo {var}={faisceau} >> /etc/environment'" ) + if getattr(args, "offline", False): + for var, valeur in OFFLINE_ENV_VARS: + commandes.append( + f"sh -c 'grep -q ^{var}= /etc/environment" + f" || echo {var}={valeur} >> /etc/environment'" + ) return commandes @@ -2765,7 +2796,7 @@ def cache_env_reload(fichier: str = "/etc/environment") -> str: séparateur final, sans effet quand le fichier est absent ou ne les porte pas, et qui ne fait pas échouer une commande sous « set -e ». """ - motif = "|".join(CACHE_ENV_VARS) + motif = "|".join(CACHE_ENV_VARS + tuple(v for v, _ in OFFLINE_ENV_VARS)) return ( f'if [ -r {fichier} ]; then eval "$(grep -E "^({motif})=" {fichier}' ' | sed "s/^/export /")"; fi' @@ -4479,6 +4510,13 @@ def build_parser() -> argparse.ArgumentParser: "cette autorité dès son premier démarrage et ses téléchargements " "passent par le cache. Absent, rien n'est posé.", ) + g_cloud.add_argument( + "--offline", + action="store_true", + help="La VM est déployée l'amont du cache coupé : ce qu'aucun cache ne " + "peut rejouer y est désactivé, l'audit de npm d'abord. Sans effet " + "sans --cache-ca.", + ) g_cloud.add_argument( "--cache-bypass", action="store_true", diff --git a/script/todo/proxmox_menu.py b/script/todo/proxmox_menu.py index c675a9c..4c996e6 100644 --- a/script/todo/proxmox_menu.py +++ b/script/todo/proxmox_menu.py @@ -1690,9 +1690,12 @@ class ProxmoxMenuMixin: return "" return self._qemu_cache_ca_path() - def _pve_set_cache_ca(self, cible, vm, ca): + def _pve_set_cache_ca(self, cible, vm, ca, hors_ligne=False): """Pose l'autorité du cache DANS la VM, par ssh. + `hors_ligne` : l'amont du cache est coupé pour ce déploiement ; la VM + reçoit en plus ce que la voie libvirt pose sous « --offline ». + Même source que la voie libvirt — `cache_files` et `cache_commands` de deploy_qemu — livrée autrement : « qm set » ne sait écrire aucun fichier, comme pour le guide et pour le fuseau. @@ -1707,7 +1710,10 @@ class ProxmoxMenuMixin: except Exception: # pragma: no cover - dépend du module return False args = types.SimpleNamespace( - distro=vm.get("distro") or "", cache_ca=ca, cache_bypass=False + distro=vm.get("distro") or "", + cache_ca=ca, + cache_bypass=False, + offline=hors_ligne, ) fichiers = mod.cache_files(args) if not fichiers: @@ -2122,7 +2128,9 @@ class ProxmoxMenuMixin: # range ses index sous l'hôte demandé, et une VM qui en # réclame un autre ne retrouve rien de ce qui est gardé. self._pve_set_apt_mirror(vm["alias"], vm, mod_qemu) - self._pve_set_cache_ca(vm["alias"], vm, ca_cache) + self._pve_set_cache_ca( + vm["alias"], vm, ca_cache, hors_ligne=bool(coupee) + ) # Après la création, qui a posé l'écran accéléré : l'accès au # nœud de rendu est une affaire de COMPTE, et il se donne # dans l'invité. diff --git a/script/todo/qemu_deploy.py b/script/todo/qemu_deploy.py index 2ade1e6..2d64186 100644 --- a/script/todo/qemu_deploy.py +++ b/script/todo/qemu_deploy.py @@ -1114,6 +1114,7 @@ class QemuDeployMixin: git_email="", cache_ca="", cache_bypass=False, + offline=False, ): """Construit la commande deploy_qemu.py d'UNE VM (utilisée pour l'aperçu dry-run ET le déploiement réel).""" @@ -1163,6 +1164,10 @@ class QemuDeployMixin: # l'approuve dès son premier démarrage, sans quoi le détournement # lui présente un certificat qu'elle rejette. parts += ["--cache-ca", cache_ca] + if offline: + # L'amont du cache sera coupé : ce qu'aucun cache ne rejoue, + # l'audit de npm d'abord, est désactivé dans la VM. + parts.append("--offline") # L'identité git de la VM. Sans ces options, deploy_qemu recopie celle # de l'HÔTE : le formulaire la montre et permet de la changer, il ne # la remplace pas par du vide. @@ -1259,6 +1264,7 @@ class QemuDeployMixin: self._qemu_cache_ca_path() if self._qemu_cache_active() else "" ), cache_bypass=bool(spec.get("cache_bypass")), + offline=bool(spec.get("offline")), ) # Où l'installateur du cache pose son autorité. Un test compare cette diff --git a/test/test_proxmox_form.py b/test/test_proxmox_form.py index a519495..edc87c0 100644 --- a/test/test_proxmox_form.py +++ b/test/test_proxmox_form.py @@ -1262,7 +1262,7 @@ class TestLeMiroirAptDesVmProxmox(unittest.TestCase): ) self.assertLess( src.index('self._pve_set_apt_mirror(vm["alias"]'), - src.index('self._pve_set_cache_ca(vm["alias"]'), + src.index("self._pve_set_cache_ca("), ) @@ -2438,7 +2438,7 @@ class TestLAutoriteDuCacheDansUneVmImbriquee(unittest.TestCase): todo._pve_write_guide = lambda *a, **k: True todo._pve_set_timezone = lambda *a, **k: True todo._qemu_import_module = lambda: None - todo._pve_set_cache_ca = lambda cible, vm, ca: ordre.append( + todo._pve_set_cache_ca = lambda cible, vm, ca, **k: ordre.append( ("autorité", cible, ca) ) todo._qemu_install_erplibre_monitored = lambda *a, **k: ordre.append( @@ -2511,5 +2511,46 @@ class TestLAutoriteDuCacheDansUneVmImbriquee(unittest.TestCase): self.assertEqual(appels, []) +class TestLAutoriteDUneVmProxmoxHorsLigne(unittest.TestCase): + """Une VM Proxmox déployée hors ligne reçoit, comme la voie libvirt, de quoi + couper l'audit de npm : aucun cache ne rejoue ce service.""" + + def test_seule_la_vm_hors_ligne_coupe_l_audit(self): + import contextlib + import io + import sys + import tempfile + + sys.argv = ["todo.py"] + from script.todo.todo import TODO + + todo = TODO.__new__(TODO) + vu = {} + + def faux_ssh(cible, cmd, timeout=120): + vu["cmd"] = cmd + return 0, "" + + todo._pve_ssh = faux_ssh + with tempfile.NamedTemporaryFile( + "w", suffix=".crt", delete=False + ) as fh: + fh.write("-----BEGIN CERTIFICATE-----\nZm F1eA==\n") + fh.write("-----END CERTIFICATE-----\n") + ca = fh.name + rendus = {} + with contextlib.redirect_stdout(io.StringIO()): + for hors_ligne in (True, False): + todo._pve_set_cache_ca( + "hote+vm-a", + {"name": "vm-a", "distro": "ubuntu"}, + ca, + hors_ligne=hors_ligne, + ) + rendus[hors_ligne] = vu.get("cmd", "") + self.assertIn("NPM_CONFIG_AUDIT", rendus[True]) + self.assertNotIn("NPM_CONFIG_AUDIT", rendus[False]) + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/test/test_qemu_cache_bypass.py b/test/test_qemu_cache_bypass.py index 1bb9107..985599d 100644 --- a/test/test_qemu_cache_bypass.py +++ b/test/test_qemu_cache_bypass.py @@ -118,6 +118,23 @@ class TestLExceptionEstPoseeAvant(unittest.TestCase): 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.""" diff --git a/test/test_qemu_cache_charge.py b/test/test_qemu_cache_charge.py index 79ff885..5a007b2 100644 --- a/test/test_qemu_cache_charge.py +++ b/test/test_qemu_cache_charge.py @@ -250,5 +250,37 @@ class TestLeRapportSeClotSurUnEchec(unittest.TestCase): self.assertEqual(len(ecrit["vms"]), 1) +class TestLaTroisiemeVmNaitHorsLigne(unittest.TestCase): + """Le déploiement coupe l'audit de npm d'une VM hors ligne : le test long + doit le lui demander, comme le formulaire, sans quoi il mesure une VM que + personne ne déploierait ainsi.""" + + def commande(self, **kw): + from unittest import mock + + with mock.patch.object(QC, "dire") as dit, mock.patch.object( + QC, "cle_publique", return_value="/tmp/cle.pub" + ): + QC.deployer("vm", None, dry_run=True, **kw) + return " ".join(str(a) for c in dit.call_args_list for a in c.args) + + def test_la_vm_hors_ligne_recoit_offline(self): + self.assertIn("--offline", self.commande(hors_ligne=True)) + + def test_les_vm_en_ligne_n_en_recoivent_pas(self): + self.assertNotIn("--offline", self.commande()) + + def test_le_temoin_sans_cache_n_en_recoit_pas(self): + self.assertNotIn( + "--offline", self.commande(avec_cache=False, hors_ligne=True) + ) + + def test_la_contre_epreuve_le_demande(self): + import inspect + + source = inspect.getsource(QC.contre_epreuve) + self.assertIn("hors_ligne=True", source) + + if __name__ == "__main__": unittest.main() diff --git a/test/test_qemu_cache_form.py b/test/test_qemu_cache_form.py index 88844d5..bf31db7 100644 --- a/test/test_qemu_cache_form.py +++ b/test/test_qemu_cache_form.py @@ -121,6 +121,25 @@ class TestCommandeProduite(unittest.TestCase): todo._qemu_cache_active = lambda: actif and bool(ca) return todo._qemu_deploy_parts_for(self.vm(), spec, dry_run=True) + def test_hors_ligne_le_drapeau_offline(self): + """Hors ligne, la VM reçoit de quoi couper l'audit de npm.""" + parts = self.parts( + {"install": None, "offline": True}, ca="/tmp/essai-ca.crt" + ) + self.assertIn("--offline", parts) + + def test_en_ligne_aucun_drapeau_offline(self): + parts = self.parts({"install": None}, ca="/tmp/essai-ca.crt") + self.assertNotIn("--offline", parts) + + def test_une_vm_exceptee_n_a_pas_de_drapeau_offline(self): + """Elle ne traverse pas le cache : rien à couper chez elle.""" + parts = self.parts( + {"install": None, "offline": True, "cache_bypass": True}, + ca="/tmp/essai-ca.crt", + ) + self.assertNotIn("--offline", parts) + def test_service_arrete_aucun_drapeau(self): """Sans interception, l'autorité n'a rien à faire dans la VM.""" parts = self.parts({"install": None}, ca="/tmp/ca.crt", actif=False) diff --git a/test/test_qemu_cache_guest.py b/test/test_qemu_cache_guest.py index 35a7d23..e1f44e8 100644 --- a/test/test_qemu_cache_guest.py +++ b/test/test_qemu_cache_guest.py @@ -33,10 +33,13 @@ from script.qemu.deploy_qemu import ( # noqa: E402 CACHE_CERT_NAME, CACHE_ENV_VARS, CACHE_TRUST, + OFFLINE_ENV_VARS, cache_env_reload, cache_family, cache_files, + cache_commands, cache_runcmd, + langue_des_messages, ) RULES_GO = RACINE / "script" / "qemu_cache" / "rules.go" @@ -228,6 +231,48 @@ class TestLesVariablesRelues(unittest.TestCase): self.assertEqual(code, 0, err) +class TestLeHorsLigneCoupeLAuditNpm(unittest.TestCase): + """Hors ligne, l'audit de npm interroge un service qu'aucun cache ne rejoue.""" + + def setUp(self): + import tempfile + + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + + def commandes(self, offline): + args = faux_args(Path(self.tmp.name), "ubuntu") + args.offline = offline + return "\n".join(cache_commands(args)) + + def test_une_vm_hors_ligne_recoit_l_audit_coupe(self): + self.assertIn("NPM_CONFIG_AUDIT=false", self.commandes(True)) + + def test_une_vm_en_ligne_garde_son_audit(self): + self.assertNotIn("NPM_CONFIG_AUDIT", self.commandes(False)) + + def test_la_variable_est_relue_apres_cloud_init(self): + """Le « npm install » lancé sans sudo vit dans une session ouverte + avant que cloud-init n'écrive la variable.""" + for var, _ in OFFLINE_ENV_VARS: + self.assertIn(var, cache_env_reload()) + + +class TestLaLangueDesMessagesDuCache(unittest.TestCase): + def test_l_option_du_deploiement_l_emporte(self): + self.assertEqual( + langue_des_messages(argparse.Namespace(lang="en")), "en" + ) + self.assertEqual( + langue_des_messages(argparse.Namespace(lang="FR")), "fr" + ) + + def test_une_valeur_inconnue_rend_le_francais(self): + self.assertEqual( + langue_des_messages(argparse.Namespace(lang="de")), "fr" + ) + + class TestAccordAvecLeGo(unittest.TestCase): """La table Python et la table Go doivent dire la même chose.