Bugfix/general

nodescraper/base/inbandcollectortask.py

@@ -55,6 +55,7 @@ def __init__(
         task_result_hooks: Optional[list[TaskResultHook]] = None,
         event_reporter: str = DEFAULT_EVENT_REPORTER,
         session_id: Optional[str] = None,
+        log_path: Optional[str] = None,
         **kwargs,
     ):
         """Creates a InBandDataCollector Class
@@ -83,6 +84,7 @@ def __init__(
             task_result_hooks=task_result_hooks,
             event_reporter=event_reporter,
             session_id=session_id,
+            log_path=log_path,
         )
         if self.system_info.os_family not in self.SUPPORTED_OS_FAMILY:
             raise SystemCompatibilityError(

nodescraper/base/match_ignore.py

@@ -57,13 +57,15 @@ def parse_mce_bank_spec(spec: Sequence[MceBankSpec]) -> frozenset[int]:
                 raise ValueError(f"Invalid MCE bank number: {entry}")
             banks.add(entry)
             continue
-
         token = str(entry).strip()
         if not token:
             raise ValueError("Empty MCE bank entry")
-
         if "-" in token:
             start_text, end_text = token.split("-", 1)
+            # Check if it looks like "-5"
+            if not start_text.strip() or not end_text.strip() or (int(end_text.strip()) < 0):
+                raise ValueError(f"Invalid MCE bank range: {entry}")
+
             start = int(start_text.strip())
             end = int(end_text.strip())
             if start < 0 or end < 0 or start > end:

nodescraper/base/regexanalyzer.py

@@ -402,8 +402,9 @@ def _is_within_interval(new_timestamp_str: str, existing_timestamps: list[str])
                         new_event.data["timestamp"] = timestamp
 
                     regex_event_list.append(new_event)
-
                 search_from = match_obj.end()
+                if search_from == match_obj.start():  # Zero-width match
+                    search_from += 1  # Force advancement
 
         all_events = list(regex_map.values()) if group else regex_event_list
 

nodescraper/cli/cli.py

@@ -649,6 +649,7 @@ def main(
             )
 
             if parsed_args.skip_sudo:
+                # Add skip_sudo to the collection_args of the last plugin config instance
                 plugin_config_inst_list[-1].global_args.setdefault("collection_args", {})[
                     "skip_sudo"
                 ] = True

nodescraper/cli/compare_runs.py

@@ -125,7 +125,7 @@ def _load_plugin_data_from_run(
             res_payload = json.loads(Path(res_path).read_text(encoding="utf-8"))
             task_res = TaskResult(**res_payload)
             plugin_name = task_res.parent
-        except (json.JSONDecodeError, TypeError, OSError) as e:
+        except (json.JSONDecodeError, TypeError, OSError, ValidationError) as e:
             logger.warning("Skipping %s: failed to load result: %s", res_path, e)
             continue
 

nodescraper/cli/dynamicparserbuilder.py

@@ -156,11 +156,14 @@ def get_literal_choices(cls, type_class_map: dict) -> Optional[list]:
         Returns:
             Optional[list]: list of valid choices for the Literal type, or None if not a Literal
         """
-        # Check if Literal is in the type_class_map
         literal_type = type_class_map.get(Literal)
-        if literal_type and literal_type.inner_type is not None:
+        if literal_type is None or literal_type.inner_type is None:
             return None
