bug fix in html & black formatter

Author: CERT-SYNETIS

src/__init__.py

@@ -1,6 +1,7 @@
 import os
 import yaml
 import json
+from logging import Logger
 from slugify import slugify
 from src.thirdparty import triageutils as triageutils
 from src.thirdparty.logging import get_logger
@@ -73,7 +74,7 @@ def _ParseMapFile(self):
         with open(os.path.join("config", "mapping.json")) as mapp:
             return json.load(mapp)
 
-    def run(self):
+    def run(self, logger: Logger):
         """Main entry point of the plugin"""
         raise NotImplementedError("[BasePlugin] run() needs to be overriden")
 

src/plugins/generaptor.py

@@ -88,8 +88,8 @@ def __init__(self, conf: dict):
         self.filebeat_dir = os.path.join(self.generaptor_dir, "filebeat")
         triageutils.create_directory_path(path=self.filebeat_dir, logger=self.logger)
 
-        self.activitiescache_share = os.path.join(
-            self.generaptor_dir, "ActivitiesCache"
+        self.activitiescache_share = Path(
+            os.path.join(self.generaptor_dir, "ActivitiesCache")
         )
         triageutils.create_directory_path(
             path=self.activitiescache_share, logger=self.logger
@@ -319,7 +319,7 @@ def generate_plaso_timeline(self, logger: Logger):
                     logger=self.logger,
                 )
         except Exception as ex:
-            self.logger.error(f"[generate_plaso_timeline] {ex}")
+            self.error(f"[generate_plaso_timeline] {ex}")
 
     @triageutils.LOG
     def generate_psort_timeline(self, plasofile: str, logger: Logger) -> str:
@@ -368,7 +368,7 @@ def generate_psort_timeline(self, plasofile: str, logger: Logger) -> str:
             s_file = os.path.join(self.plaso_folder, f"psort-{self.hostname}.jsonl")
             return s_file
         except Exception as ex:
-            self.logger.error(f"[generate_psort_timeline] {ex}")
+            self.error(f"[generate_psort_timeline] {ex}")
             return ""
 
     @triageutils.LOG
@@ -408,8 +408,8 @@ def get_evtx(self, evtx_folder: Path, logger: Logger) -> list:
         if not evtx_folder:
             raise Exception("No evtx folder")
         records.extend(
-            triageutils.search_files(
-                src=evtx_folder, pattern=".evtx", logger=self.logger
+            triageutils.search_files_by_extension(
+                dir=evtx_folder, extension=".evtx", logger=self.logger
             )
         )
         if len(records):
@@ -695,12 +695,20 @@ def generaptor_parse_evtx(self, logger: Logger):
                 if self.is_logstash_active:
                     _file_infos = triageutils.get_file_informations(filepath=_f)
                     _analytics = triageutils.generate_analytics(logger=self.logger)
-                    _analytics["log"]["file"]["eventcount"] = _res.get("nb_events_read", 0)
-                    _analytics["log"]["file"]["eventsent"] = _res.get("nb_events_sent", 0)
+                    _analytics["log"]["file"]["eventcount"] = _res.get(
+                        "nb_events_read", 0
+                    )
+                    _analytics["log"]["file"]["eventsent"] = _res.get(
+                        "nb_events_sent", 0
+                    )
                     _analytics["log"]["file"]["path"] = str(_f)
                     _analytics["log"]["file"]["size"] = _file_infos.get("fileSize", 0)
-                    _analytics["log"]["file"]["lastaccessed"] = _file_infos.get("lastAccessTime", 0)
-                    _analytics["log"]["file"]["creation"] = _file_infos.get("creationTime", 0)
+                    _analytics["log"]["file"]["lastaccessed"] = _file_infos.get(
+                        "lastAccessTime", 0
+                    )
+                    _analytics["log"]["file"]["creation"] = _file_infos.get(
+                        "creationTime", 0
+                    )
                     _analytics["csirt"]["client"] = self.clientname
                     _analytics["csirt"]["hostname"] = self.hostname
                     _analytics["csirt"]["application"] = "generaptor_parse_evtx"
@@ -742,7 +750,7 @@ def generaptor_parse_mft(self, logger: Logger):
                 )
                 _analyzer.analyze()
             else:
-                self.logger.error(f"[generaptor_parse_mft] No $MFT found")
+                self.error(f"[generaptor_parse_mft] No $MFT found")
         except Exception as ex:
             self.error(f"[generaptor_parse_mft] {str(ex)}")
             raise ex
@@ -765,7 +773,7 @@ def generaptor_parse_usnjrnl(self, logger: Logger):
                 )
                 _analyzer.analyze()
             else:
-                self.logger.error(f"[generaptor_parse_usnjrnl] No $UsnJrnl%3A$J found")
+                self.error(f"[generaptor_parse_usnjrnl] No $UsnJrnl%3A$J found")
         except Exception as ex:
             self.error(f"[generaptor_parse_usnjrnl] {str(ex)}")
             raise ex
@@ -864,17 +872,18 @@ def generaptor_get_consolehost_history(self, logger: Logger):
             ):
                 self.info(f"[generaptor_get_consolehost_history] Parse: {_f}")
                 try:
-                    _username = _f.parts[_f.parts.index('Users')+1]
+                    _username = _f.parts[_f.parts.index("Users") + 1]
                 except Exception as errorname:
                     self.error(f"{errorname}")
                     _username = time.time()
                 _dst = self.psreadline_dir / Path(f"{_username}")
-                triageutils.copy_file(src=_f, dst=_dst, overwrite=True, logger=self.logger)
+                triageutils.copy_file(
+                    src=_f, dst=_dst, overwrite=True, logger=self.logger
+                )
         except Exception as ex:
             self.error(f"[generaptor_get_consolehost_history] {str(ex)}")
             raise ex
 
-
     @triageutils.LOG
     def run(self, logger: Logger):
         """Fonction principale qui exécute tout le triage de generaptor
@@ -898,10 +907,10 @@ def run(self, logger: Logger):
                 if self.is_logstash_active:
                     self.ymlcreator(logger=self.logger)
                     self.check_docker_image(
-                            image_name=self.docker_images["filebeat"]["image"],
-                            tag=self.docker_images["filebeat"]["tag"],
-                            logger=self.logger,
-                        )
+                        image_name=self.docker_images["filebeat"]["image"],
+                        tag=self.docker_images["filebeat"]["tag"],
+                        logger=self.logger,
+                    )
                     self.generaptor_filebeat(logger=self.logger)
                 if self.config["run"]["generaptor"]["timeline"]:
                     self.info("[generaptor] Run PLASO")

src/plugins/hayabusa.py

@@ -22,17 +22,23 @@ def __init__(self, conf: dict):
                 )
             )
             if _evtx_folder:
-                self.hayabusa_dir = _evtx_folder.parent
+                self.evtx_dir = _evtx_folder.parent
+                self.hayabusa_dir = Path(
+                    os.path.join(self.upload_dir, self.hostname, "Hayabusa")
+                )
+                triageutils.create_directory_path(
+                    path=self.hayabusa_dir, logger=self.logger
+                )
             else:
                 self.error("[HAYABUSA] No evtx folder")
                 raise Exception("[HAYABUSA] No evtx folder")
-            self.output_json = f"{os.path.join(self.hayabusa_dir,self.clientname)}_HAYABUSA_SIGMA.jsonl"
+            self.output_json = f"{self.hayabusa_dir}/HAYABUSA_SIGMA.jsonl"
         except Exception as ex:
             self.error(f"[init] {ex}")
             raise ex
 
     @triageutils.LOG
-    def exec_hayabusa(self, log_folder=None, logger=None):
+    def exec_hayabusa(self, log_folder: Path, logger=None):
         """Exécution du binaire hayabusa sur un dossier
 
         Args:
@@ -41,8 +47,6 @@ def exec_hayabusa(self, log_folder=None, logger=None):
 
         """
         try:
-            if not log_folder:
-                log_folder = self.hayabusa_dir
             cmd = [
                 self.hayabusa_bin_path,
                 "json-timeline",
@@ -142,8 +146,12 @@ def send_analytics_to_elk(self, event_sent: int, logger=None):
                 _analytics["log"]["file"]["eventsent"] = event_sent
                 _analytics["log"]["file"]["path"] = self.output_json.name
                 _analytics["log"]["file"]["size"] = _file_infos.get("fileSize", 0)
-                _analytics["log"]["file"]["lastaccessed"] = _file_infos.get("lastAccessTime", 0)
-                _analytics["log"]["file"]["creation"] = _file_infos.get("creationTime", 0)
+                _analytics["log"]["file"]["lastaccessed"] = _file_infos.get(
+                    "lastAccessTime", 0
+                )
+                _analytics["log"]["file"]["creation"] = _file_infos.get(
+                    "creationTime", 0
+                )
                 _analytics["csirt"]["client"] = self.clientname
                 _analytics["csirt"]["hostname"] = self.hostname
                 _analytics["csirt"]["application"] = "hayabusa"
@@ -167,7 +175,7 @@ def run(self, logger=None):
 
         """
         try:
