Merge pull request #32 from glrs/feature/core

Enhance code and tackle weaknesses found in testing

lib/core_utils/common.py

@@ -80,14 +80,38 @@ def get_path(file_name: str) -> Optional[Path]:
 
         Returns:
             Optional[Path]: A Path object representing the full path to the specified
-                configuration file, or None if the file is not found.
+                configuration file, or None if the file is not found or is invalid.
         """
-        config_file = YggdrasilUtilities.CONFIG_DIR / file_name
+        # Convert to Path object
+        requested_path = Path(file_name)
 
-        if config_file.exists():
-            return config_file
-        else:
-            logging.error(f"Configuration file '{file_name}' not found.")
+        # If file_name is absolute or tries to go outside CONFIG_DIR, return None immediately
+        if requested_path.is_absolute():
+            logging.error(f"Absolute paths are not allowed: '{file_name}'")
+            return None
+
+        # Construct the path within CONFIG_DIR
+        config_file = YggdrasilUtilities.CONFIG_DIR / requested_path
+
+        # Check if the constructed path is still within CONFIG_DIR (no directory traversal)
+        try:
+            # Resolve both paths to their absolute forms and ensure CONFIG_DIR is a parent of config_file
+            config_file_resolved = config_file.resolve()
+            config_dir_resolved = YggdrasilUtilities.CONFIG_DIR.resolve()
+
+            if config_dir_resolved not in config_file_resolved.parents:
+                logging.error(
+                    f"Attempted directory traversal outside config dir: '{file_name}'"
+                )
+                return None
+
+            if config_file_resolved.exists():
+                return config_file_resolved
+            else:
+                logging.error(f"Configuration file '{file_name}' not found.")
+                return None
+        except Exception as e:
+            logging.error(f"Error resolving config file path '{file_name}': {e}")
             return None
 
     @staticmethod
@@ -116,7 +140,9 @@ def get_last_processed_seq() -> str:
 
         if seq_file and seq_file.is_file():
             with open(seq_file) as file:
-                return file.read().strip()
+                content = file.read().strip()
+                # If the file is empty, return "0"
+                return content if content else "0"
         else:
             # Otherwise return a default sequence value of your choice.
             # NOTE: Zero (0) means start from the beginning. Note ideal!
@@ -134,5 +160,15 @@ def save_last_processed_seq(last_processed_seq: str) -> None:
         seq_file = YggdrasilUtilities.get_path(".last_processed_seq")
 
         if seq_file:
-            with open(seq_file, "w") as file:
-                file.write(last_processed_seq)
+            try:
+                with open(seq_file, "w") as file:
+                    file.write(last_processed_seq)
+            except Exception as e:
+                logging.error(f"Failed to save last processed seq: {e}")
+                # Don't re-raise, just log and exit the method gracefully
+        else:
+            logging.warning(
+                "Failed to save last processed seq:"
+                "'.last_processed_seq' File not found in the configurations."
+            )
+            pass

lib/core_utils/config_loader.py

@@ -84,13 +84,19 @@ def _load_config(self, file_name, is_path=False):
                 # TODO: Perform validation and error checking on the loaded data if needed.
                 self._config = types.MappingProxyType(config)
         except json.JSONDecodeError as e:
+            # Set config to empty immutable mapping before raising
+            self._config = types.MappingProxyType({})
             raise json.JSONDecodeError(
                 f"Error parsing config file '{config_file}': {e}", e.doc, e.pos
             )
         except TypeError as e:
+            # Set config to empty immutable mapping before raising
+            self._config = types.MappingProxyType({})
             raise TypeError(f"Error parsing config file '{config_file}': {e}")
         except Exception as e:
             logging.error(f"Unexpected error loading config file '{config_file}': {e}")
+            # Set config to empty immutable mapping before raising
+            self._config = types.MappingProxyType({})
             raise
 
         return self._config

lib/module_utils/ngi_report_generator.py

@@ -24,22 +24,22 @@ def generate_ngi_report(
     # Convert list of samples into a space-separated string for the command line
     samples_str = " ".join(sample_list)
 
-    # Command to activate the environment and run the NGI report generation
-    activate_env_cmd = configs.get("activate_ngi_cmd")
-    if not activate_env_cmd:
-        logging.error(
-            "NGI environment activation command not found in the configuration. "
-            "NGI report will not be generated."
-        )
-        return False
+    try:
+        # Command to activate the environment and run the NGI report generation
+        activate_env_cmd = configs.get("activate_ngi_cmd")
+        if not activate_env_cmd:
+            logging.error(
+                "NGI environment activation command not found in the configuration. "
+                "NGI report will not be generated."
+            )
+            return False
 
-    report_cmd = (
-        f"ngi_reports project_summary -d {project_path} -p {project_id} "
-        f"-s '{user_name}' -y --no_txt --samples {samples_str}"
-    )
-    full_cmd = f"{activate_env_cmd} && {report_cmd}"
+        report_cmd = (
+            f"ngi_reports project_summary -d {project_path} -p {project_id} "
+            f"-s '{user_name}' -y --no_txt --samples {samples_str}"
+        )
+        full_cmd = f"{activate_env_cmd} && {report_cmd}"
 
-    try:
         # Execute the combined command
         # NOTE: Perhaps use `check=False` to prevent raising
         #       CalledProcessError on non-zero exit codes

lib/module_utils/report_transfer.py

@@ -13,6 +13,12 @@ def transfer_report(
 ) -> bool:
     try:
         report_transfer_config = configs["report_transfer"]
+        if not isinstance(report_transfer_config, dict):
+            logging.error(
+                "Invalid configuration type for 'report_transfer'. Expected a dictionary."
+            )
+            return False
+
         server = report_transfer_config["server"]
         user = report_transfer_config["user"]
         destination_path = report_transfer_config["destination"]
@@ -42,7 +48,7 @@ def transfer_report(
     ]
 
     # logging.debug(f"RSYNC command: {' '.join(rsync_command)}")
-
+    result = None
     try:
         # Execute the rsync command
         result = subprocess.run(
@@ -59,5 +65,9 @@ def transfer_report(
         return False
     except Exception as e:
         logging.error(f"Unexpected error during report transfer: {e}")
-        logging.error(f"RSYNC output: {result.stdout}")
+        # Check if result is not None before accessing its attributes
+        if result is not None:
+            logging.error(f"RSYNC output: {result.stdout}")
+        else:
+            logging.error("RSYNC output: No output available due to early error.")
         return False

lib/module_utils/slurm_utils.py

@@ -25,6 +25,11 @@ def generate_slurm_script(
     Returns:
         bool: True if the script is successfully written, False otherwise.
     """
+    if not isinstance(template_fpath, (str, Path)):
+        raise TypeError("template_fpath must be a string or a Path object")
+    if not isinstance(output_fpath, (str, Path)):
+        raise TypeError("output_fpath must be a string or a Path object")
+
     try:
         template_path = Path(template_fpath)
         output_path = Path(output_fpath)