[FIX] déploiement qemu : reprendre une image périmée avant d'abandonner

Une image gardée vieillit : le répertoire « latest » d'une distribution avance
à chaque version mineure, et la somme publiée cesse de décrire celle du
disque. La vérification la supprimait puis sortait en erreur — une campagne
entière, trois VM, perdue pour une péremption qu'un seul téléchargement
répare. Un écart sur une image DÉJÀ présente la fait désormais reprendre une
fois ; un second écart porte sur des octets fraîchement téléchargés, et arrête
tout. Sans liste de miroirs, et pour des sommes injoignables, rien ne change.
Vérifié : quatre tests de comportement, deux mutations attrapées.

--- EN ---

A kept image ages: a distribution's « latest » directory moves with each point
release, and the published sum stops describing the one on disk. The check
deleted it and exited — a whole campaign, three VMs, lost to a staleness one
download repairs. A mismatch on an image ALREADY present now makes it fetched
once more; a second mismatch is on freshly downloaded bytes, and stops
everything. With no mirror list, and for unreachable sums, nothing changes.
Checked: four behaviour tests, two mutations caught.

Assisted-by: Claude Opus 5

script/qemu/deploy_qemu.py

@@ -1776,7 +1776,11 @@ def expected_sum(sums: str, filename: str) -> str:
 
 
 def verify_sha256(
-    url: str, image: Path, dry_run: bool, distro: str = "ubuntu"
+    url: str,
+    image: Path,
+    dry_run: bool,
+    distro: str = "ubuntu",
+    urls: tuple[str, ...] = (),
 ) -> None:
     """Vérifie l'empreinte via les sommes que la distribution publie.
 
@@ -1810,17 +1814,32 @@ def verify_sha256(
         print(f"  ⚠ empreinte absente pour {filename}, image NON vérifiée")
         return
 
-    h = hashlib.new(algo)
+    def empreinte() -> str:
-    with image.open("rb") as fh:
+        h = hashlib.new(algo)
-        for chunk in iter(lambda: fh.read(1 << 20), b""):
+        with image.open("rb") as fh:
-            h.update(chunk)
+            for chunk in iter(lambda: fh.read(1 << 20), b""):
-    if h.hexdigest() != expected:
+                h.update(chunk)
+        return h.hexdigest()
+
+    obtenu = empreinte()
+    # Un écart sur une image GARDÉE ne prouve pas une substitution : le
+    # répertoire « latest » d'une distribution avance à chaque version
+    # mineure, et la somme publiée cesse alors de décrire celle du disque.
+    # Supprimer puis abandonner coûtait toute une campagne pour une simple
+    # péremption. L'image est reprise UNE fois ; un second écart, lui, porte
+    # sur des octets fraîchement téléchargés et arrête tout.
+    if obtenu != expected and urls:
+        print(f"  Somme {algo} NON conforme : image reprise une fois.")
+        image.unlink(missing_ok=True)
+        download_image(list(urls), image, dry_run)
+        obtenu = empreinte()
+    if obtenu != expected:
         image.unlink(
             missing_ok=True
         )  # évite la réutilisation du cache corrompu
         sys.exit(
             f"Somme {algo} NON conforme ! Image supprimée : {image}\n"
-            f"  attendu : {expected}\n  obtenu  : {h.hexdigest()}"
+            f"  attendu : {expected}\n  obtenu  : {obtenu}"
         )
     print(f"  Somme {algo} conforme.")
 
@@ -5731,7 +5750,13 @@ def main() -> None:
             args.distro, args.image_path, args.dry_run, tuple(urls)
         )
         if do_verify:
-            verify_sha256(url, args.image_path, args.dry_run, args.distro)
+            verify_sha256(
+                url,
+                args.image_path,
+                args.dry_run,
+                args.distro,
+                tuple(urls),
+            )
         print("\nTerminé (téléchargement seul).")
         return
 
@@ -5848,7 +5873,13 @@ def main() -> None:
             args.distro, args.image_path, args.dry_run, tuple(urls)
         )
         if do_verify:
-            verify_sha256(url, args.image_path, args.dry_run, args.distro)
+            verify_sha256(
+                url,
+                args.image_path,
+                args.dry_run,
+                args.distro,
+                tuple(urls),
+            )
 
         print(f"\n== 2-3/5 Disque de travail {disk} ({args.disk_size}) ==")
         prepare_disk(args.image_path, disk, args.disk_size, runner, args.force)

test/test_qemu_sommes.py

@@ -160,5 +160,100 @@ class LaVerificationEstLeDefaut(unittest.TestCase):
         self.assertIn("image.unlink", corps)
 
 
+class UneImageGardeeQuiAVieilli(unittest.TestCase):
+    """« latest » avance à chaque version mineure : l'image gardée cesse de
+    correspondre à la somme publiée sans que rien ne soit corrompu.
+
+    La supprimer puis abandonner coûtait la campagne entière — trente minutes
+    et trois VM — là où un seul téléchargement suffit. Un second écart, lui,
+    porte sur des octets neufs : c'est une panne d'intégrité, et elle arrête.
+    """
+
+    NOM = "ubuntu-24.04-server-cloudimg-amd64.img"
+    URL = f"https://miroir.invalid/d/{NOM}"
+
+    def jouer(
+        self,
+        sur_disque,
+        retelecharge=None,
+        publie=b"publie",
+        urls=("https://m/i",),
+    ):
+        """Rend (sorti, nombre de téléchargements, contenu final).
+
+        « publie » est ce que le fichier de sommes DÉCLARE ; « retelecharge »
+        ce qu'un nouveau téléchargement pose vraiment. Les confondre rend le
+        second écart impossible à éprouver : les octets repris concordent
+        alors toujours, et le cas « faux deux fois » n'existe plus.
+        """
+        import hashlib
+        import tempfile
+        from unittest import mock
+
+        pose = retelecharge if retelecharge is not None else publie
+        somme = hashlib.sha256(publie).hexdigest()
+        sums = f"{somme}  {self.NOM}\n".encode()
+
+        class Reponse:
+            def __enter__(self_inner):
+                return self_inner
+
+            def __exit__(self_inner, *a):
+                return False
+
+            def read(self_inner):
+                return sums
+
+        with tempfile.TemporaryDirectory() as d:
+            image = Path(d) / self.NOM
+            image.write_bytes(sur_disque)
+            appels = []
+
+            def faux_telechargement(urls_, dest, dry_run, timeout=None):
+                appels.append(tuple(urls_))
+                Path(dest).write_bytes(pose)
+
+            sorti = False
+            with mock.patch.object(
+                DQ.urllib.request, "urlopen", lambda *a, **k: Reponse()
+            ), mock.patch.object(DQ, "download_image", faux_telechargement):
+                try:
+                    DQ.verify_sha256(self.URL, image, False, "ubuntu", urls)
+                except SystemExit:
+                    sorti = True
+            final = image.read_bytes() if image.exists() else b""
+        return sorti, len(appels), final
+
+    def test_une_image_perimee_est_reprise_une_fois(self):
+        sorti, n, final = self.jouer(b"vieille", b"publie")
+        self.assertFalse(sorti, "une péremption arrête encore le déploiement")
+        self.assertEqual(
+            1, n, "l'image n'a pas été reprise exactement une fois"
+        )
+        self.assertEqual(b"publie", final)
+
+    def test_un_second_ecart_arrete_tout(self):
+        """Des octets fraîchement téléchargés qui ne concordent pas ne sont
+        plus une péremption : la vérification doit alors refuser."""
+        sorti, n, final = self.jouer(b"vieille", b"faux-aussi")
+        self.assertTrue(sorti, "une image fausse deux fois passe")
+        self.assertEqual(1, n, "la reprise boucle au lieu d'arrêter")
+        self.assertEqual(b"", final, "l'image fausse est restée sur le disque")
+
+    def test_sans_miroir_le_comportement_ne_change_pas(self):
+        """Appelée sans liste de miroirs — le mode épinglé, un appelant tiers
+        — la vérification garde sa forme d'avant : supprimer et sortir."""
+        sorti, n, final = self.jouer(b"vieille", b"publie", urls=())
+        self.assertTrue(sorti)
+        self.assertEqual(0, n)
+        self.assertEqual(b"", final)
+
+    def test_une_image_conforme_ne_declenche_rien(self):
+        sorti, n, final = self.jouer(b"publie", b"publie")
+        self.assertFalse(sorti)
+        self.assertEqual(0, n, "une image conforme a été retéléchargée")
+        self.assertEqual(b"publie", final)
+
+
 if __name__ == "__main__":
     unittest.main()