-            self.exec_hayabusa(logger=self.logger)
+            self.exec_hayabusa(log_folder=self.evtx_dir, logger=self.logger)
             if self.is_logstash_active:
                 _event_sent = self.send_to_elk(logger=self.logger)
                 self.send_analytics_to_elk(event_sent=_event_sent, logger=self.logger)

src/plugins/kape.py

@@ -854,12 +854,14 @@ def kape_get_consolehost_history(self, logger: Logger):
             ):
                 self.info(f"[kape_get_consolehost_history] Parse: {_f}")
                 try:
-                    _username = _f.parts[_f.parts.index('Users')+1]
+                    _username = _f.parts[_f.parts.index("Users") + 1]
                 except Exception as errorname:
                     self.error(f"{errorname}")
                     _username = time.time()
                 _dst = self.psreadline_dir / Path(f"{_username}")
-                triageutils.copy_file(src=_f, dst=_dst, overwrite=True, logger=self.logger)
+                triageutils.copy_file(
+                    src=_f, dst=_dst, overwrite=True, logger=self.logger
+                )
         except Exception as ex:
             self.error(f"[kape_get_consolehost_history] {str(ex)}")
             raise ex
@@ -943,11 +945,11 @@ def run(self, logger: Logger):
                 except Exception as err_reg:
                     self.error(f"[kape ERROR] {str(err_reg)}")
             if self.config["run"]["kape"].get("psreadline", False):
-                    self.info("[kape] Run PSReadline")
-                    try:
-                        self.kape_get_consolehost_history(logger=self.logger)
-                    except Exception as err_reg:
-                        self.error(f"[kape ERROR] {str(err_reg)}")
+                self.info("[kape] Run PSReadline")
+                try:
+                    self.kape_get_consolehost_history(logger=self.logger)
+                except Exception as err_reg:
+                    self.error(f"[kape ERROR] {str(err_reg)}")
             if self.config["run"]["kape"].get("iis", False):
                 try:
                     self.info("[KAPE] Run IIS")

src/plugins/o365.py

@@ -10,7 +10,7 @@ class Plugin(BasePlugin):
     O365 plugin pour triage
     """
 
-    def __init__(self, conf=None):
+    def __init__(self, conf: dict):
         super().__init__(config=conf)
         self.o365_dir = os.path.join(self.upload_dir, self.hostname, "o365")
         triageutils.create_directory_path(path=self.o365_dir, logger=self.logger)
@@ -77,7 +77,7 @@ def o365_get_csv_files(self, logger=None) -> list:
         return records
 
     @triageutils.LOG
-    def o365_send_json_results(self, json_file=None, logger=None):
+    def o365_send_json_results(self, json_file: Path, logger=None):
         """Fonction qui envoie les résultats json o365 vers ELK"""
         try:
             try:

src/plugins/orc.py

@@ -77,9 +77,10 @@ def kill_docker_container(self, logger: Logger):
         _docker.close()
 
     @triageutils.LOG
-    def rename_orc_file(self, filepath: Path, logger: Logger):
+    def rename_orc_file(self, filepath: Path, logger: Logger, LOGLEVEL: str ="NOLOG"):
         """
         Rename file by keeping only real file name
+        logger and LOGLEVEL are used by LOG decorator
 
         return:
             Path: file's new path
@@ -99,7 +100,7 @@ def rename_orc_file(self, filepath: Path, logger: Logger):
             _new_path = Path(_path) / Path(_new_name)
 
             if triageutils.file_exists(file=_new_path, LOGLEVEL="NOLOG"):
-                self.info(f"[rename_orc_file] File exists !")
+                #self.info(f"[rename_orc_file] File exists !")
                 _parent = Path(_path) / Path(str(round(time.time() * 1000)))
                 triageutils.create_directory_path(path=_parent, LOGLEVEL="NOLOG")
                 _new_path = _parent / Path(_new_name).name
@@ -152,7 +153,7 @@ def extract_all_7z(self, logger: Logger):
             for _file in triageutils.search_files_by_extension_generator(
                 src=self.orc_dir, extension=".data", logger=self.logger
             ):
-                records.append(self.rename_orc_file(filepath=_file, logger=self.logger))
+                records.append(self.rename_orc_file(filepath=_file, logger=self.logger, LOGLEVEL="NOLOG"))
             return records
         except Exception as ex:
             self.error(f"[extract_all_7z] {str(ex)}")