-        return None
+
+        values = literal_type.inner_type
+        if isinstance(values, (list, tuple)):
+            return list(values)
+        return [values]
 
     def add_argument(
         self,
@@ -181,14 +184,16 @@ def add_argument(
         """
         add_kw = {} if help_text is None else {"help": help_text}
         # Check for Literal types and extract choices
-        literal_choices = None
-        if Literal in type_class_map and annotation:
-            # Extract all arguments from the annotation
-            args = get_args(annotation)
-            for arg in args:
-                if get_origin(arg) is Literal:
-                    literal_choices = list(get_args(arg))
-                    break
+        literal_choices = self.get_literal_choices(type_class_map)
+        if literal_choices is None and annotation is not None:
+            # fall back to pulling the choices out of the raw annotation
+            if get_origin(annotation) is Literal:
+                literal_choices = list(get_args(annotation))
+            else:
+                for arg in get_args(annotation):
+                    if get_origin(arg) is Literal:
+                        literal_choices = list(get_args(arg))
+                        break
 
         if list in type_class_map:
             type_class = type_class_map[list]
@@ -225,7 +230,7 @@ def add_argument(
                 type=str,
                 required=required,
                 choices=literal_choices,
-                metavar=f"{{{','.join(literal_choices)}}}",
+                metavar=f"{{{','.join(str(choice) for choice in literal_choices)}}}",
                 **add_kw,
             )
         elif float in type_class_map:

nodescraper/cli/helper.py

@@ -30,6 +30,7 @@
 import logging
 import os
 import sys
+from copy import deepcopy
 from pathlib import Path
 from typing import Optional, Sequence, Tuple
 
@@ -115,14 +116,15 @@ def get_plugin_configs(
 
     base_config.global_args["system_interaction_level"] = system_interaction_level
 
-    plugin_configs = [base_config]
+    # Copy each until we are done
+    plugin_configs = [deepcopy(base_config)]
 
     if plugin_config_input:
         for config in plugin_config_input:
             if os.path.exists(config):
                 plugin_configs.append(ModelArgHandler(PluginConfig).process_file_arg(config))
             elif config in built_in_configs:
-                plugin_configs.append(built_in_configs[config])
+                plugin_configs.append(deepcopy(built_in_configs[config]))
             else:
                 raise argparse.ArgumentTypeError(f"No plugin config found for: {config}")
 
@@ -269,7 +271,11 @@ def parse_gen_plugin_config(
     """
     try:
         config = build_config(
-            config_reg, plugin_reg, logger, parsed_args.plugins, parsed_args.built_in_configs
+            config_reg,
+            plugin_reg,
+            logger,
+            parsed_args.plugins,
+            parsed_args.built_in_configs,
         )
 
         config.name = parsed_args.config_name.split(".")[0]
@@ -367,7 +373,7 @@ def generate_reference_config(
 
         data_model = obj.result_data.system_data
         if data_model is None:
-            logger.warning("Plugin: %s data model not found: %s, skipping", obj.source)
+            logger.warning("Plugin: %s data model not found: %s, skipping", obj.source, data_model)
             continue
 
         plugin = plugin_reg.plugins.get(obj.source)
@@ -585,15 +591,20 @@ def dump_to_csv(all_rows: list, filename: str, fieldnames: list[str], logger: lo
         fieldnames (list[str]): header for csv file
         logger (logging.Logger): isntance of logger
     """
+    DEFAULT_ERR_MSG_PREFIX = "Could not dump data to csv file"
     try:
         with open(filename, "w", newline="") as f:
             writer = csv.DictWriter(f, fieldnames=fieldnames)
             writer.writeheader()
             for row in all_rows:
                 writer.writerow(row)
+        logger.info("Data written to csv file: %s", filename)
+    except FileNotFoundError as fnf_exp:
+        logger.error("%s, File not found: %s", DEFAULT_ERR_MSG_PREFIX, fnf_exp)
+    except ValueError as val_exp:
+        logger.error("%s, Value error: %s", DEFAULT_ERR_MSG_PREFIX, val_exp)
     except Exception as exp:
-        logger.error("Could not dump data to csv file: %s", exp)
-    logger.info("Data written to csv file: %s", filename)
+        logger.error("%s, Exception: %s", DEFAULT_ERR_MSG_PREFIX, exp)
 
 
 def generate_summary(

nodescraper/cli/inputargtypes.py

@@ -86,6 +86,13 @@ class ModelArgHandler(Generic[TModelType]):
     def __init__(self, model: Type[TModelType]) -> None:
         self.model = model
 
+    def arg_check(self, json_arg_input_data: dict):
+        """Check the validity of the JSON argument input data.
+        It must be a dict which can be ** into a pydantic model of the specified type.
+        """
+        if not isinstance(json_arg_input_data, dict):
+            raise argparse.ArgumentTypeError("Input data must be a dictionary.")
+
     def process_file_arg(self, file_path: str) -> TModelType:
         """load a json file into a pydantic model
 
@@ -99,6 +106,7 @@ def process_file_arg(self, file_path: str) -> TModelType:
             TModelType: model instance
         """
         data = json_arg(file_path)
+        self.arg_check(data)
         try:
             return self.model(**data)
         except ValidationError as e:

nodescraper/configregistry.py

@@ -79,16 +79,19 @@ def load_configs(self, config_path: Optional[str] = None):
         config_path = Path(config_path)
 
         for config_file in config_path.glob("*.json"):
-            with open(config_file, "r", encoding="utf-8") as in_file:
-                try:
-                    file_data = json.load(in_file)
-                    config_model = PluginConfig(**file_data)
-                    if config_model.name:
-                        self.configs[config_model.name] = config_model
-                    else:
-                        self.configs[config_file.name] = config_model
-                except (ValidationError, json.JSONDecodeError):
-                    pass
+            try:
+                with open(config_file, "r", encoding="utf-8") as in_file:
+                    try:
+                        file_data = json.load(in_file)
+                        config_model = PluginConfig(**file_data)
+                        if config_model.name:
+                            self.configs[config_model.name] = config_model
+                        else:
+                            self.configs[config_file.name] = config_model
+                    except (ValidationError, json.JSONDecodeError, TypeError) as e:
+                        raise RuntimeError(f"Failed to load config from {config_file}: {e}") from e
+            except (OSError, IOError, FileNotFoundError) as e:
+                raise RuntimeError(f"Failed to open config file {config_file}") from e
 
     @staticmethod
     def _entry_points_for_group(group: str):
@@ -107,7 +110,9 @@ def _entry_points_for_group(group: str):
             return all_eps.get(group, [])  # type: ignore[assignment, attr-defined, arg-type]
 
     @staticmethod
-    def _resolve_entry_point_config(loaded: Any) -> PluginConfig | dict[str, Any] | None:
+    def _resolve_entry_point_config(
+        loaded: Any,
+    ) -> PluginConfig | dict[str, Any] | None:
         """Resolve a loaded entry point object into a plugin config.
 
         Args:

nodescraper/connection/inband/inband.py

@@ -25,6 +25,7 @@
 ###############################################################################
 import abc
 import os
+from pathlib import Path
 from typing import Optional
 
 from pydantic import BaseModel
@@ -67,6 +68,41 @@ def log_model(self, log_path: str) -> None:
     def contents_str(self) -> str:
         pass
 
+    def resolve_write_path(self, log_path: str) -> Path:
+        """Resolve where this artifact should be written and create any parent dirs
+
+        The filename is always treated as relative to log_path. Any anchor and
+        parent dir references are dropped so that a filename can never write
+        outside of log_path, while intentional sub directories are preserved.
+
+        Args:
+            log_path (str): dir that the artifact should be written into
+
+        Raises:
+            IOError: if the filename has no usable name or the dirs cannot be created
+
+        Returns:
+            Path: path to write the artifact to
+        """
+        filename_pathlib = Path(self.filename)
+        parts = [
+            part
+            for part in filename_pathlib.parts
+            if part not in (filename_pathlib.anchor, os.pardir, os.curdir)
+        ]
+
+        if not parts:
+            raise IOError(f"Artifact filename '{self.filename}' does not contain a file name")
+
+        write_path = Path(log_path).joinpath(*parts)
+
+        try:
+            write_path.parent.mkdir(parents=True, exist_ok=True)
+        except Exception as e:
+            raise IOError(f"Failed to create log directory {write_path.parent}: {e}") from e
+
+        return write_path
+
     @classmethod
     def from_bytes(
         cls,
@@ -107,8 +143,10 @@ def log_model(self, log_path: str) -> None:
         Args:
             log_path (str): Path for file
         """
-        path = os.path.join(log_path, self.filename)
-        with open(path, "w", encoding="utf-8") as f:
+        # the filename can look like a folder or a file, make sure it goes in log_path
+        write_path = self.resolve_write_path(log_path)
+
+        with open(write_path, "w", encoding="utf-8") as f:
             f.write(self.contents)
 
     def contents_str(self) -> str:
@@ -131,8 +169,9 @@ def log_model(self, log_path: str) -> None:
         Args:
             log_path (str): Path for file
         """
-        log_name = os.path.join(log_path, self.filename)
-        with open(log_name, "wb") as f:
+        write_path = self.resolve_write_path(log_path)
+
+        with open(write_path, "wb") as f:
             f.write(self.contents)
 
     def contents_str(self) -> str:

nodescraper/connection/inband/inbandremote.py

@@ -154,16 +154,17 @@ def run_command(
 
         try:
             stdin, stdout, stderr = self.client.exec_command(cmd_str, timeout=timeout)
-
             if write_password:
                 stdin.write(
-                    self.ssh_params.password.get_secret_value()
-                    if self.ssh_params.password
-                    else "" + "\n"
+                    (
+                        self.ssh_params.password.get_secret_value()
+                        if self.ssh_params.password
+                        else ""
+                    )
+                    + "\n"
                 )
                 stdin.flush()
                 stdin.channel.shutdown_write()
-
             stdout_str = stdout.read().decode("utf-8", errors="replace")
             stderr_str = stderr.read().decode("utf-8", errors="replace")
             exit_code = stdout.channel.recv_exit_status()