Feature/progress bar (#49)

* Documentation de la fonction d'entropie * Essai de fusion (1/2) * Implementation de la fonction de validation du texte déchiffrer * Récupération du main (1/2) * Mise en place des tests unitaires (1/4) * Revert "Mise en place des tests unitaires (1/4)" This reverts commit e965801. * Mise en place des tests unitaires (1/4) * Définition de la classe de gestion des rapports de ission et implémentatioon de la fonction de génération des rapport de synthèse * Implémentation de la fonction de recherche d'anciens rapports * Mise en place des tests liés à l'analyzer aes cbc * Corrections de typage et de logique * Correction du comportement à la levée de l'exception * Correction de la lgoque de test de test_exception_déchiffrer * Corection de la logique de test_verification_texte_dechiffre * fix: Utilisation de pathlib pour une gestion portable des chemins. * merge réussi * Validation des tests et corrections du rapport mission * chekpoint: Blowfish _Analyzer.identifier_algo() * add: Intiialisation de Blowfish_Analyzer et implémentation de Blowfish_Analyzer.identifier_algo() * Implémentation de Blowfish_Analyzer.filtrer_dictionnaire_par_indice() * Correction de bug * Implementer de Blowfish déchiffrer * Integration de blowfish dans le detecteur crypto * Gestion des bugs de chemin et d'import * Update gitignore * add: Fichier run_tests.py qui run tous les tests et donne le feedback * fix: Standardisation uniforme de tous les imports pour uniformité * tests aes_gcm * fix: Correction de la validation de la taille de clé dans Blowfish_Analyzer pour respecter l'intervalle de 4 à 56 bytes. * fix: Amélioration et uniformisation de la gestion des erreurs pour les différents cas. * fix: Correction des erreurs de logique * merged * test de l'analyzer Fernet * Correction de l'orchestrateur et ajustements correspondants * Lancement du système de stepping pour la progress bar * fix: Corrections mineures * Integration de la progress bar et correction correspondantes --------- Co-authored-by: e-mandy <andymfrd02@gmail.com> Co-authored-by: Let Me Cook <badaroumouwafic@gmail.com>

Author: 3 people

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'))
+

rapport_mission.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

requirements.txt

@@ -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:
  """