Feature/progress bar

main.py

@@ -1,8 +1,11 @@
+import sys
+import time
 from src.detecteur_crypto import DetecteurCryptoOrchestrateur, ResultatAnalyse
 from src.analyzers.blowfish_analyzer import Blowfish_Analyzer
 from src.analyzers.aes_cbc_analyzer import Aes_Cbc_Analyzer
 from src.interface_console import consoleInterface
 import os
+from rich.progress import track
 # print(DetecteurCryptoOrchestrateur().analyser_fichier_specifique('data/mission1.enc'))
 
 # try:
@@ -14,4 +17,5 @@
 #     print("Erreur: Le fichier 'mission3.enc' est introuvable.")
 
 consoleInterface()
-# print(DetecteurCryptoOrchestrateur().mission_complete_automatique('data/', 'keys/wordlist.txt'))
+# print(DetecteurCryptoOrchestrateur().mission_complete_automatique('data/', 'keys/wordlist.txt'))
+

requirements.txt

@@ -2,4 +2,6 @@ cffi==1.17.1
 cryptography==43.0.0
 pycparser==2.22
 pycryptodome==3.23.0
-rich==14.1.0
+rich==14.1.0
+tqdm==4.67.1
+colorama==0.4.6

src/detecteur_crypto.py

@@ -11,10 +11,11 @@
 from src.analyzers.aes_gcm_analyzer import Aes_Gcm_Analyzer
 from src.analyzers.fernet_analyzer import FernetAnalyzer
 from src.rapport_mission import rapport_mission
-
 # Import des modules utilitaries
 from src.utils import verifier_texte_dechiffre
-
+from rich.progress import Progress
+from rich.markdown import Markdown
+from rich.console import Console
 class ResultatAnalyse:
     """
         Classe représentant un résultat d'analyse.
@@ -36,6 +37,9 @@ class DetecteurCryptoOrchestrateur:
                 -Lance et coordonnes le processus de dechiffrement 
     """
 
+    _NBR_OPERATION_MISSION = 4 
+    _NBR_OPERATION_ANALYSE = 3
+
     def __init__(self):
         """
         Initialisation de tous les modules d'analyse disponibles 
@@ -55,7 +59,7 @@ def __init__(self):
             "tentatives_total": 0
         }
 
-    def analyser_fichier_specifique(self, chemin_fichier_chiffre: str) -> ResultatAnalyse:
+    def analyser_fichier_specifique(self, chemin_fichier_chiffre: str, progress : Progress, task, error:bool, nbr_opr_mission: int) -> ResultatAnalyse:
         """
         ANALYSE D'UN FICHIER SPÉCIFIQUE
         - Sélection du fichier à analyser
@@ -64,21 +68,32 @@ def analyser_fichier_specifique(self, chemin_fichier_chiffre: str) -> ResultatAn
 
         Args:
             chemin_fichier_chiffre(str): chemin du fichier chiffré à analyser
