Prevent Linux's list_tasks from yielding completely broken process instances

volatility3/framework/plugins/linux/pslist.py

@@ -1,6 +1,7 @@
 # This file is Copyright 2021 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
+import logging
 import datetime
 import dataclasses
 import contextlib
@@ -15,6 +16,8 @@ from volatility3.framework.symbols.linux.extensions import elf
 from volatility3.plugins import timeliner
 from volatility3.plugins.linux import elfs
 
+vollog = logging.getLogger(__name__)
+
 
 @dataclasses.dataclass
 class TaskFields:
@@ -112,6 +115,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             A TaskFields object with the fields to show in the plugin output.
         """
         name = utility.array_to_string(task.comm)
+
         if decorate_comm:
             if task.is_kernel_thread:
                 name = f"[{name}]"
@@ -250,6 +254,14 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
 
         # Note that the init_task itself is not yielded, since "ps" also never shows it.
         for task in init_task.tasks:
+            # the task list is often smeared in samples, espeically towards the end
+            # this stops the processing
+            if not task.has_valid_name_and_pid():
+                vollog.debug(
+                    f"Found an smeared/invalid task at offset {task.vol.offset:#x}"
+                )
+                break
+
             if filter_func(task):
                 continue
 

volatility3/framework/symbols/linux/extensions/__init__.py

@@ -307,6 +307,22 @@ class module(generic.GenericIntelProcess):
 
 
 class task_struct(generic.GenericIntelProcess):
+    def has_valid_name_and_pid(self) -> bool:
+        """
+        Ensures the pid and name for this process are sane
+        When these are broken it is a sure sign of smear
+        during process enumeration
+        """
+        try:
+            pid = self.pid
+            name = utility.array_to_string(self.comm)
+        except exceptions.InvalidAddressException:
+            return False
+
+        # ensure the pid within the bounds of a signed int
+        # and that we have something of a name
+        return (0 < pid < 2147483647) and len(name) > 0
+
     def add_process_layer(
         self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
     ) -> Optional[str]: