[FIX] cache qemu : soustraire l'invité qui n'a pas de magasin
Le détournement est TRANSPARENT et vaut pour tout le pont : ne pas donner l'autorité à une VM ne la dispense pas d'être interceptée, elle échoue sur « self-signed certificate in certificate chain ». NixOS n'a pas d'ancre de confiance par fichier, et la poser par déclaration arriverait trop tard — la première reconstruction EST le premier téléchargement. La VM est donc exceptée par son adresse MAC, avant sa création, et cela se dit. L'image Proxmox n'est mise en place qu'une fois complète : « wget -O » écrivait dans la cible, et une coupure y figeait un fichier tronqué que le test de présence acceptait à chaque déploiement suivant. --- EN --- Interception is TRANSPARENT and covers the whole bridge: withholding the authority from a VM does not spare it, it fails on "self-signed certificate in certificate chain". NixOS has no per-file trust anchor, and declaring one would come too late — the first rebuild IS the first download. The VM is therefore exempted by MAC, before creation, and it is said. The Proxmox image is put in place only once complete: "wget -O" wrote into the target, and an interruption froze a truncated file there that the presence test accepted on every later deployment. Assisted-by: Claude Opus 5
This commit is contained in:
parent
f3edd806bc
commit
4f4664f804
11 changed files with 399 additions and 29 deletions
|
|
@ -37,7 +37,39 @@ from script.proxmox import proxmox_deploy as pve # noqa: E402
|
|||
# Les scripts qui lancent une descente. Le verrou les cherche TOUS : deux
|
||||
# descentes de piles différentes se disputeraient la RAM, le disque et
|
||||
# ~/.ssh/config aussi sûrement que deux de la même.
|
||||
SCRIPTS = ("deep_proxmox.py", "deep_qemu.py", "install_nixos.py")
|
||||
SCRIPTS = (
|
||||
"deep_proxmox.py",
|
||||
"deep_qemu.py",
|
||||
"install_nixos.py",
|
||||
"qemu_cache.py",
|
||||
)
|
||||
|
||||
# L'autorité du cache de téléchargement, telle que le service l'écrit sur
|
||||
# l'hôte. Une seule définition pour tout long_test/ : une seconde dériverait
|
||||
# en silence, et un test qui pose la mauvaise autorité échoue comme s'il n'en
|
||||
# posait aucune.
|
||||
CACHE_CA = "/var/lib/erplibre_go_qemu_cache/ca.crt"
|
||||
|
||||
|
||||
def drapeaux_cache():
|
||||
"""Les drapeaux de cache à passer à la CLI QEMU pour une VM locale.
|
||||
|
||||
Le détournement est TRANSPARENT et vaut pour tout le pont libvirt : une
|
||||
VM créée ici est interceptée qu'elle le demande ou non. Ne rien passer ne
|
||||
la laisse donc pas en direct — elle reçoit un certificat qu'elle ne
|
||||
reconnaît pas, et chaque téléchargement HTTPS échoue sur « self-signed
|
||||
certificate in certificate chain » : l'image, puis le gestionnaire de
|
||||
paquets, puis tout le reste de l'étage.
|
||||
|
||||
Deux issues, et elles se valent pour le test : approuver l'autorité quand
|
||||
l'hôte en porte une, ou demander une exception par adresse MAC sinon. Sur
|
||||
un hôte sans cache, l'exception ne fait rien et le dit — la CLI n'exige
|
||||
pas d'installer un cache pour pouvoir s'en passer.
|
||||
"""
|
||||
if os.path.isfile(CACHE_CA):
|
||||
return ["--cache-ca", CACHE_CA]
|
||||
return ["--cache-bypass"]
|
||||
|
||||
|
||||
# Une étape bloquée ne doit pas bloquer le test : chaque appel est borné, et le
|
||||
# journal dit lequel a expiré. Généreux, parce que chaque étage est plus lent
|
||||
|
|
@ -458,6 +490,7 @@ class Descente:
|
|||
pub = cle_publique()
|
||||
if pub:
|
||||
argv += ["--ssh-key", pub]
|
||||
argv += drapeaux_cache()
|
||||
if self.dry_run:
|
||||
print(" " + " ".join(shlex.quote(a) for a in argv))
|
||||
return nom
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ sys.path.insert(0, os.path.join(RACINE, "long_test"))
|
|||
sys.path.insert(0, RACINE)
|
||||
|
||||
from descente import ( # noqa: E402
|
||||
CACHE_CA,
|
||||
Descente,
|
||||
cle_publique,
|
||||
detruire_etage1,
|
||||
|
|
@ -95,7 +96,7 @@ DISTRO = "arch"
|
|||
VERSION = "latest"
|
||||
|
||||
CLI = os.path.join(RACINE, "script/qemu/deploy_qemu.py")
|
||||
CA = "/var/lib/erplibre_go_qemu_cache/ca.crt"
|
||||
CA = CACHE_CA
|
||||
CACHE_BIN = "/usr/local/bin/erplibre_go_qemu_cache"
|
||||
SERVICE = "erplibre-go-qemu-cache.service"
|
||||
CONF = "/etc/erplibre_go_qemu_cache/env"
|
||||
|
|
|
|||
|
|
@ -958,17 +958,35 @@ def image_fetch_cmd(
|
|||
continuer reviendrait à installer un système que personne n'a regardé.
|
||||
"""
|
||||
cible = f"{repertoire}/{nom}"
|
||||
partiel = f"{cible}.partiel"
|
||||
somme = lambda f: ( # noqa: E731 - une expression, pas une fonction
|
||||
f"echo {shlex.quote(f'{sha256} {f}')} | sha256sum -c -"
|
||||
)
|
||||
# Le téléchargement va dans un nom PROVISOIRE, et n'est renommé qu'une
|
||||
# fois complet. « wget -O » écrivait dans la cible : une coupure —
|
||||
# réseau, disque plein, Ctrl-C — y figeait une image tronquée que le
|
||||
# « [ -s ] » ci-dessous acceptait à chaque déploiement suivant. Avec une
|
||||
# somme, elle échouait pour toujours sans dire quoi effacer ; sans somme,
|
||||
# elle servait à créer une VM.
|
||||
recuperation = f"wget -nv -O {shlex.quote(partiel)} {shlex.quote(url)}"
|
||||
if sha256:
|
||||
# Vérifiée AVANT d'être mise en place : une image fausse ne devient
|
||||
# jamais celle que le prochain déploiement trouvera « déjà présente ».
|
||||
recuperation += f" && {somme(partiel)}"
|
||||
recuperation += f" && mv {shlex.quote(partiel)} {shlex.quote(cible)}"
|
||||
|
||||
cmd = (
|
||||
f"mkdir -p {shlex.quote(repertoire)} && "
|
||||
f"if [ -s {shlex.quote(cible)} ]; then "
|
||||
f'echo "image déjà présente : {cible}"; else '
|
||||
f"wget -nv -O {shlex.quote(cible)} {shlex.quote(url)}; "
|
||||
f"{recuperation}; "
|
||||
f"fi"
|
||||
)
|
||||
if sha256:
|
||||
cmd += (
|
||||
f" && echo {shlex.quote(f'{sha256} {cible}')}" " | sha256sum -c -"
|
||||
)
|
||||
# Et la cible elle-même, fraîche ou déjà en cache : le cas visé est un
|
||||
# fichier substitué entre deux déploiements, qu'aucun test de présence
|
||||
# ne voit.
|
||||
cmd += f" && {somme(cible)}"
|
||||
return cmd
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -2799,6 +2799,32 @@ OFFLINE_BOOTCMD = [
|
|||
|
||||
CACHE_CERT_NAME = "erplibre-cache.crt"
|
||||
|
||||
# Les familles à qui AUCUN des trois gestes de CACHE_TRUST ne s'applique.
|
||||
#
|
||||
# « nix » en est une : NixOS n'a pas de dossier d'ancres inscriptible — /usr
|
||||
# n'y existe qu'en lecture seule et /etc est produit par le système —, aucune
|
||||
# commande de mise à jour du magasin, et son faisceau est un lien du store que
|
||||
# seule une reconstruction change. Y recopier une ligne calquée sur apt
|
||||
# rendrait le contrôle vert sans rien poser : la VM resterait interceptée et
|
||||
# sans autorité, c'est-à-dire muette sur chaque téléchargement HTTPS.
|
||||
#
|
||||
# Le détournement, lui, porte sur tout le pont. Une VM de ces familles doit
|
||||
# donc en être SOUSTRAITE, faute de quoi elle ne télécharge plus rien — et le
|
||||
# message qu'elle rendrait, « self-signed certificate in certificate chain »,
|
||||
# ne dit rien d'une famille sans magasin.
|
||||
CACHE_SANS_AUTORITE = frozenset({"nix"})
|
||||
|
||||
|
||||
def cache_sans_autorite(distro: str) -> bool:
|
||||
"""La VM de `distro` peut-elle recevoir l'autorité du cache ?
|
||||
|
||||
Non quand sa famille n'a pas de magasin de certificats où l'écrire. Le
|
||||
déploiement l'exempte alors du détournement plutôt que de la laisser
|
||||
échouer sur un certificat qu'elle ne peut pas apprendre.
|
||||
"""
|
||||
famille = cache_family(distro)
|
||||
return bool(famille) and famille in CACHE_SANS_AUTORITE
|
||||
|
||||
|
||||
def cache_family(distro: str) -> str:
|
||||
"""Famille de gestionnaire de paquets d'un système du catalogue.
|
||||
|
|
@ -5237,6 +5263,18 @@ def main() -> None:
|
|||
print(f"\n== 5/5 virt-install (--osinfo {resolved_osinfo}) ==")
|
||||
ensure_network(network_name(args.network), runner)
|
||||
# Avant la création, et non après : la VM télécharge dès cloud-init.
|
||||
# Une famille sans magasin de certificats ne peut pas apprendre
|
||||
# l'autorité du cache : on l'en SOUSTRAIT plutôt que de la laisser buter
|
||||
# sur un certificat inconnu à chaque téléchargement. Décidé ici et non
|
||||
# demandé à l'opérateur — c'est le catalogue qui sait.
|
||||
if cache_sans_autorite(getattr(args, "distro", "")) and not getattr(
|
||||
args, "cache_bypass", False
|
||||
):
|
||||
args.cache_bypass = True
|
||||
print(
|
||||
f"\n {args.distro} n'a pas de magasin de certificats :"
|
||||
" la VM est soustraite au cache et téléchargera en direct."
|
||||
)
|
||||
cache_bypass_apply(args, runner)
|
||||
virt_install(args, disk, seed, resolved_osinfo, runner, installer)
|
||||
if installer:
|
||||
|
|
|
|||
|
|
@ -1760,10 +1760,22 @@ class ProxmoxMenuMixin:
|
|||
)
|
||||
fichiers = mod.cache_files(args)
|
||||
if not fichiers:
|
||||
# Distribution hors table, ou autorité illisible : la VM
|
||||
# télécharge en direct, ce qui marche tant qu'aucune règle ne la
|
||||
# vise. Poser le fichier au mauvais endroit ne marcherait pas et
|
||||
# ne dirait rien.
|
||||
# Une distribution dont le magasin de confiance n'a pas de forme
|
||||
# par fichier — un système déclaratif n'a pas d'ancre où écrire —
|
||||
# ou une autorité illisible. Poser le fichier au mauvais endroit
|
||||
# ne marcherait pas.
|
||||
#
|
||||
# Mais le silence était faux : on n'arrive ici que lorsque l'hôte
|
||||
# Proxmox est lui-même une VM de CE pont, donc ses invités sont
|
||||
# détournés. Celui-ci n'aura rien pour reconnaître le cache, et
|
||||
# chaque téléchargement HTTPS échouera sur « self-signed
|
||||
# certificate in certificate chain » — plus tard, dans la VM,
|
||||
# loin d'ici. C'est pourquoi il est nommé.
|
||||
cle = (
|
||||
"no trust store for this distribution, its downloads "
|
||||
"will fail"
|
||||
)
|
||||
print(f" ⚠ {t(cle)} : {vm.get('distro') or '?'}")
|
||||
return False
|
||||
morceaux = []
|
||||
for chemin, mode, contenu, _proprio in fichiers:
|
||||
|
|
|
|||
|
|
@ -4530,10 +4530,6 @@ TRANSLATIONS = {
|
|||
"fr": "Partir d'une machine NixOS qu'on possède déjà",
|
||||
"en": "Use a NixOS machine you already have",
|
||||
},
|
||||
"Choice (1-2, default 1): ": {
|
||||
"fr": "Choix (1-2, défaut 1) : ",
|
||||
"en": "Choice (1-2, default 1): ",
|
||||
},
|
||||
"Nested QEMU depth: run it": {
|
||||
"fr": "Profondeur QEMU imbriqué : le lancer",
|
||||
"en": "Nested QEMU depth: run it",
|
||||
|
|
@ -7272,6 +7268,14 @@ TRANSLATIONS = {
|
|||
"fr": "autorité du cache de téléchargement non posée",
|
||||
"en": "download cache authority not installed",
|
||||
},
|
||||
"no trust store for this distribution, its downloads "
|
||||
"will fail": {
|
||||
"fr": (
|
||||
"pas de magasin de confiance pour cette distribution, ses "
|
||||
"téléchargements échoueront"
|
||||
),
|
||||
"en": "no trust store for this distribution, its downloads will fail",
|
||||
},
|
||||
"Timezone": {
|
||||
"fr": "Fuseau horaire",
|
||||
"en": "Timezone",
|
||||
|
|
|
|||
|
|
@ -2552,5 +2552,74 @@ class TestLAutoriteDUneVmProxmoxHorsLigne(unittest.TestCase):
|
|||
self.assertNotIn("NPM_CONFIG_AUDIT", rendus[False])
|
||||
|
||||
|
||||
class TestUnInviteQueLAutoriteNAtteintPas(unittest.TestCase):
|
||||
"""Une distribution dont le magasin de confiance n'a pas de forme par
|
||||
fichier ne reçoit rien à poser.
|
||||
|
||||
On n'arrive ici que lorsque l'hôte Proxmox est lui-même une VM de ce pont
|
||||
— c'est la condition de _pve_cache_ca. Son invité est donc détourné par
|
||||
le cache, et sans autorité chaque téléchargement HTTPS échoue sur
|
||||
« self-signed certificate in certificate chain ». Le taire déplace la
|
||||
panne dans la VM, des minutes plus tard et sans sa cause.
|
||||
"""
|
||||
|
||||
def _todo(self):
|
||||
import sys
|
||||
|
||||
sys.argv = ["todo.py"]
|
||||
from script.todo.todo import TODO
|
||||
|
||||
return TODO.__new__(TODO)
|
||||
|
||||
class _Sans:
|
||||
@staticmethod
|
||||
def cache_files(args):
|
||||
return []
|
||||
|
||||
class _Avec:
|
||||
@staticmethod
|
||||
def cache_files(args):
|
||||
return [
|
||||
("/usr/local/share/ca-certificates/x.crt", "0644", "PEM", "")
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def cache_commands(args):
|
||||
return ["update-ca-certificates"]
|
||||
|
||||
def _poser(self, mod, distro):
|
||||
import contextlib
|
||||
import io
|
||||
|
||||
todo = self._todo()
|
||||
vu = {"ssh": []}
|
||||
todo._qemu_import_module = lambda: mod
|
||||
todo._pve_ssh = lambda cible, cmd, timeout=120: (
|
||||
vu["ssh"].append(cmd) or (0, "")
|
||||
)
|
||||
with contextlib.redirect_stdout(io.StringIO()) as sortie:
|
||||
vu["rendu"] = todo._pve_set_cache_ca(
|
||||
"pve+vm-a", {"distro": distro}, "/var/lib/cache/ca.crt"
|
||||
)
|
||||
vu["ecrit"] = sortie.getvalue()
|
||||
return vu
|
||||
|
||||
def test_la_distribution_sans_magasin_est_nommee(self):
|
||||
vu = self._poser(self._Sans, "nixos")
|
||||
self.assertFalse(vu["rendu"])
|
||||
self.assertEqual([], vu["ssh"], "ssh lancé pour rien")
|
||||
self.assertIn("⚠", vu["ecrit"])
|
||||
self.assertIn("nixos", vu["ecrit"])
|
||||
|
||||
def test_celle_qui_en_a_un_le_pose_sans_avertir(self):
|
||||
"""L'avertissement ne doit pas se banaliser : le cas ordinaire sort
|
||||
un « ✓ » et rien d'autre."""
|
||||
vu = self._poser(self._Avec, "debian")
|
||||
self.assertTrue(vu["rendu"])
|
||||
self.assertEqual(1, len(vu["ssh"]))
|
||||
self.assertIn("update-ca-certificates", vu["ssh"][0])
|
||||
self.assertNotIn("⚠", vu["ecrit"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
|
|
|
|||
|
|
@ -464,17 +464,50 @@ class TestAucunSystemeNestOublie(unittest.TestCase):
|
|||
)
|
||||
|
||||
def test_chaque_famille_sait_poser_lautorite(self):
|
||||
from script.qemu.deploy_qemu import CACHE_TRUST, DISTROS, cache_family
|
||||
"""Chaque famille du catalogue sait poser l'autorité, OU est
|
||||
explicitement soustraite au détournement.
|
||||
|
||||
manquantes = sorted(
|
||||
{cache_family(d) for d in DISTROS} - set(CACHE_TRUST)
|
||||
Le détournement porte sur tout le pont : une famille qui ne sait pas
|
||||
apprendre l'autorité et qu'on laisse passer n'échoue pas à
|
||||
l'installation du certificat — elle échoue sur CHAQUE téléchargement,
|
||||
avec un message qui ne dit rien de la cause.
|
||||
"""
|
||||
from script.qemu.deploy_qemu import (
|
||||
CACHE_SANS_AUTORITE,
|
||||
CACHE_TRUST,
|
||||
DISTROS,
|
||||
cache_family,
|
||||
)
|
||||
|
||||
familles = {cache_family(d) for d in DISTROS} - {""}
|
||||
manquantes = sorted(familles - set(CACHE_TRUST) - CACHE_SANS_AUTORITE)
|
||||
self.assertEqual(
|
||||
manquantes,
|
||||
[],
|
||||
f"familles sans commande de confiance : {manquantes}",
|
||||
)
|
||||
|
||||
def test_une_famille_exemptee_nest_pas_aussi_dans_la_table(self):
|
||||
"""Les deux listes s'excluent : une famille qui sait poser l'autorité
|
||||
n'a rien à faire parmi les exemptées, et l'y laisser soustrairait au
|
||||
cache une VM qui pouvait parfaitement en vivre."""
|
||||
from script.qemu.deploy_qemu import CACHE_SANS_AUTORITE, CACHE_TRUST
|
||||
|
||||
self.assertEqual(set(), set(CACHE_TRUST) & CACHE_SANS_AUTORITE)
|
||||
|
||||
def test_une_exemptee_est_bien_soustraite_au_detournement(self):
|
||||
"""L'exemption ne vaut que si le déploiement la POSE : la déclarer et
|
||||
laisser la VM sur le pont ne change rien à son sort."""
|
||||
from script.qemu.deploy_qemu import cache_sans_autorite
|
||||
|
||||
self.assertTrue(cache_sans_autorite("nixos"))
|
||||
for connue in ("debian", "ubuntu", "fedora", "arch", "opensuse"):
|
||||
with self.subTest(distro=connue):
|
||||
self.assertFalse(cache_sans_autorite(connue))
|
||||
# Un système hors catalogue n'est pas « exempté » : il n'a pas de
|
||||
# famille, et c'est une autre question que celle-ci.
|
||||
self.assertFalse(cache_sans_autorite("inconnue"))
|
||||
|
||||
def test_la_famille_vient_du_catalogue_et_nest_pas_recopiee(self):
|
||||
"""Deux tables qui disent la même chose dérivent : c'est ce qui a
|
||||
laissé Proxmox de côté."""
|
||||
|
|
|
|||
|
|
@ -270,6 +270,36 @@ class LesDeuxCheminsLisentLaMemeSomme(unittest.TestCase):
|
|||
# et non se contenter d'un avertissement dans le journal.
|
||||
self.assertIn("&& echo", cmd)
|
||||
|
||||
def test_the_download_lands_on_its_final_name_only_when_complete(self):
|
||||
"""« wget -O » écrivait dans la CIBLE : une coupure — réseau, disque
|
||||
plein, Ctrl-C — y figeait une image tronquée que le test de présence
|
||||
acceptait à chaque déploiement suivant. Avec une somme, elle échouait
|
||||
pour toujours sans dire quoi effacer ; sans somme, elle servait à
|
||||
créer une VM.
|
||||
|
||||
Mesuré : un « .partiel » laissé par une coupure est repris et la
|
||||
cible finit identique à l'amont."""
|
||||
from script.proxmox import proxmox_deploy as pve
|
||||
|
||||
cmd = pve.image_fetch_cmd("http://x/i.qcow2", "i.qcow2")
|
||||
self.assertIn("-O", cmd)
|
||||
self.assertIn(".partiel", cmd)
|
||||
# Le « mv » suit le téléchargement, et par « && » : un wget en échec
|
||||
# ne doit RIEN mettre en place.
|
||||
self.assertLess(cmd.index("wget"), cmd.index("mv "))
|
||||
self.assertIn("&& mv ", cmd)
|
||||
|
||||
def test_a_fresh_download_is_checked_before_it_is_installed(self):
|
||||
"""Une image fausse ne doit jamais devenir celle que le prochain
|
||||
déploiement trouvera « déjà présente »."""
|
||||
from script.proxmox import proxmox_deploy as pve
|
||||
|
||||
cmd = pve.image_fetch_cmd(
|
||||
"http://x/i.qcow2", "i.qcow2", sha256="c" * 64
|
||||
)
|
||||
i_somme = cmd.index("sha256sum")
|
||||
self.assertLess(i_somme, cmd.index("mv "))
|
||||
|
||||
def test_a_cached_image_is_verified_too(self):
|
||||
"""Le cas qu'on veut prendre est un fichier substitué ou tronqué
|
||||
entre deux déploiements : le test de présence ne regarde que la
|
||||
|
|
@ -279,7 +309,11 @@ class LesDeuxCheminsLisentLaMemeSomme(unittest.TestCase):
|
|||
cmd = pve.image_fetch_cmd(
|
||||
"http://x/i.qcow2", "i.qcow2", sha256="b" * 64
|
||||
)
|
||||
self.assertLess(cmd.index("fi"), cmd.index("sha256sum"))
|
||||
# La DERNIÈRE vérification, celle qui porte sur le fichier en
|
||||
# place : la première garde le téléchargement frais avant de
|
||||
# l'installer, et se trouve donc AVANT le « fi ».
|
||||
self.assertLess(cmd.index("fi"), cmd.rindex("sha256sum"))
|
||||
self.assertEqual(2, cmd.count("sha256sum"))
|
||||
|
||||
def test_nothing_is_appended_without_a_sum(self):
|
||||
from script.proxmox import proxmox_deploy as pve
|
||||
|
|
|
|||
|
|
@ -22,7 +22,10 @@ configuration ENTIÈRE, sans message : la VM démarre sans compte ni clé.
|
|||
"""
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -47,10 +50,11 @@ def _deploy_qemu():
|
|||
DQ = _deploy_qemu()
|
||||
|
||||
|
||||
def _cloud_config(distro):
|
||||
args = DQ.build_parser().parse_args(
|
||||
["--distro", distro, "--hostname", "x"]
|
||||
)
|
||||
def _cloud_config(distro, cache_ca=""):
|
||||
argv = ["--distro", distro, "--hostname", "x"]
|
||||
if cache_ca:
|
||||
argv += ["--cache-ca", cache_ca]
|
||||
args = DQ.build_parser().parse_args(argv)
|
||||
return DQ.build_cloud_config(args, None, ["ssh-ed25519 AAAA essai"])
|
||||
|
||||
|
||||
|
|
@ -149,5 +153,85 @@ class LeSeedEntier(unittest.TestCase):
|
|||
self.assertNotIn("/etc/default/keyboard", chemins)
|
||||
|
||||
|
||||
class LAutoriteDuCacheDansLeSeedAssemble(unittest.TestCase):
|
||||
"""Le clavier et l'autorité du cache arrivent par la MÊME clé.
|
||||
|
||||
Trois écritures se disputent « write_files: » — le guide, l'autorité du
|
||||
cache, le clavier — et cloud-init n'en lit qu'UNE : un document qui porte
|
||||
la clé deux fois garde la dernière, sans erreur ni message. Ce qui est
|
||||
perdu ne se voit alors qu'à l'usage, chez l'invité, et seulement pour ce
|
||||
qui en dépendait.
|
||||
|
||||
Les tests plus haut regardent les fonctions une à une ; ceux-ci regardent
|
||||
le document ASSEMBLÉ, seul endroit où cette collision existe.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
if yaml is None:
|
||||
self.skipTest("PyYAML absent")
|
||||
dossier = tempfile.mkdtemp(prefix="cache-ca-")
|
||||
self.addCleanup(shutil.rmtree, dossier, ignore_errors=True)
|
||||
# Une autorité INVENTÉE : cache_files exige « BEGIN CERTIFICATE » et
|
||||
# ne lit rien d'autre du fichier. Reprendre celle d'un hôte réel
|
||||
# figerait dans le dépôt le certificat d'une machine.
|
||||
self.ca = os.path.join(dossier, "ca.crt")
|
||||
with open(self.ca, "w", encoding="utf-8") as fh:
|
||||
fh.write(
|
||||
"-----BEGIN CERTIFICATE-----\nZXNzYWk=\n"
|
||||
"-----END CERTIFICATE-----\n"
|
||||
)
|
||||
|
||||
def _chemins(self, distro):
|
||||
doc = yaml.safe_load(_cloud_config(distro, self.ca))
|
||||
return [f["path"] for f in doc.get("write_files", [])]
|
||||
|
||||
def test_the_key_appears_exactly_once(self):
|
||||
"""Deux « write_files: » dans le même document : PyYAML garde le
|
||||
second et jette le premier, sans rien dire."""
|
||||
for distro in ("debian", "ubuntu", "fedora", "arch", "nixos"):
|
||||
with self.subTest(distro=distro):
|
||||
texte = _cloud_config(distro, self.ca)
|
||||
lignes = texte.splitlines()
|
||||
self.assertEqual(
|
||||
1, lignes.count("write_files:"), "\n".join(lignes[:5])
|
||||
)
|
||||
|
||||
def test_the_authority_reaches_the_assembled_document(self):
|
||||
"""Ce qu'aucun test ne gardait : l'autorité peut disparaître du
|
||||
document tout en restant correcte dans cache_files."""
|
||||
for distro in ("debian", "ubuntu", "fedora", "arch", "opensuse"):
|
||||
with self.subTest(distro=distro):
|
||||
chemins = self._chemins(distro)
|
||||
self.assertTrue(
|
||||
any(c.endswith(DQ.CACHE_CERT_NAME) for c in chemins),
|
||||
chemins,
|
||||
)
|
||||
|
||||
def test_debian_keeps_both_the_keyboard_and_the_authority(self):
|
||||
"""La seule distribution qui demande les deux par write_files : si
|
||||
une écriture en écrase une autre, c'est ici que cela se voit."""
|
||||
chemins = self._chemins("debian")
|
||||
self.assertIn("/etc/default/keyboard", chemins)
|
||||
self.assertTrue(any(c.endswith(DQ.CACHE_CERT_NAME) for c in chemins))
|
||||
|
||||
def test_the_trust_command_reaches_runcmd(self):
|
||||
"""Le fichier posé sans la commande qui relit le magasin ne sert à
|
||||
rien : les deux moitiés voyagent séparément."""
|
||||
doc = yaml.safe_load(_cloud_config("debian", self.ca))
|
||||
self.assertTrue(
|
||||
any("update-ca-certificates" in str(c) for c in doc["runcmd"]),
|
||||
doc["runcmd"],
|
||||
)
|
||||
|
||||
def test_a_system_with_no_anchor_gets_nothing_and_keeps_the_rest(self):
|
||||
"""Un système déclaratif n'a pas d'ancre où écrire. Il ne reçoit donc
|
||||
pas l'autorité — et le reste du document lui parvient quand même."""
|
||||
doc = yaml.safe_load(_cloud_config("nixos", self.ca))
|
||||
chemins = [f["path"] for f in doc.get("write_files", [])]
|
||||
self.assertFalse([c for c in chemins if DQ.CACHE_CERT_NAME in c])
|
||||
self.assertIn("users", doc)
|
||||
self.assertEqual("ca", doc["keyboard"]["layout"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -1176,20 +1176,19 @@ class TestLeMenuDesDeuxTests(unittest.TestCase):
|
|||
src.index("--detruire --dry-run"), src.index('"--detruire"')
|
||||
)
|
||||
|
||||
def test_everything_the_lock_knows_can_be_undone(self):
|
||||
def test_the_two_copies_of_the_list_agree(self):
|
||||
"""Le verrou (long_test/descente.py) et le défaire (le menu) portent
|
||||
chacun la liste des tests longs, faute de pouvoir la partager : le
|
||||
menu lance ces scripts en sous-processus et n'importe jamais
|
||||
long_test/, qui traîne avec lui le module Proxmox.
|
||||
|
||||
L'inclusion, et non l'égalité : un test long que le verrou connaît
|
||||
crée des machines, donc il DOIT être défaisable. L'inverse n'est pas
|
||||
vrai — le menu défait aussi le test du cache, que le verrou ne
|
||||
surveille pas. Exiger l'égalité ferait échouer ce test sur un écart
|
||||
qui est un choix, pas une dérive."""
|
||||
L'égalité, dans les deux sens, et chacun de ses sens couvre une
|
||||
panne : un test long absent du VERROU laisse une autre descente
|
||||
détruire ses machines pendant qu'il tourne ; absent du DÉFAIRE, il
|
||||
laisse ses VM derrière lui, sans rien pour les reprendre."""
|
||||
from script.todo.longtest_menu import SCRIPTS_DEFAISABLES
|
||||
|
||||
self.assertTrue(set(moteur.SCRIPTS) <= set(SCRIPTS_DEFAISABLES))
|
||||
self.assertEqual(set(moteur.SCRIPTS), set(SCRIPTS_DEFAISABLES))
|
||||
self.assertIn("install_nixos.py", moteur.SCRIPTS)
|
||||
|
||||
def test_the_host_options_are_built_from_the_host_dict(self):
|
||||
|
|
@ -1767,5 +1766,50 @@ class TestLeMenu(unittest.TestCase):
|
|||
self.assertIn("Long tests", src)
|
||||
|
||||
|
||||
class LeCacheDeLEtage1(unittest.TestCase):
|
||||
"""L'étage 1 est une VM du pont libvirt local, et le cache détourne tout
|
||||
ce pont. Ne rien lui passer ne la laisse pas en direct : elle est
|
||||
interceptée sans autorité, et chaque téléchargement HTTPS de l'étage —
|
||||
l'image, puis le gestionnaire de paquets — échoue sur « self-signed
|
||||
certificate in certificate chain »."""
|
||||
|
||||
def setUp(self):
|
||||
self.dossier = tempfile.mkdtemp(prefix="descente-ca-")
|
||||
self.addCleanup(shutil.rmtree, self.dossier, ignore_errors=True)
|
||||
self.addCleanup(setattr, moteur, "CACHE_CA", moteur.CACHE_CA)
|
||||
|
||||
def test_the_authority_is_trusted_when_the_host_has_one(self):
|
||||
ca = os.path.join(self.dossier, "ca.crt")
|
||||
with open(ca, "w", encoding="utf-8") as fh:
|
||||
fh.write("-----BEGIN CERTIFICATE-----\n")
|
||||
moteur.CACHE_CA = ca
|
||||
self.assertEqual(["--cache-ca", ca], moteur.drapeaux_cache())
|
||||
|
||||
def test_a_host_without_one_asks_for_an_exception(self):
|
||||
"""L'exception par MAC, et non le silence : sur un hôte sans cache
|
||||
elle ne fait rien et le dit, ce qui ne coûte qu'une ligne ; sur un
|
||||
hôte qui en porte un, le silence coûte l'étage."""
|
||||
moteur.CACHE_CA = os.path.join(self.dossier, "absent.crt")
|
||||
self.assertEqual(["--cache-bypass"], moteur.drapeaux_cache())
|
||||
|
||||
def test_the_first_floor_passes_them(self):
|
||||
"""Les deux piles héritent de creer_etage1 : le drapeau posé là les
|
||||
couvre toutes les deux."""
|
||||
import inspect
|
||||
|
||||
src = inspect.getsource(moteur.Descente.creer_etage1)
|
||||
self.assertIn("argv += drapeaux_cache()", src)
|
||||
|
||||
def test_the_long_tests_share_one_authority(self):
|
||||
"""Une seconde définition du chemin dériverait en silence, et poser
|
||||
la mauvaise autorité échoue comme n'en poser aucune."""
|
||||
import sys as _sys
|
||||
|
||||
_sys.path.insert(0, os.path.join(RACINE, "long_test"))
|
||||
import qemu_cache
|
||||
|
||||
self.assertEqual(moteur.CACHE_CA, qemu_cache.CA)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
|
|
|
|||
Loading…
Reference in a new issue