-
+            progress (Progress) : la progress bar à mettre à jour
+            error(bool): nécessaire pour déterminer les erreurs et définir le message de final de la progress bar 
         Returns:
             ResultatAnalyse: résultat de l'analyse
         """
         debut_analyse = time.time()
 
         try:
             # Vérification de l'existence du fichier
-            time.sleep(0.3) # TODO : Intégrer la progress bar -> step : Verification du chemin de fichier fourni
+            avance = (100/(self._NBR_OPERATION_ANALYSE * nbr_opr_mission))
+            time.sleep(0.3) # Done : Intégrer la progress bar -> step : Verification du chemin de fichier fourni
+            progress.update(task_id=task, description="Verification du chemin de fichier fourni", advance=avance * 0.3)
+            time.sleep(1)
+
             if not os.path.isfile(Path('data')/f"{chemin_fichier_chiffre}"):
-                print("Erreur: Fichier non trouvé")
+                time.sleep(0.3) # TODO : Intégrer la progress bar -> step : Verification du chemin de fichier fourni
+                progress.update(task_id=task, description="Fichier non trouvé ❌ (Aborting...)", advance=((avance * self._NBR_OPERATION_ANALYSE) - avance * 0.3) )
+                time.sleep(1)
+                error = True
                 return ResultatAnalyse("", b"", 0.0, b"", 0.0, 0)
 
             # Initialisation des variables
             time.sleep(0.5) # TODO : Mise à jour de la progress bar -> step : Initialisation des utilitaires pour l'identification
+            progress.update(task_id=task, description="Initialisation des utilitaires pour l'identification", advance=avance*0.2)
+            time.sleep(1)
+
             algorithme_detecte = ""
             cle = b""
             score_probabilite = 0.0
@@ -87,22 +102,39 @@ def analyser_fichier_specifique(self, chemin_fichier_chiffre: str) -> ResultatAn
 
             # Parcours des algorithmes disponibles
             scores_algorithmes = {}
+
+            # Pour les arrêts en cas d'erreurs, servira à upgrade la progress bar
+            cumul_progress_avance = 0
+
             for nom_algo, analyzer in self.analyzers.items():
+                avance_algo = avance/(len(self.analyzers)*3 * 0.5)
                 time.sleep(0.5) # TODO : Mise à jour de la progress bar -> step : Utilisation de {algrorithme} pour déterminer le chiffrement
-
+                progress.update(task_id=task, description=f"Utilisation de {nom_algo} pour déterminer le chiffrement", advance=avance_algo)
+                time.sleep(1)
+
                 score = analyzer.identifier_algo(f"data/{chemin_fichier_chiffre}")
                 scores_algorithmes[nom_algo] = score
+
                 time.sleep(0.5) # TODO : Mise à jour de la progress bar -> step : Analyse des résultats d'identification
-                # print(f"{nom_algo}: score {score:.2f}")
+                progress.update(task_id=task, description="Analyse des résultats d'identification", advance=avance_algo)
+                time.sleep(1)
+
+                cumul_progress_avance += 2 * avance_algo
 
                 if score > 0.9 :  # Seuil de confiance
                     time.sleep(1) # TODO : Mise à jour de la progress bar -> step : Détection réussie pour {algorithme} et préparation du rapport d'analyse
+                    progress.update(task_id=task, description=f"Détection réussie pour {nom_algo} et préparation du rapport d'analyse", advance=((100/nbr_opr_mission) - cumul_progress_avance))
+                    time.sleep(1)
+
                     algorithme_detecte = nom_algo
                     score_probabilite = score
-                    # print(f"Algorithme détecté: {algorithme_detecte} (score: {score:.2f})")
                     break
                 else :
-                    pass # TODO : Intégrer la progress bar -> step : Echec d'identification pour {algorithme} 
+                    time.sleep(1) # TODO : Intégrer la progress bar -> step : Echec d'identification pour {algorithme}
+                    progress.update(task_id=task, description=f"Echec d'identification pour {nom_algo}", advance=avance_algo)
+                    time.sleep(1)
+                    cumul_progress_avance += avance_algo
+
             if not algorithme_detecte:
                 print("Aucun algorithme correctement détecté ")
                 temps_execution = time.time() - debut_analyse
@@ -115,15 +147,25 @@ def analyser_fichier_specifique(self, chemin_fichier_chiffre: str) -> ResultatAn
         except Exception as e:
             print(f"Erreur lors de l'analyse: {str(e)}")
             temps_execution = time.time() - debut_analyse
+            error = True
             return ResultatAnalyse("", b"", 0.0, b"", temps_execution, 0, chemin_fichier_chiffre)
 
     def __tenter_dechiffrement_avec_dictionnaire(self, chemin_fichier: str, cles_candidates: list[bytes], analyzer: CryptoAnalyzer, resultat: ResultatAnalyse):
+        """
+            Tente de déchiffrer un fichier avec les clés candidates et l'analyzer correspondant
+
+            Args: 
+                chemin_fichier(str) : chemin vers le fichier
+                cles_candidates(list[bytes]) : les clés candidates retenus par le dossier de clés sur la base des indices
+                analyzer(CryptoAnalyzer) : l'Analyzer correspondant à ce fichier
+                resultat(ResultatAnalyse) : les résultats de l'analyse de fichier 
+
+            Returns :
+                bool : si une erreur est survenue ou non
+        """
         for j, cle in enumerate(cles_candidates):
             resultat.nb_tentatives += 1
-
-            if j % 100 == 0:  # retour visuel tous les 100 essais
-                print(f"Tentative {j+1}/{len(cles_candidates)}...")
-
+
             texte_dechiffre = analyzer.dechiffrer(chemin_fichier, cle).decode('utf-8')
             succes =  verifier_texte_dechiffre(texte_dechiffre)['taux_succes']
 
@@ -132,9 +174,10 @@ def __tenter_dechiffrement_avec_dictionnaire(self, chemin_fichier: str, cles_can
                 resultat.texte_dechiffre = texte_dechiffre
                 resultat.taux_succes = succes
                 print(f"Clé trouvée après {j+1} tentatives!")
-                return 
+                return False
 
         print("Aucune clé valide trouvée")
+        return True
 
     def mission_complete_automatique(self, dossier_chiffres: str, chemin_dictionnaire: str) -> List[ResultatAnalyse]:
         """
