fix(dof): l import par clé API existe enfin — le helper n avait jamais été écrit

opt/pincabos/tests/test_dof_import_api.py

@@ -0,0 +1,149 @@
+"""L'import DOF Config Tool par clé API (PINCABOS_DOF_IMPORT_API_V1).
+
+La page DOF appelait `/usr/local/sbin/pincabos-dof-online-api-import` depuis
+toujours… et ce programme n'a JAMAIS existé : ni dans le dépôt, ni dans
+l'historique git, ni dans le master. La page affichait donc « Import DOF via
+API indisponible » sur tous les cabinets, et seul l'import ZIP manuel
+fonctionnait. Trouvé le 09/09/2026 en cherchant pourquoi un testeur n'avait
+aucune LED en table : sans les `directoutputconfig*.ini` de SON cabinet, DOF
+n'a rien à jouer.
+
+Le contrat vient du client de référence (mkalkbrenner/dof_configtool_client,
+DownloadController.php) :
+
+    GET http://configtool.vpuniverse.com/api.php?query=getconfig&apikey=<clé>
+        -> une archive ZIP de *.ini, *.xml, *.png
+
+Quand la clé est refusée, l'API répond du TEXTE et non un ZIP. C'est ce texte
+qui explique l'erreur : on le montre plutôt qu'un « échec » muet.
+"""
+import io
+import unittest
+import zipfile
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from _charge import RACINE, charger
+
+R = Path(RACINE)
+REL = "usr/local/sbin/pincabos-dof-online-api-import"
+OUTIL = R / REL
+
+
+def _module():
+    # le programme n'a pas d'extension .py : on passe par le chargeur du banc
+    return charger(REL, "dof_import")
+
+
+def _archive(noms) -> bytes:
+    tampon = io.BytesIO()
+    with zipfile.ZipFile(tampon, "w") as z:
+        for n in noms:
+            z.writestr(n, "contenu de " + n)
+    return tampon.getvalue()
+
+
+class LeProgrammeExiste(unittest.TestCase):
+    """C'est tout le sujet : la WebApp l'appelle par ce chemin exact."""
+
+    def test_au_chemin_attendu_par_la_webapp(self):
+        self.assertTrue(OUTIL.is_file())
+        appel = (R / "opt/pincabos/web/pincabos_webapp_dof.py").read_text(encoding="utf-8")
+        self.assertIn("/usr/local/sbin/pincabos-dof-online-api-import", appel)
+
+    def test_executable(self):
+        self.assertTrue(OUTIL.stat().st_mode & 0o111, "doit être exécutable")
+
+    def test_l_url_est_celle_du_client_de_reference(self):
+        s = OUTIL.read_text(encoding="utf-8")
+        self.assertIn("configtool.vpuniverse.com/api.php?query=getconfig&apikey=", s)
+
+
+class LaReponseNEstPasToujoursUneArchive(unittest.TestCase):
+    def setUp(self):
+        self.m = _module()
+
+    def test_une_archive_est_reconnue(self):
+        self.assertTrue(self.m.est_une_archive(_archive(["directoutputconfig30.ini"])))
+
+    def test_du_texte_ne_l_est_pas(self):
+        self.assertFalse(self.m.est_une_archive(b"Invalid API key"))
+
+    def test_le_message_de_l_api_est_rendu_a_l_utilisateur(self):
+        # une clé refusée doit expliquer pourquoi, pas dire « échec »
+        self.m.telecharger = lambda cle: b"Invalid API key"
+        code = self.m.main(["mauvaise-cle"])
+        self.assertEqual(code, 2)
+
+    def test_sans_cle_on_refuse_avant_tout_reseau(self):
+        self.assertEqual(self.m.main([]), 64)
+        self.assertEqual(self.m.main(["   "]), 64)
+
+
+class LExtractionEstFiltree(unittest.TestCase):
+    """Une archive vient du réseau : on n'en extrait jamais un chemin."""
+
+    def setUp(self):
+        self.m = _module()
+
+    def test_seuls_les_fichiers_dof_sont_retenus(self):
+        z = zipfile.ZipFile(io.BytesIO(_archive([
+            "directoutputconfig30.ini", "cabinet.xml", "DirectOutputShapes.png",
+            "lisezmoi.txt", "script.sh"])))
+        with z:
+            noms = [b for _, b in self.m.noms_utiles(z)]
+        self.assertEqual(sorted(noms),
+                         ["DirectOutputShapes.png", "cabinet.xml", "directoutputconfig30.ini"])
+
+    def test_aucune_traversee_de_chemin(self):
+        # « ../../etc/passwd.ini » ne doit jamais sortir du dossier cible
+        z = zipfile.ZipFile(io.BytesIO(_archive([
+            "../../etc/passwd.ini", "sous/dossier/directoutputconfig31.ini"])))
+        with z:
+            noms = [b for _, b in self.m.noms_utiles(z)]
+        self.assertEqual(sorted(noms), ["directoutputconfig31.ini", "passwd.ini"])
+        for n in noms:
+            self.assertNotIn("/", n)
+            self.assertNotIn("..", n)
+
+    def test_pose_dans_le_dossier(self):
+        with TemporaryDirectory() as tmp:
+            cible = Path(tmp) / "directoutputconfig"
+            z = zipfile.ZipFile(io.BytesIO(_archive(["directoutputconfig30.ini", "cabinet.xml"])))
+            with z:
+                bilan = self.m.poser(z, self.m.noms_utiles(z), cible, True)
+            self.assertEqual(bilan["ecrits"], 2)
+            self.assertTrue((cible / "directoutputconfig30.ini").is_file())
+
+    def test_un_fichier_identique_n_est_pas_reecrit(self):
+        with TemporaryDirectory() as tmp:
+            cible = Path(tmp) / "directoutputconfig"
+            donnees = _archive(["cabinet.xml"])
+            for attendu in ({"ecrits": 1, "inchanges": 0}, {"ecrits": 0, "inchanges": 1}):
+                z = zipfile.ZipFile(io.BytesIO(donnees))
+                with z:
+                    bilan = self.m.poser(z, self.m.noms_utiles(z), cible, True)
+                self.assertEqual(bilan["ecrits"], attendu["ecrits"])
+                self.assertEqual(bilan["inchanges"], attendu["inchanges"])
+
+    def test_sans_force_on_ne_remplace_pas(self):
+        with TemporaryDirectory() as tmp:
+            cible = Path(tmp) / "directoutputconfig"
+            cible.mkdir(parents=True)
+            (cible / "cabinet.xml").write_text("le mien", encoding="utf-8")
+            z = zipfile.ZipFile(io.BytesIO(_archive(["cabinet.xml"])))
+            with z:
+                bilan = self.m.poser(z, self.m.noms_utiles(z), cible, False)
+            self.assertEqual(bilan["gardes"], 1)
+            self.assertEqual((cible / "cabinet.xml").read_text(encoding="utf-8"), "le mien")
+
+
+class LesDeuxDossiersSontServis(unittest.TestCase):
+    """VPX (PrefPath) et VPinFE lisent chacun le leur : un seul servi donne un
+    cabinet à moitié muet."""
+
+    def test_le_dossier_canonique_est_toujours_present(self):
+        m = _module()
+        self.assertIn(m.CANONIQUE, m.dossiers_dof() + [m.CANONIQUE])
+        s = OUTIL.read_text(encoding="utf-8")
+        self.assertIn("VPinFE", s, "l'intention doit être écrite noir sur blanc")

usr/local/sbin/pincabos-dof-online-api-import

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""PinCabOS — import DOF Config Tool par clé API (PINCABOS_DOF_IMPORT_API_V1).
+
+    pincabos-dof-online-api-import <clé API> [force|noforce]
+
+La page DOF de la WebApp appelle ce programme depuis toujours… et il n'a
+jamais existé : ni dans le dépôt, ni dans l'historique git, ni dans le master.
+La page affichait donc « Import DOF via API indisponible » sur tous les
+cabinets, et seul l'import ZIP manuel fonctionnait. Trouvé le 09/09/2026 en
+cherchant pourquoi un testeur n'avait aucune LED en table : sans les
+`directoutputconfig*.ini` de SON cabinet, DOF n'a rien à jouer.
+
+Le contrat de l'API vient du client de référence de Michael Kalkbrenner
+(github.com/mkalkbrenner/dof_configtool_client, DownloadController.php) :
+
+    GET http://configtool.vpuniverse.com/api.php?query=getconfig&apikey=<clé>
+        -> une archive ZIP contenant *.ini, *.xml et *.png
+
+Quand la clé est mauvaise, l'API répond du texte et non un ZIP : c'est ce
+texte qui explique l'erreur, on le montre tel quel plutôt qu'un « échec ».
+
+Les fichiers vont dans TOUS les dossiers DOF du cabinet : VPX (PrefPath) et
+VPinFE lisent chacun le leur, et un seul des deux servi donne un cabinet
+à moitié muet.
+"""
+from __future__ import annotations
+
+import glob
+import hashlib
+import shutil
+import ssl
+import sys
+import tempfile
+import urllib.error
+import urllib.parse
+import urllib.request
+import zipfile
+from datetime import datetime
+from pathlib import Path
+
+API = "configtool.vpuniverse.com/api.php?query=getconfig&apikey="
+CABINETS_GLOB = "/home/pinball/.local/share/VPinballX/*/directoutputconfig"
+CANONIQUE = Path("/home/pinball/.pincabos/vpx/directoutputconfig")
+SAUVEGARDES = Path("/opt/pincabos/backups/dof-configtool")
+JOUEUR = 1000
+# Ce que le Config Tool livre, et rien d'autre : on n'extrait jamais un nom
+# venu du réseau sans le filtrer (traversée de chemin).
+EXTENSIONS = (".ini", ".xml", ".png")
+DELAI = 120
+
+
+def dossiers_dof() -> list:
+    """VPX et VPinFE lisent chacun leur dossier : on sert les deux."""
+    trouves = [Path(d) for d in sorted(glob.glob(CABINETS_GLOB))]
+    if CANONIQUE not in trouves:
+        trouves.append(CANONIQUE)
+    return [d for d in trouves if d.parent.exists()] or [CANONIQUE]
+
+
+def telecharger(cle: str) -> bytes:
+    """L'archive, ou une exception qui porte le message de l'API."""
+    erreurs = []
+    for schema in ("https://", "http://"):
+        url = schema + API + urllib.parse.quote(cle, safe="")
+        try:
+            ctx = ssl.create_default_context()
+            with urllib.request.urlopen(url, timeout=DELAI, context=ctx) as r:
+                return r.read()
+        except urllib.error.HTTPError as exc:
+            corps = (exc.read() or b"")[:400].decode("utf-8", "replace").strip()
+            erreurs.append("%s : HTTP %s %s" % (schema.rstrip(":/"), exc.code, corps))
+        except Exception as exc:                       # réseau, TLS, DNS…
+            erreurs.append("%s : %s" % (schema.rstrip(":/"), exc))
+    raise RuntimeError(" ; ".join(erreurs))
+
+
+def est_une_archive(donnees: bytes) -> bool:
+    return donnees[:4] == b"PK\x03\x04"
+
+
+def noms_utiles(zf: zipfile.ZipFile) -> list:
+    """Seuls les fichiers attendus, et jamais de chemin : le nom de base suffit.
+
+    Une archive venue du réseau peut contenir « ../../etc/passwd » ; on ne
+    reconstruit donc aucun répertoire et on refuse tout nom vide ou absolu.
+    """
+    out = []
+    for info in zf.infolist():
+        if info.is_dir():
+            continue
+        base = Path(info.filename).name
+        if not base or base.startswith("."):
+            continue
+        if base.lower().endswith(EXTENSIONS):
+            out.append((info, base))
+    return out
+
+
+def sauvegarder(dossier: Path, horodatage: str) -> Path | None:
+    presents = [f for f in dossier.glob("*") if f.is_file()
+                and f.name.lower().endswith(EXTENSIONS)]
+    if not presents:
+        return None
+    cible = SAUVEGARDES / horodatage / dossier.name
+    cible.mkdir(parents=True, exist_ok=True)
+    for f in presents:
+        shutil.copy2(f, cible / f.name)
+    return cible
+
+
+def poser(zf: zipfile.ZipFile, entrees: list, dossier: Path, ecraser: bool) -> dict:
+    dossier.mkdir(parents=True, exist_ok=True)
+    bilan = {"ecrits": 0, "inchanges": 0, "gardes": 0}
+    for info, base in entrees:
+        contenu = zf.read(info)
+        cible = dossier / base
+        if cible.exists():
+            if hashlib.sha256(cible.read_bytes()).digest() == hashlib.sha256(contenu).digest():
+                bilan["inchanges"] += 1
+                continue
+            if not ecraser:
+                bilan["gardes"] += 1
+                continue
+        cible.write_bytes(contenu)
+        try:
+            shutil.chown(cible, JOUEUR, JOUEUR)
+        except (LookupError, PermissionError, OSError):
+            pass
+        bilan["ecrits"] += 1
+    try:
+        shutil.chown(dossier, JOUEUR, JOUEUR)
+    except (LookupError, PermissionError, OSError):
+        pass
+    return bilan
+
+
+def main(argv=None) -> int:
+    argv = list(sys.argv[1:] if argv is None else argv)
+    if not argv or not argv[0].strip():
+        print("usage: pincabos-dof-online-api-import <clé API> [force|noforce]",
+              file=sys.stderr)
+        return 64
+    cle = argv[0].strip()
+    ecraser = (argv[1].strip().lower() == "force") if len(argv) > 1 else True
+
+    print("Téléchargement depuis le DOF Config Tool…")
+    try:
+        donnees = telecharger(cle)
+    except RuntimeError as exc:
+        print("NOGO [X] Téléchargement impossible : %s" % exc, file=sys.stderr)
+        return 1
+
+    if not est_une_archive(donnees):
+        # L'API répond du texte quand la clé est refusée : c'est LUI le message
+        # utile. Le montrer vaut mieux qu'un « échec » sans explication.
+        message = donnees[:400].decode("utf-8", "replace").strip() or "(réponse vide)"
+        print("NOGO [X] Le Config Tool n'a pas renvoyé d'archive.", file=sys.stderr)
+        print("         Sa réponse : %s" % message, file=sys.stderr)
+        print("         Vérifiez la clé API sur configtool.vpuniverse.com.", file=sys.stderr)
+        return 2
+
+    horodatage = datetime.now().strftime("%Y%m%d-%H%M%S")
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = Path(tmp) / "config.zip"
+        archive.write_bytes(donnees)
+        try:
+            zf = zipfile.ZipFile(archive)
+        except zipfile.BadZipFile as exc:
+            print("NOGO [X] Archive illisible : %s" % exc, file=sys.stderr)
+            return 2
+        with zf:
+            entrees = noms_utiles(zf)
+            if not entrees:
+                print("NOGO [X] L'archive ne contient aucun fichier DOF "
+                      "(*.ini, *.xml, *.png).", file=sys.stderr)
+                return 2
+            inis = sum(1 for _, b in entrees if b.lower().endswith(".ini"))
+            print("Archive reçue : %d fichiers, dont %d directoutputconfig."
+                  % (len(entrees), inis))
+
+            total = {"ecrits": 0, "inchanges": 0, "gardes": 0}
+            for dossier in dossiers_dof():
+                sauvegarde = sauvegarder(dossier, horodatage)
+                bilan = poser(zf, entrees, dossier, ecraser)
+                for k in total:
+                    total[k] += bilan[k]
+                print("  %s : %d écrits, %d inchangés, %d gardés%s"
+                      % (dossier, bilan["ecrits"], bilan["inchanges"], bilan["gardes"],
+                         " (sauvegarde : %s)" % sauvegarde if sauvegarde else ""))
+
+    if total["gardes"] and not ecraser:
+        print("INFO [=] %d fichiers différents ont été gardés : relancez avec "
+              "« force » pour les remplacer." % total["gardes"])
+    print("GO [OK] Import DOF Config Tool terminé (%d fichiers écrits)." % total["ecrits"])
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())