added utest + updates to collector for windows

Author: alexandraBara

nodescraper/plugins/inband/kernel_module/analyzer_args.py

@@ -23,6 +23,7 @@
 # SOFTWARE.
 #
 ###############################################################################
+import re
 
 from nodescraper.models import AnalyzerArgs
 from nodescraper.plugins.inband.kernel_module.kernel_module_data import (
@@ -32,7 +33,8 @@
 
 class KernelModuleAnalyzerArgs(AnalyzerArgs):
     kernel_modules: dict[str, dict] = {}
-    modules_filter: list[str] = ["amd"]
+
+    regex_filter: list[str] = ["amd"]
     regex_match: bool = True
 
     @classmethod
@@ -45,8 +47,16 @@ def build_from_model(cls, datamodel: KernelModuleDataModel) -> "KernelModuleAnal
         Returns:
             KernelModuleAnalyzerArgs: instance of analyzer args class
         """
+
+        pattern_regex = re.compile("|".join(datamodel.regex_filter), re.IGNORECASE)
+        filtered_mods = {
+            name: data
+            for name, data in datamodel.kernel_modules.items()
+            if pattern_regex.search(name)
+        }
+
         return cls(
-            kernel_modules=datamodel.kernel_modules,
-            modules_filter=datamodel.modules_filter,
+            kernel_modules=filtered_mods,
+            regex_filter=datamodel.regex_filter,
             regex_match=datamodel.regex_match,
         )

nodescraper/plugins/inband/kernel_module/kernel_module_analyzer.py

@@ -40,20 +40,62 @@ class KernelModuleAnalyzer(DataAnalyzer[KernelModuleDataModel, KernelModuleAnaly
     DATA_MODEL = KernelModuleDataModel
 
     def filter_modules_by_pattern(
-        self, modules: dict[str, dict], patterns: list[str]
-    ) -> dict[str, dict]:
-        pattern_regex = re.compile("|".join(patterns), re.IGNORECASE)
+        self, modules: dict[str, dict], patterns: list[str] = None
+    ) -> tuple[dict[str, dict], list[str]]:
+        if patterns is None:
+            return modules, []
 
-        return {name: data for name, data in modules.items() if pattern_regex.search(name)}
+        matched_modules = {}
+        unmatched_patterns = []
 
-    def filter_modules_by_name(
-        self, modules: dict[str, dict], names: list[str] = None
-    ) -> dict[str, dict]:
+        pattern_match_flags = {p: False for p in patterns}
 
-        if not names:
-            return {name: data for name, data in modules.items()}
+        for mod_name in modules:
+            for p in patterns:
+                if re.search(p, mod_name, re.IGNORECASE):
+                    matched_modules[mod_name] = modules[mod_name]
+                    pattern_match_flags[p] = True
+                    break
 
-        return {name: data for name, data in modules.items() if name in names}
+        unmatched_patterns = [p for p, matched in pattern_match_flags.items() if not matched]
+
+        return matched_modules, unmatched_patterns
+
+    def filter_modules_by_name_and_param(
+        self, modules: dict[str, dict], to_match: dict[str, dict]
+    ) -> tuple[dict[str, dict], dict[str, dict]]:
+        if not to_match:
+            return modules, {}
+
+        filtered = {}
+        unmatched = {}
+
+        for mod_name, expected_data in to_match.items():
+            expected_params = expected_data.get("parameters", {})
+            actual_data = modules.get(mod_name)
+
+            if not actual_data:
+                # Module completely missing
+                unmatched[mod_name] = expected_data
+                continue
+
+            actual_params = actual_data.get("parameters", {})
+            param_mismatches = {}
+
+            for param, expected_val in expected_params.items():
+                actual_val = actual_params.get(param)
+                if actual_val != expected_val:
+                    param_mismatches[param] = {
+                        "expected": expected_val,
+                        "actual": actual_val if actual_val is not None else "<missing>",
+                    }
+
+            if param_mismatches:
+                unmatched[mod_name] = {"parameters": param_mismatches}
+            else:
+                filtered[mod_name] = actual_data
+
+        return filtered, unmatched
 
     def analyze_data(
         self, data: KernelModuleDataModel, args: Optional[KernelModuleAnalyzerArgs] = None
@@ -70,32 +112,74 @@ def analyze_data(
         if not args:
             args = KernelModuleAnalyzerArgs()
 
-        self.result.message = "Kernel modules analyzed"
         self.result.status = ExecutionStatus.OK
-        filtered_modules = {}
+
         if args.regex_match:
             try:
-                filtered_modules = self.filter_modules_by_pattern(
-                    data.kernel_modules, args.modules_filter
+                filtered_modules, unmatched_pattern = self.filter_modules_by_pattern(
+                    data.kernel_modules, args.regex_filter
                 )
             except re.error:
                 self._log_event(
                     category=EventCategory.RUNTIME,
                     description="KernelModule regex is invalid",
-                    data=data,
+                    data={"regex_filters": {args.regex_filter}},
                     priority=EventPriority.ERROR,
                 )
                 self.result.message = "Kernel modules failed to match regex"
                 self.result.status = ExecutionStatus.ERROR
                 return self.result
 
-        else:
-            filtered_modules = self.filter_modules_by_name(data.kernel_modules, args.modules_filter)
+            if unmatched_pattern:
+                self._log_event(
+                    category=EventCategory.RUNTIME,
+                    description="KernelModules did not match all patterns",
+                    data={"unmatched_pattern: ": unmatched_pattern},
+                    priority=EventPriority.INFO,
+                )
+                self.result.message = "Kernel modules failed to match every pattern"
+                self.result.status = ExecutionStatus.ERROR
+                return self.result
+
+            self._log_event(
+                category=EventCategory.RUNTIME,
+                description="KernelModules analyzed",
+                data={"filtered_modules": filtered_modules},
+                priority=EventPriority.INFO,
+            )
+            return self.result
+
+        elif args.kernel_modules:
+            filtered_modules, not_matched = self.filter_modules_by_name_and_param(
+                data.kernel_modules, args.kernel_modules
+            )
+
+            # no modules matched
+            if not filtered_modules and not_matched:
+                self._log_event(
+                    category=EventCategory.RUNTIME,
+                    description="KernelModules: no modules matched",
+                    data=args.kernel_modules,
+                    priority=EventPriority.ERROR,
+                )
+                self.result.message = "Kernel modules not matched"
+                self.result.status = ExecutionStatus.ERROR
+                return self.result
+            # some modules matched
+            elif filtered_modules and not_matched:
+
+                self._log_event(
+                    category=EventCategory.RUNTIME,
+                    description="KernelModules: not all modules matched",
+                    data=not_matched,
+                    priority=EventPriority.ERROR,
+                )
+                self.result.message = "Kernel modules not matched"
+                self.result.status = ExecutionStatus.ERROR
+                return self.result
+            # all modules matched
+            else:
+                self.result.message = "Kernel modules matched"
+                return self.result
 
-        self._log_event(
-            category=EventCategory.RUNTIME,
-            description="KernelModules analyzed",
-            data=filtered_modules,
-            priority=EventPriority.INFO,
-        )
-        return self.result
+            return self.result

nodescraper/plugins/inband/kernel_module/kernel_module_collector.py

@@ -40,9 +40,9 @@ def parse_proc_modules(self, output):
         modules = {}
         for line in output.strip().splitlines():
             parts = line.split()
-            if len(parts) < 6:
+            if not parts:
                 continue
-            name, size, instances, deps, state, offset = parts[:6]
+            name = parts[0]
             modules[name] = {
                 "parameters": {},
             }
@@ -89,26 +89,18 @@ def collect_data(
             tuple[TaskResult, KernelModuleDataModel | None]: tuple containing the task result and kernel data model or None if not found.
         """
         kernel_modules = {}
+        km_data: KernelModuleDataModel | None = None
         if self.system_info.os_family == OSFamily.WINDOWS:
             res = self._run_sut_cmd("wmic os get Version /Value")
             if res.exit_code == 0:
-                kernel_modules = [line for line in res.stdout.splitlines() if "Version=" in line][
-                    0
-                ].split("=")[1]
+                for line in res.stdout.splitlines():
+                    if line.startswith("Version="):
+                        version = line.split("=", 1)[1]
+                        kernel_modules = {version: {"parameters": {}}}
+                        break
+
         else:
             kernel_modules, res = self.collect_all_module_info()
-            """
-            for mod, info in kernel_modules.items():
-                print(f"Module: {mod}")
-                for key, val in info.items():
-                    if key == "parameters":
-                        print("  Parameters:")
-                        for pname, pval in val.items():
-                            print(f"    {pname} = {pval}")
-                    else:
-                        print(f"  {key}: {val}")
-                print()
-            """
 
         if not kernel_modules:
             self._log_event(

nodescraper/plugins/inband/kernel_module/kernel_module_data.py

@@ -29,5 +29,5 @@
 
 class KernelModuleDataModel(DataModel):
     kernel_modules: dict
-    modules_filter: list[str] = ["amd"]
+    regex_filter: list[str] = ["amd"]
     regex_match: bool = True

test/unit/plugin/test_kernel_module_analyzer.py

@@ -0,0 +1,137 @@
+import pytest
+
+from nodescraper.enums.eventcategory import EventCategory
+from nodescraper.enums.executionstatus import ExecutionStatus
+from nodescraper.models.systeminfo import OSFamily
+from nodescraper.plugins.inband.kernel_module.analyzer_args import (
+    KernelModuleAnalyzerArgs,
+)
+from nodescraper.plugins.inband.kernel_module.kernel_module_analyzer import (
+    KernelModuleAnalyzer,
+)
+from nodescraper.plugins.inband.kernel_module.kernel_module_data import (
+    KernelModuleDataModel,
+)
+
+
+@pytest.fixture
+def sample_modules():
+    return {
+        "modA": {"parameters": {"p": 1}},
+        "otherMod": {"parameters": {"p": 2}},
+        "TESTmod": {"parameters": {"p": 3}},
+        "amdABC": {"parameters": {"p": 3}},
+    }
+
+
+@pytest.fixture
+def data_model(sample_modules):
+    return KernelModuleDataModel(kernel_modules=sample_modules)
+
+
+@pytest.fixture
+def analyzer(system_info):
+    # ensure no SystemCompatibilityError
+    system_info.os_family = OSFamily.LINUX
+    return KernelModuleAnalyzer(system_info=system_info)
+
+
+def test_filter_modules_by_pattern_none(sample_modules, analyzer):
+    matched, unmatched = analyzer.filter_modules_by_pattern(sample_modules, None)
+    assert matched == sample_modules
+    assert unmatched == []
+
+
+def test_filter_modules_by_pattern_strict(sample_modules, analyzer):
+    matched, unmatched = analyzer.filter_modules_by_pattern(sample_modules, [r"mod$"])
+    assert set(matched) == {"otherMod", "TESTmod"}
+    assert unmatched == []
+
+
+def test_filter_modules_by_pattern_unmatched(sample_modules, analyzer):
+    # pattern "foo" never matches
+    matched, unmatched = analyzer.filter_modules_by_pattern(sample_modules, ["foo"])
+    assert matched == {}
+    assert unmatched == ["foo"]
+
+
+def test_filter_name_and_param_all_match(sample_modules, analyzer):
+    to_match = {"modA": {"parameters": {"p": 1}}}
+    matched, unmatched = analyzer.filter_modules_by_name_and_param(sample_modules, to_match)
+    assert matched == {"modA": sample_modules["modA"]}
+    assert unmatched == {}
+
+
+def test_filter_name_and_param_param_mismatch(sample_modules, analyzer):
+    to_match = {"modA": {"parameters": {"p": 999}}}
+    matched, unmatched = analyzer.filter_modules_by_name_and_param(sample_modules, to_match)
+    assert matched == {}
+    assert "modA" in unmatched
+    assert "p" in unmatched["modA"]["parameters"]
+
+
+def test_filter_name_and_param_missing_module(sample_modules, analyzer):
+    to_match = {"bogus": {"parameters": {"x": 1}}}
+    matched, unmatched = analyzer.filter_modules_by_name_and_param(sample_modules, to_match)
+    assert matched == {}
+    assert "bogus" in unmatched
+
+
+def test_analyze_data_default(data_model, analyzer):
+    result = analyzer.analyze_data(data_model, None)
+    assert result.status == ExecutionStatus.OK
+
+
+def test_analyze_data_regex_success(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(regex_match=True, regex_filter=["^TESTmod$"])
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status == ExecutionStatus.OK
+    ev = result.events[0]
+    assert ev.description == "KernelModules analyzed"
+    # only TESTmod should have been kept
+    fm = ev.data["filtered_modules"]
+    assert set(fm) == {"TESTmod"}
+
+
+def test_analyze_data_regex_invalid_pattern(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(regex_match=True, regex_filter=["*invalid"])
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status in (ExecutionStatus.ERROR, ExecutionStatus.EXECUTION_FAILURE)
+    assert any(EventCategory.RUNTIME.value in ev.category for ev in result.events)
+
+
+def test_analyze_data_regex_unmatched_patterns(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(regex_match=True, regex_filter=["modA", "nope"])
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert any(ev.description == "KernelModules did not match all patterns" for ev in result.events)
+
+
+def test_analyze_data_name_only_success(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(
+        regex_match=False, kernel_modules={"modA": {"parameters": {"p": 1}}}
+    )
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status == ExecutionStatus.OK
+    assert result.message == "Kernel modules matched"
+    assert result.events == []
+
+
+def test_analyze_data_name_only_no_match(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(regex_match=False, kernel_modules={"XYZ": {"parameters": {}}})
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert any("no modules matched" in ev.description.lower() for ev in result.events)
+
+
+def test_analyze_data_name_only_partial_match(data_model, analyzer):
+    args = KernelModuleAnalyzerArgs(
+        regex_match=False,
+        kernel_modules={
+            "modA": {"parameters": {"p": 1}},
+            "otherMod": {"parameters": {"wrong": 0}},
+        },
+    )
+    result = analyzer.analyze_data(data_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert any("not all modules matched" in ev.description.lower() for ev in result.events)