diff --git a/script/qemu/deploy_qemu.py b/script/qemu/deploy_qemu.py index c7d13fc..46ab3e1 100755 --- a/script/qemu/deploy_qemu.py +++ b/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) - with image.open("rb") as fh: - for chunk in iter(lambda: fh.read(1 << 20), b""): - h.update(chunk) - if h.hexdigest() != expected: + def empreinte() -> str: + h = hashlib.new(algo) + with image.open("rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + 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) diff --git a/test/test_qemu_sommes.py b/test/test_qemu_sommes.py index 05e4b92..b3e3176 100644 --- a/test/test_qemu_sommes.py +++ b/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()