@@ -152,75 +195,113 @@ def mission_complete_automatique(self, dossier_chiffres: str, chemin_dictionnair
 
         debut_mission = time.time()
         resultats: list[ResultatAnalyse] = []
-
         try:
-            # Récupération des fichiers .enc
-            fichiers_enc = [f for f in os.listdir(dossier_chiffres) if f.endswith(".enc")]
-
-            if not fichiers_enc:
-                print("Aucun fichier .enc trouvé dans le dossier")
-                return []
-
-            print(f"{len(fichiers_enc)} fichiers .enc détectés")
-            print("\nANALYSE SÉQUENTIELLE DES FICHIERS")
-
-            for i, fichier in enumerate(fichiers_enc, 0):
-
-                print(f"\nFICHIER {i+1}/{len(fichiers_enc)}: {fichier}")
-
-                chemin_fichier = os.path.join(dossier_chiffres, fichier)
+            with Progress() as progress :
+                # Récupération des fichiers .enc
+                fichiers_enc = [f for f in os.listdir(dossier_chiffres) if f.endswith(".enc")]
 
-                # Analyse du fichier
-                resultat = self.analyser_fichier_specifique(fichier)
+                if not fichiers_enc:
+                    print("Aucun fichier .enc trouvé dans le dossier")
+                    return []
 
-                # Tentative de déchiffrement si algorithme détecté
-                if resultat.algo:
-                    print(f"\nTENTATIVE DE DÉCHIFFREMENT")
+                print(f"{len(fichiers_enc)} fichiers .enc détectés")
+                print("\nANALYSE SÉQUENTIELLE DES FICHIERS")
+                time.sleep(0.5) 
+                for i, fichier in enumerate(fichiers_enc, 0):
+                    print(f"\nFICHIER {i+1}/{len(fichiers_enc)}: {fichier}")
+
+                    # TODO: New progress bar -> step: Analyse du fichier mission{i+1}.enc
+                    task = progress.add_task(f"Analyse du fichier mission{i+1}.enc...", total=100)
+                    time.sleep(0.5)
+
+                    chemin_fichier = os.path.join(dossier_chiffres, fichier)
+
+                    # Analyse du fichier
+                    error = False
+                    resultat = self.analyser_fichier_specifique(fichier, progress, task, error, self._NBR_OPERATION_MISSION)
+
+                    # Tentative de déchiffrement si algorithme détecté
+                    if resultat.algo:
+                        # TODO: MAJ de la progress bar -> step: Amorçage de la phase de déchiffrement
+                        progress.update(task, description="Amorçage de la phase de déchiffrement...", advance=((100/self._NBR_OPERATION_MISSION) * 0.5))
+                        time.sleep(1)
+
+                        analyzer = self.analyzers[resultat.algo]
+
+                        # TODO: MAJ de la progress bar -> step: Récupération des clés candidates
+                        progress.update(task, description="Récupération des clés candidates", advance=(100/self._NBR_OPERATION_MISSION)*0.5)
+                        time.sleep(1)
+
+                        cles_candidates = analyzer.generer_cles_candidates(chemin_dictionnaire)
+
+                        if cles_candidates:
+                            print(f"Test de {len(cles_candidates)} clés candidates...")
+                            # TODO: MAJ de la progress bar -> step: Test de déchiffrement
+                            progress.update(task, description="Test de déchiffrement", advance=(100/self._NBR_OPERATION_MISSION))
+                            time.sleep(3)
+
+                            error = self.__tenter_dechiffrement_avec_dictionnaire(chemin_fichier, cles_candidates, analyzer, resultat) 
+
+                        else:
+                            # TODO: MAJ de la progress bar -> step: Abort et récupération des résultats d'analyse
+                            progress.update(task, description="Aucune clé candidate générée ❌ (Aborting ...)", advance=(100/self._NBR_OPERATION_MISSION))
+                            time.sleep(3)
+                            error = True
+
 
-                    analyzer = self.analyzers[resultat.algo]
-                    cles_candidates = analyzer.generer_cles_candidates(chemin_dictionnaire)
+                    resultats.append(resultat)
 
-                    if cles_candidates:
-                        print(f"Test de {len(cles_candidates)} clés candidates...")
+                    # retour visuel
+                    if resultat.algo:
+                        # TODO: MAJ de la progress bar -> step: Finalsation et retour de résultats
+                        progress.update(task, description="Finalisation et retour des résultats", advance=(100/self._NBR_OPERATION_MISSION))
+                        time.sleep(3)
 
-                        self.__tenter_dechiffrement_avec_dictionnaire(chemin_fichier, cles_candidates, analyzer, resultat)
+                        print(f"{fichier}: {resultat.algo} (score: {resultat.score_probabilite:.2f})")
+
+                        message = "[bold green] Mission terminée. ✅[/bold green]\n\n" if not error else "[bold red] Mission terminée: Déchiffrement non concluant. ❌ [/bold red]\n\n"
+                        Console().print(message)
                     else:
-                        print("   Aucune clé candidate générée")
+                        progress.update(task, description="Aborting et récupération des résultats d'analyse...", advance=100)
+                        time.sleep(0.5)  # TODO: MAJ de la progress bar -> step: Abort et récupération des résultats d'analyse
+                        Console().print(f"[bold yellow] Mission terminée: Aucun algorithme détecté. ⚠️[/bold yellow]\n\n")
+
+                    progress.remove_task(task)
 
-                resultats.append(resultat)
+                # Rapport de synthèse final
+                with Progress() as progress :
+                    task = progress.add_task("Préparation des rapports", total=100) # TODO: New progress bar -> step: Préparation des rapports (1 to 100%)
+
+                    while not progress.finished :
+                        progress.update(task, description="Préparation des rapports", advance=2)
+                        time.sleep(0.2)
+                    for i in range(len(fichiers_enc)) :
+                        resultat = {
+                            'algorithme': resultats[i].algo,
+                            'fichier': resultats[i].fichier,
+                            'cle': resultats[i].cle,
+                            'tentatives': resultats[i].nb_tentatives,
+                            'temps_execution': resultats[i].temps_execution,
+                            'taux_succes': resultats[i].taux_succes,
+                            'statut_succes' : 'Succès' if resultats[i].taux_succes > 60 else 'Echec',
+                            'texte_dechiffre' : resultats[i].texte_dechiffre
+                        }
+                        rapport_mission().generer_rapport_synthese(resultat)
+                        progress.update(task, description="Mission complète effectuée.") # TODO: MAJ de la progress bar -> step: Mission complète effectuée
+
+                # Mise à jour des statistiques globales
+                self.missions_completees.append({
+                    "dossier": dossier_chiffres,
+                    "resultats": resultats,
+                    "temps_total": time.time() - debut_mission
+                })
 
-                # retour visuel
-                if resultat.algo:
-                    print(f"{fichier}: {resultat.algo} (score: {resultat.score_probabilite:.2f}) \n\n")
-                else:
-                    print(f"{fichier}: Aucun algorithme détecté")
-
-            # Rapport de synthèse final
-            for i in range(6) :
-                resultat = {
-                    'algorithme': resultats[i].algo,
-                    'fichier': resultats[i].fichier,
-                    'cle': resultats[i].cle,
-                    'tentatives': resultats[i].nb_tentatives,
-                    'temps_execution': resultats[i].temps_execution,
-                    'taux_succes': resultats[i].taux_succes,
-                    'statut_succes' : 'Succès' if resultats[i].taux_succes > 60 else 'Echec',
-                    'texte_dechiffre' : resultats[i].texte_dechiffre
-                }
-                rapport_mission().generer_rapport_synthese(resultat)
-
-            # Mise à jour des statistiques globales
-            self.missions_completees.append({
-                "dossier": dossier_chiffres,
-                "resultats": resultats,
-                "temps_total": time.time() - debut_mission
-            })
-
-            return resultats
+                return resultats
 
         except Exception as e:
             print(f"Erreur lors de la mission complète: {str(e)}")
             return []
+
 
     def attaque_dictionnaire_manuelle(self, chemin_fichier: str, algorithme_choisi: str, chemin_dictionnaire: str) -> ResultatAnalyse:
         """