pre-process module triaging to improve readability

volatility3/framework/plugins/linux/modxview.py

@@ -82,7 +82,8 @@ def run_modules_scanners(
         kernel_name: str,
         run_hidden_modules: bool = True,
     ) -> Dict[str, List[extensions.module]]:
-        """Run module scanning plugins and aggregate the results.
+        """Run module scanning plugins and aggregate the results. It is designed
+        to not operate any inter-plugin results triage.
 
         Args:
             run_hidden_modules: specify if the hidden_modules plugin should be run
@@ -128,46 +129,50 @@ def run_modules_scanners(
     def _generator(self):
         kernel_name = self.config["kernel"]
         run_results = self.run_modules_scanners(self.context, kernel_name)
-        modules_offsets = {}
-        for key in ["lsmod", "check_modules", "hidden_modules"]:
-            modules_offsets[key] = set(module.vol.offset for module in run_results[key])
-
-        seen_addresses = set()
-        for modules_list in run_results.values():
-            for module in modules_list:
-                if module.vol.offset in seen_addresses:
-                    continue
-                seen_addresses.add(module.vol.offset)
+        aggregated_modules = {}
+        # We want to be explicit on the plugins results we are interested in
+        for plugin_name in ["lsmod", "check_modules", "hidden_modules"]:
+            # Iterate over each recovered module
+            for module in run_results[plugin_name]:
+                # Use offsets as unique keys, whether a module
+                # appears in many plugin runs or not
+                if aggregated_modules.get(module.vol.offset):
+                    # Append the plugin to the list of originating plugins
+                    aggregated_modules[module.vol.offset][1].append(plugin_name)
+                else:
+                    aggregated_modules[module.vol.offset] = (module, [plugin_name])
 
-                if self.config.get("plain_taints"):
-                    taints = tainting.Tainting.get_taints_as_plain_string(
+        for module_offset, (module, originating_plugins) in aggregated_modules.items():
+            # Tainting parsing capabilities applied to the module
+            if self.config.get("plain_taints"):
+                taints = tainting.Tainting.get_taints_as_plain_string(
+                    self.context,
+                    kernel_name,
+                    module.taints,
+                    True,
+                )
+            else:
+                taints = ",".join(
+                    tainting.Tainting.get_taints_parsed(
                         self.context,
                         kernel_name,
                         module.taints,
                         True,
                     )
-                else:
-                    taints = ",".join(
-                        tainting.Tainting.get_taints_parsed(
-                            self.context,
-                            kernel_name,
-                            module.taints,
-                            True,
-                        )
-                    )
-
-                yield (
-                    0,
-                    (
-                        module.get_name() or NotAvailableValue(),
-                        format_hints.Hex(module.vol.offset),
-                        module.vol.offset in modules_offsets["lsmod"],
-                        module.vol.offset in modules_offsets["check_modules"],
-                        module.vol.offset in modules_offsets["hidden_modules"],
-                        taints or NotAvailableValue(),
-                    ),
                 )
 
+            yield (
+                0,
+                (
+                    module.get_name() or NotAvailableValue(),
+                    format_hints.Hex(module_offset),
+                    "lsmod" in originating_plugins,
+                    "check_modules" in originating_plugins,
+                    "hidden_modules" in originating_plugins,
+                    taints or NotAvailableValue(),
+                ),
+            )
+
     def run(self):
         columns = [
             ("Name", str),