Add dumping dirty pages to Linux Malfind

volatility3/framework/plugins/linux/malware/malfind.py

@@ -18,7 +18,7 @@ class Malfind(interfaces.plugins.PluginInterface):
     """Lists process memory ranges that potentially contain injected code."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 3)
+    _version = (1, 0, 4)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -37,6 +37,16 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 element_type=int,
                 optional=True,
             ),
+            requirements.IntRequirement(
+                name="dump-size",
+                description="Amount of bytes to dump for each dirty region/page found - Default 64 bytes",
+                optional=True,
+            ),
+            requirements.BooleanRequirement(
+                name="dump-page",
+                description="Dump each dirty page and content - Default off",
+                optional=True,
+            ),
         ]
 
     def _list_injections(
@@ -51,14 +61,36 @@ def _list_injections(
 
         proc_layer = self.context.layers[proc_layer_name]
 
+        dump_size = self.config.get("dump-size", None) or 64
+
+        # Dumping page defaults to off, as in case a whole r-xp region is dirty
+        # this would likely dump 1000's of pages which might not always be wise nor necessary
+
+        dump_page = self.config.get("dump-page") or False
+
         for vma in task.mm.get_vma_iter():
             vma_name = vma.get_name(self.context, task)
             vollog.debug(
                 f"Injections : processing PID {task.pid} : VMA {vma_name} : {hex(vma.vm_start)}-{hex(vma.vm_end)}"
             )
+
+            # If is_suspicious returns true, this means at least one page
+            # in the region is dirty. If dump_page is true, then we dump
+            # all dirty pages
+
             if vma.is_suspicious(proc_layer) and vma_name != "[vdso]":
-                data = proc_layer.read(vma.vm_start, 64, pad=True)
-                yield vma, vma_name, data
+                malicious_pages = vma.get_malicious_pages(proc_layer)
+                offset = 0
+                if dump_page:
+                    # Dumping each dirty page
+                    for page_addr in malicious_pages:
+                        offset = page_addr - vma.vm_start
+                        data = proc_layer.read(page_addr, dump_size, pad=True)
+                        yield vma, f"{vma_name}, page address: {page_addr:#x}, offset: {offset:#x}", data, offset
+                else:
+                    # Original behaviour - Dump the start of the region (not necessarily matching the dirty page)
+                    data = proc_layer.read(vma.vm_start, dump_size, pad=True)
+                    yield vma, vma_name, data, offset
 
     def _generator(self, tasks):
         # determine if we're on a 32 or 64 bit kernel
@@ -70,13 +102,15 @@ def _generator(self, tasks):
         for task in tasks:
             process_name = utility.array_to_string(task.comm)
 
-            for vma, vma_name, data in self._list_injections(task):
+            for vma, vma_name, data, offset in self._list_injections(task):
                 if is_32bit_arch:
                     architecture = "intel"
                 else:
                     architecture = "intel64"
 
-                disasm = renderers.Disassembly(data, vma.vm_start, architecture)
+                disasm = renderers.Disassembly(
+                    data, vma.vm_start + offset, architecture
+                )
 
                 yield (
                     0,

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

@@ -1273,6 +1273,42 @@ def get_name(self, context, task) -> Optional[str]:
         except exceptions.InvalidAddressException:
             return None
 
+    def get_malicious_pages(self, proclayer) -> List[int]:
+        """Identifies and returns a list of potentially malicious memory pages.
+
+        A page is considered malicious if it is:
+            - Executable (protection flags match 'r-x')
+            - Dirty (modified since process start, according to proclayer.is_dirty())
+
+        Args:
+            proclayer: The process's memory layer
+
+        Returns:
+            List[int]: A list of virtual addresses for pages flagged as potentially malicious.
+        """
+
+        malicious_pages = []
+        flags_str = self.get_protection()
+
+        if (
+            proclayer
+            and "r-x" in flags_str
+            and self.vm_file.dereference().vol.offset != 0
+        ):
+            for i in range(self.vm_start, self.vm_end, proclayer.page_size):
+                try:
+                    if proclayer.is_dirty(i):
+                        vollog.debug(f"Found malicious (dirty+exec) page at {hex(i)} !")
+                        malicious_pages.append(i)
+                except (
+                    exceptions.PagedInvalidAddressException,
+                    exceptions.InvalidAddressException,
+                ) as excp:
+                    vollog.debug(f"Unable to translate address {hex(i)} : {excp}")
+                    # Abort as it is likely that other addresses in the same range will also fail
+                    break
+        return malicious_pages
+
     # used by malfind
     def is_suspicious(self, proclayer=None):
         ret = False
@@ -1288,7 +1324,7 @@ def is_suspicious(self, proclayer=None):
                 try:
                     if proclayer.is_dirty(i):
                         vollog.warning(
-                            f"Found malicious (dirty+exec) page at {hex(i)} !"
+                            f"Found malicious page(s) inside (dirty+exec) region {hex(self.vm_start)} !"
                         )
                         # We do not attempt to find other dirty+exec pages once we have found one
                         ret = True