Merge pull request #2037 from volatilityfoundation/release/v2.28.2

Release/v2.28.2

.github/workflows/build-pyinstaller.yml

@@ -28,7 +28,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install pyinstaller
-          pip install -e .[full,cloud,arrow]
+          pip install -e .[full,cloud]
 
       - name: Pyinstall executable
         run: |

doc/source/basics.rst

@@ -15,8 +15,8 @@ Memory layers
 
 A memory layer is a body of data that can be accessed by requesting data at a specific address.  At its lowest level
 this data is stored on a phyiscal medium (RAM) and very early computers addressed locations in memory directly.  However,
-as the size of memory increased and it became more difficult to manage memory most architectures moved to a "paged" model 
-of memory, where the available memory is cut into specific fixed-sized pages.  To help further, programs can ask for any address 
+as the size of memory increased and it became more difficult to manage memory most architectures moved to a "paged" model
+of memory, where the available memory is cut into specific fixed-sized pages.  To help further, programs can ask for any address
 and the processor will look up their (virtual) address in a map, to find out where the (physical) address that it lives at is,
 in the actual memory of the system.
 
@@ -24,18 +24,18 @@ Volatility can work with these layers as long as it knows the map (so, for examp
 address `9`).  The automagic that runs at the start of every volatility session often locates the kernel's memory map, and creates
 a kernel virtual layer, which allows for kernel addresses to be looked up and the correct data returned.  There can, however, be
 several maps, and in general there is a different map for each process (although a portion of the operating system's memory is
-usually mapped to the same location across all processes).  The maps may take the same address but point to a different part of 
-physical memory.  It also means that two processes could theoretically share memory, both having a virtual address mapped to the 
+usually mapped to the same location across all processes).  The maps may take the same address but point to a different part of
+physical memory.  It also means that two processes could theoretically share memory, both having a virtual address mapped to the
 same physical address.  See the worked example below for more information.
 
 To translate an address on a layer, call :py:meth:`layer.mapping(offset, length, ignore_errors) <volatility3.framework.interfaces.layers.TranslationLayerInterface.mapping>` and it will return a list of chunks without overlap, in order,
-for the requested range.  If a portion cannot be mapped, an exception will be thrown unless `ignore_errors` is true.  Each 
-chunk will contain the original offset of the chunk, the translated offset, the original size and the translated size of 
+for the requested range.  If a portion cannot be mapped, an exception will be thrown unless `ignore_errors` is true.  Each
+chunk will contain the original offset of the chunk, the translated offset, the original size and the translated size of
 the chunk, as well as the lower layer the chunk lives within.
 
 Worked example
 ^^^^^^^^^^^^^^
-    
+
 The operating system and two programs may all appear to have access to  all of physical memory, but actually the maps they each have
 mean they each see something different:
 
@@ -65,12 +65,12 @@ is a permissions model for Intel addressing which is not discussed further here)
 
 In Volatility 3 mappings are represented by a directed graph of layers, whose end nodes are
 :py:class:`DataLayers <volatility3.framework.interfaces.layers.DataLayerInterface>` and whose internal nodes are :py:class:`TranslationLayers <volatility3.framework.interfaces.layers.TranslationLayerInterface>`.
-In this way, a raw memory image in the LiME file format and a page file can be combined to form a single Intel virtual 
-memory layer.  When requesting addresses from the Intel layer, it will use the Intel memory mapping algorithm, along 
+In this way, a raw memory image in the LiME file format and a page file can be combined to form a single Intel virtual
+memory layer.  When requesting addresses from the Intel layer, it will use the Intel memory mapping algorithm, along
 with the address of the directory table base or page table map, to translate that
 address into a physical address, which will then either be directed towards the swap layer or the LiME layer.  Should it
-be directed towards the LiME layer, the LiME file format algorithm will translate the new address to determine where 
-within the file the data is stored.  When the :py:meth:`layer.read() <volatility3.framework.interfaces.layers.TranslationLayerInterface.read>` 
+be directed towards the LiME layer, the LiME file format algorithm will translate the new address to determine where
+within the file the data is stored.  When the :py:meth:`layer.read() <volatility3.framework.interfaces.layers.TranslationLayerInterface.read>`
 method is called, the translation is done automatically and the correct data gathered and combined.
 
 .. note:: Volatility 2 had a similar concept, called address spaces, but these could only stack linearly one on top of another.
@@ -150,6 +150,9 @@ a table, or inserted into a database like Elastic Search and trawled using an ex
 The renderers only need to know how to process very basic types (booleans, strings, integers, bytes) and a few additional specific
 ones (disassembly and various absent values).
 
+Renderers can also be added to volatility automatically.  There is an additional arrow/parquet format renderer available (but requires
+the pyarrow dependency to be installed), but this is not shipped with the EXE version because it doubles the size of the executable.
+
 Configuration Tree
 ------------------
 

doc/source/getting-started-mac-tutorial.rst

@@ -1,6 +1,12 @@
 macOS Tutorial
 ==============
 
+.. warning::
+
+    As of the Volatility 3 parity release, macOS analysis support is no longer actively maintained.
+    The existing macOS plugins remain available but may not receive future updates or bug fixes.
+    For more details, see the `official announcement <https://volatilityfoundation.org/announcing-the-official-parity-release-of-volatility-3/>`_.
+
 This guide will give you a brief overview of how volatility3 works as well as a demonstration of several of the plugins available in the suite.
 
 Acquiring memory

test/test_cli.py

@@ -0,0 +1,94 @@
+# volatility3 command line tests
+#
+# These require no memory image, but the conftest --volatility option must
+# still be supplied for collection to succeed.
+
+#
+# IMPORTS
+#
+
+import argparse
+from urllib.request import urlopen
+
+import pytest
+
+from volatility3.cli import CommandLine
+from volatility3.framework import contexts, interfaces
+from volatility3.framework.configuration import requirements
+
+
+#
+# HELPER CLASSES AND FUNCTIONS
+#
+
+
+class URIConfigurable(interfaces.configuration.ConfigurableInterface):
+    """A configurable offering nothing but a single URIRequirement."""
+
+    @classmethod
+    def get_requirements(cls):
+        return [
+            requirements.URIRequirement(
+                name="testfile", description="A file to be located"
+            )
+        ]
+
+
+def populate_uri_requirement(value: str):
+    """Run the given value through the command line's config population.
+
+    Args:
+        value: The value as it would arrive from the command line
+    Returns:
+        The value as it was stored in the context's configuration
+    """
+
+    context = contexts.Context()
+    CommandLine().populate_config(
+        context,
+        {"testplugin": URIConfigurable},
+        argparse.Namespace(testfile=value),
+        "plugins.TestPlugin",
+    )
+
+    return context.config["plugins.TestPlugin.testfile"]
+
+
+#
+# TESTS
+#
+
+
+def test_uri_requirement_path_becomes_an_openable_url(tmp_path):
+    """A filesystem path must become a URL the framework can actually open.
+
+    The URL used to be assembled by hand, which left an empty authority
+    section in place on platforms where pathname2url already returns a
+    leading "///".
+    """
+
+    testfile = tmp_path / "memory dump.raw"
+    testfile.write_bytes(b"volatility")
+
+    location = populate_uri_requirement(str(testfile))
+
+    assert location == testfile.as_uri()
+    with urlopen(location) as fp:
+        assert fp.read() == b"volatility"
+
+
+def test_uri_requirement_leaves_a_url_alone(tmp_path):
+    """A value that already carries a scheme must be passed through as is."""
+
+    testfile = tmp_path / "memory.raw"
+    testfile.write_bytes(b"volatility")
+    url = testfile.as_uri()
+
+    assert populate_uri_requirement(url) == url
+
+
+def test_uri_requirement_rejects_a_missing_file(tmp_path):
+    """A path that does not exist must be reported rather than converted."""
+
+    with pytest.raises(FileNotFoundError):
+        populate_uri_requirement(str(tmp_path / "absent.raw"))

volatility3/cli/__init__.py

@@ -17,11 +17,12 @@ import io
 import json
 import logging
 import os
+import pathlib
 import sys
 import tempfile
 import traceback
 from typing import Any, Dict, List, Optional, Tuple, Type, Union
-from urllib import parse, request
+from urllib import parse
 
 try:
     import argcomplete
@@ -30,11 +31,10 @@ try:
 except ImportError:
     HAS_ARGCOMPLETE = False
 
-from volatility3.cli import text_filter
 import volatility3.plugins
 import volatility3.symbols
 from volatility3 import framework
-from volatility3.cli import text_renderer, volargparse
+from volatility3.cli import text_filter, text_renderer, volargparse
 from volatility3.framework import (
     automagic,
     configuration,
@@ -380,6 +380,17 @@ class CommandLine:
             )
             self.populate_requirements_argparse(plugin_parser, plugin_list[plugin])
 
+        # One last pass to get the renderer after we've loaded up plugins,
+        # so we can determine whether to show the banner on normal output or not...
+        known_args = [arg for arg in sys.argv[1:] if arg != "--help" and arg != "-h"]
+        partial_args, _ = parser.parse_known_args(known_args)
+
+        # Display banner - redirect to stderr if using structured output
+        banner_output = sys.stdout
+        if renderers[partial_args.renderer].structured_output:
+            banner_output = sys.stderr
+        banner_output.write(f"Volatility 3 Framework {constants.PACKAGE_VERSION}\n")
+
         ###
         # PASS TO UI
         ###
@@ -392,12 +403,6 @@ class CommandLine:
             argcomplete.autocomplete(parser)
         args = parser.parse_args()
 
-        # Display banner - redirect to stderr if using structured output
-        banner_output = sys.stdout
-        if renderers[args.renderer].structured_output:
-            banner_output = sys.stderr
-        banner_output.write(f"Volatility 3 Framework {constants.PACKAGE_VERSION}\n")
-
         if args.plugin is None:
             parser.error(
                 f"Please select a plugin to run (see '{self.CLI_NAME} --help' for options"
@@ -489,7 +494,7 @@ class CommandLine:
                 )
                 args.save_config = "config.json"
             if args.save_config:
-                vollog.debug("Writing out configuration data to {args.save_config}")
+                vollog.debug(f"Writing out configuration data to {args.save_config}")
                 if os.path.exists(os.path.abspath(args.save_config)):
                     parser.error(
                         f"Cannot write configuration: file {args.save_config} already exists"
@@ -739,7 +744,10 @@ class CommandLine:
                                     raise FileNotFoundError(
                                         f"Non-existent file {value} passed to URIRequirement"
                                     )
-                                value = f"file://{request.pathname2url(os.path.abspath(value))}"
+                                # as_uri builds a correctly formed file URL on
+                                # every platform, whereas prefixing the scheme
+                                # by hand leaves too many slashes on Windows
+                                value = pathlib.Path(os.path.abspath(value)).as_uri()
                     if isinstance(requirement, requirements.ListRequirement):
                         if not isinstance(value, list):
                             raise TypeError(

volatility3/cli/text_renderer.py

@@ -3,6 +3,7 @@
 #
 import csv
 import datetime
+import itertools
 import json
 import logging
 import random
@@ -623,3 +624,121 @@ class JsonLinesRenderer(JsonRenderer):
         for line in result:
             outfd.write(json.dumps(line, sort_keys=True))
             outfd.write("\n")
+
+
+class MermaidRenderer(CLIRenderer):
+    _type_renderers = {
+        format_hints.Bin: optional(lambda x: f"0b{x:b}"),
+        format_hints.Hex: optional(lambda x: f"0x{x:x}"),
+        format_hints.HexBytes: optional(hex_bytes_as_text),
+        format_hints.MultiTypeData: optional(multitypedata_as_text),
+        interfaces.renderers.Disassembly: optional(display_disassembly),
+        bytes: optional(lambda x: " ".join([f"{b:02x}" for b in x])),
+        datetime.datetime: optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
+        "default": optional(lambda x: f"{x}"),
+    }
+
+    name = "mermaid"
+    structured_output = True
+
+    @staticmethod
+    def _mermaid_label(text: str) -> str:
+        """Escape a value for use inside a Mermaid node label (``["..."]``).
+
+        Double quotes terminate the label, so they must be replaced with the
+        Mermaid-supported entity. Newlines inside cell renderings are
+        converted to ``<br>`` so each row remains a single Mermaid node.
+        """
+        return text.replace('"', "&quot;").replace("\n", "<br>")
+
+    def get_render_options(self):
+        pass
+
+    def render(self, grid: interfaces.renderers.TreeGrid) -> None:
+        """Render the TreeGrid as a Mermaid ``graph TD`` flowchart.
+
+        The renderer is plugin-agnostic: it derives the parent/child
+        relationship from each node's ``path_depth`` in the grid, rather
+        than from any particular column (such as PID/PPID). This means
+        any tree-shaped plugin output -- pstree, vadwalk, handles tree,
+        future plugins -- renders without modification.
+
+        The algorithm maintains a parent stack while walking the rows in
+        traversal order:
+
+        * descending one or more levels pushes the previously-emitted
+          node onto the stack (once per level descended) so it becomes
+          the current parent;
+        * ascending pops the same number of levels off the stack;
+        * the top of the stack is always the parent of the next emitted
+          node, or empty for a root-level node.
+
+        Args:
+            grid: The TreeGrid object to render
+        """
+        outfd = sys.stdout
+
+        sys.stderr.write("Formatting...\n")
+
+        def format_row(node: interfaces.renderers.TreeNode) -> str:
+            """Build a Mermaid node label from every column of ``node``."""
+            cells = []
+            for column_index, column in enumerate(grid.columns):
+                renderer = self._type_renderers.get(
+                    column.type, self._type_renderers["default"]
+                )
+                value = renderer(node.values[column_index])
+                cells.append(f"{column.name}:{self._mermaid_label(value)}")
+            return "<br>".join(cells)
+
+        rows: List[Tuple[int, str]] = []
+
+        def visitor(
+            node: interfaces.renderers.TreeNode,
+            accumulator: List[Tuple[int, str]],
+        ) -> List[Tuple[int, str]]:
+            accumulator.append((node.path_depth, format_row(node)))
+            return accumulator
+
+        if not grid.populated:
+            grid.populate(visitor, rows)
+        else:
+            grid.visit(node=None, function=visitor, initial_accumulator=rows)
+
+        # Stable, unique per-node IDs. We never reuse a column value (e.g.
+        # PID) because (a) PID is not guaranteed unique across a TreeGrid,
+        # (b) it is plugin-specific, and (c) Mermaid IDs must avoid
+        # characters like parentheses that may appear in column data.
+        node_ids = itertools.count(1)
+
+        parent_stack: List[str] = []
+        prev_depth = 0
+        prev_id: Optional[str] = None
+
+        lines: List[str] = ["graph TD"]
+        for depth, label in rows:
+            node_id = f"n{next(node_ids)}"
+            if prev_id is not None:
+                if depth > prev_depth:
+                    # Descended one or more levels. Push prev_id once per
+                    # level so subsequent pops align even when the tree
+                    # skips levels (e.g. depth 1 -> depth 3).
+                    for _ in range(depth - prev_depth):
+                        parent_stack.append(prev_id)
+                elif depth < prev_depth:
+                    for _ in range(prev_depth - depth):
+                        if parent_stack:
+                            parent_stack.pop()
+                # depth == prev_depth: sibling, keep the same parent
+
+            if parent_stack:
+                parent = parent_stack[-1]
+                lines.append(f'\t{parent} --> {node_id}["{label}"]')
+            else:
+                # Root-level node: declare it on its own.
+                lines.append(f'\t{node_id}["{label}"]')
+
+            prev_id = node_id
+            prev_depth = depth
+
+        outfd.write("\n".join(lines) + "\n")

volatility3/cli/volshell/__init__.py

@@ -377,7 +377,7 @@ class VolShell(cli.CommandLine):
                 )
                 args.save_config = "config.json"
             if args.save_config:
-                vollog.debug("Writing out configuration data to {args.save_config}")
+                vollog.debug(f"Writing out configuration data to {args.save_config}")
                 if os.path.exists(os.path.abspath(args.save_config)):
                     parser.error(
                         f"Cannot write configuration: file {args.save_config} already exists"

volatility3/framework/automagic/windows.py

@@ -29,15 +29,63 @@ The self-referential indices for older versions of windows are listed below:
 
 import logging
 import struct
-from typing import Generator, Iterable, List, Optional, Tuple, Type
+from typing import Generator, Iterable, List, Optional, Tuple, Type, Union
 
-from volatility3.framework import constants, interfaces, layers
+from volatility3.framework import constants, exceptions, interfaces, layers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import intel
 
 vollog = logging.getLogger(__name__)
 
 
+class Intel32LayerCheck:
+    # These addresses are at a fixed location:
+    # "The KUSER_SHARED_DATA structure is a single page (4096 bytes) in memory
+    # that is mapped at a fixed, hardcoded address in both kernel and user side of VAS."
+    # See: https://www.microsoft.com/en-us/msrc/blog/2022/04/randomizing-the-kuser_shared_data-structure-on-windows
+    KUSER_USER_SPACE_ADDR = 0x7FFE0000
+    KUSER_KERNEL_SPACE_ADDR = 0xFFDF0000
+    # This field offset did not change across Windows versions.
+    # Instead of storing a complete struct definition only for this check,
+    # define it locally here.
+    KUSER_SHARED_DATA_NTMAJOR_OFF = 0x26C
+    NT_MAJOR_VALIDS = [3, 4, 5, 6, 10]
+
+    @classmethod
+    def check(cls, layer: intel.Intel):
+        """Generates a single response of True or False depending on whether the space is a valid Windows AS"""
+        # This constraint verifies that _KUSER_SHARED_DATA is shared
+        # between user and kernel address spaces.
+        kaddr = uaddr = None
+        try:
+            kaddr = layer._translate(cls.KUSER_KERNEL_SPACE_ADDR)[0]
+            uaddr = layer._translate(cls.KUSER_USER_SPACE_ADDR)[0]
+            if kaddr != 0 and kaddr == uaddr:
+                return True
+        except (
+            exceptions.PagedInvalidAddressException,
+            exceptions.InvalidAddressException,
+        ):
+            # Translation failed, most likely because of UADDR
+            pass
+
+        # Validate by reading the _KUSER_SHARED_DATA.NtMajorVersion field
+        if kaddr is not None:
+            data = layer.read(
+                cls.KUSER_KERNEL_SPACE_ADDR + cls.KUSER_SHARED_DATA_NTMAJOR_OFF,
+                4,
+                pad=True,
+            )
+            if struct.unpack("<I", data)[0] in cls.NT_MAJOR_VALIDS:
+                return True
+
+        return False
+
+
+class Intel64LayerCheck(Intel32LayerCheck):
+    KUSER_KERNEL_SPACE_ADDR = 0xFFFFF78000000000
+
+
 class DtbSelfReferential:
     """A generic DTB test which looks for a self-referential pointer at *any*
     index within the page."""
@@ -45,12 +93,14 @@ class DtbSelfReferential:
     def __init__(
         self,
         layer_type: Type[layers.intel.Intel],
+        layer_check: Union[Intel32LayerCheck.check, Intel64LayerCheck.check],
         ptr_struct: str,
         mask: int,
         valid_range: Iterable[int],
         reserved_bits: int,
     ) -> None:
         self.layer_type = layer_type
+        self.layer_check = layer_check
         self.ptr_struct = ptr_struct
         self.ptr_size = struct.calcsize(ptr_struct)
         self.mask = mask
@@ -92,6 +142,7 @@ class DtbSelfRef32bit(DtbSelfReferential):
     def __init__(self):
         super().__init__(
             layer_type=layers.intel.WindowsIntel,
+            layer_check=Intel32LayerCheck.check,
             ptr_struct="I",
             mask=0xFFFFF000,
             valid_range=[0x300],
@@ -103,6 +154,7 @@ class DtbSelfRef64bit(DtbSelfReferential):
     def __init__(self) -> None:
         super().__init__(
             layer_type=layers.intel.WindowsIntel32e,
+            layer_check=Intel64LayerCheck.check,
             ptr_struct="Q",
             mask=0x3FFFFFFFFFF000,
             valid_range=range(0x100, 0x1FF),
@@ -114,6 +166,7 @@ class DtbSelfRef64bitOldWindows(DtbSelfReferential):
     def __init__(self) -> None:
         super().__init__(
             layer_type=layers.intel.WindowsIntel32e,
+            layer_check=Intel64LayerCheck.check,
             ptr_struct="Q",
             mask=0x3FFFFFFFFFF000,
             valid_range=[0x1ED],
@@ -125,6 +178,7 @@ class DtbSelfRefPae(DtbSelfReferential):
     def __init__(self) -> None:
         super().__init__(
             layer_type=layers.intel.WindowsIntelPAE,
+            layer_check=Intel32LayerCheck.check,
             ptr_struct="Q",
             valid_range=[0x3],
             mask=0x3FFFFFFFFFF000,
@@ -199,13 +253,24 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
         (
             "Detecting Self-referential pointer for recent windows",
             [DtbSelfRef64bit()],
-            [(0x150000, 0x150000), (0x550000, 0x1A0000), (0x900000, 0x100000)],
+            [
+                (0x150000, 0x150000),
+                (0x550000, 0x1A0000),
+                (0x900000, 0x100000),
+            ],
         ),
         (
             "Older windows fixed location self-referential pointers",
             [DtbSelfRefPae(), DtbSelfRef32bit(), DtbSelfRef64bitOldWindows()],
             [(0x30000, 0x1000000)],
         ),
+        (
+            "Very large memory with high DTBs (slow)",
+            [DtbSelfRef64bit()],
+            [
+                (0xA00000, 0x5000000),
+            ],
+        ),
     ]
 
     @classmethod
@@ -306,18 +371,6 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
                     )
                 return max_ptr
 
-            def page_table_is_dummy(page_table, ptr_size: int):
-                """Verify that a page table has at least 12 valid pointers"""
-                valid_pointers = 0
-                for _ in get_valid_page_table_pointers(page_table, ptr_size):
-                    valid_pointers += 1
-                    # 10 is an arbitrary constant
-                    if valid_pointers >= 10:
-                        # Do not consume the entire generator to enhance performance
-                        return False
-                vollog.debug(f"Found {valid_pointers} valid pointers")
-                return True
-
             hits = sorted(list(hits), key=sort_by_tests)
 
             vollog.debug(f"WindowsIntelStacker hits: {hits}")
@@ -326,13 +379,6 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
                 # Turn the page tables into integers and find the largest one
                 page_table = base_layer.read(page_map_offset, 0x1000)
                 ptr_size = struct.calcsize(test.ptr_struct)
-                # Modern windows can have a dummy page table with only about 2 entries, so sanity check
-                if page_table_is_dummy(page_table, ptr_size):
-                    vollog.debug(
-                        f"DTB {page_map_offset:x} contains less than 12 valid pointers, ignoring"
-                    )
-                    continue
-
                 max_pointer = get_max_pointer(page_table, test, ptr_size)
 
                 if max_pointer <= base_layer.maximum_address:
@@ -351,12 +397,18 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
                             config_path, "page_map_offset"
                         )
                     ] = page_map_offset
-                    layer = test.layer_type(
+                    tmp_layer = test.layer_type(
                         context,
                         config_path=config_path,
                         name=new_layer_name,
                         metadata={"os": "Windows"},
                     )
+                    if not test.layer_check(tmp_layer):
+                        vollog.debug(
+                            f"DTB {page_map_offset:x} failed {test.layer_type.__name__} _KUSER_SHARED_DATA check, ignoring"
+                        )
+                        continue
+                    layer = tmp_layer
                     break
                 else:
                     vollog.debug(

volatility3/framework/constants/_version.py

@@ -1,7 +1,7 @@
 # We use the SemVer 2.0.0 versioning scheme
 VERSION_MAJOR = 2  # Number of releases of the library with a breaking change
 VERSION_MINOR = 28  # Number of changes that only add to the interface
-VERSION_PATCH = 0  # Number of changes that do not change the interface
+VERSION_PATCH = 2  # Number of changes that do not change the interface
 VERSION_SUFFIX = ""
 
 PACKAGE_VERSION = (

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

@@ -2,23 +2,31 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
-from typing import List, Tuple, Optional
 import logging
-from volatility3.framework import interfaces
-from volatility3.framework import renderers, symbols
+from typing import Iterable, List, Tuple, Optional
+from enum import IntEnum
+from volatility3.framework import exceptions, interfaces, renderers, symbols
 from volatility3.framework.configuration import requirements
 from volatility3.framework.objects import utility
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.linux import pslist
+from volatility3.plugins.linux import pslist, proc
 
 vollog = logging.getLogger(__name__)
 
 
+class MaliciousFlags(IntEnum):
+    RWX = 0
+    RX = 1
+    X_DIRTY = 2
+
+
 class Malfind(interfaces.plugins.PluginInterface):
     """Lists process memory ranges that potentially contain injected code."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 4)
+    _version = (1, 1, 0)
+
+    MAX_DUMPSIZE_DEFAULT = 1024 * 1024 * 1024  # 1 Gb
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -31,6 +39,9 @@ class Malfind(interfaces.plugins.PluginInterface):
             requirements.VersionRequirement(
                 name="pslist", component=pslist.PsList, version=(4, 0, 0)
             ),
+            requirements.VersionRequirement(
+                name="proc", component=proc.Maps, version=(1, 0, 3)
+            ),
             requirements.ListRequirement(
                 name="pid",
                 description="Filter on specific process IDs",
@@ -38,66 +49,130 @@ class Malfind(interfaces.plugins.PluginInterface):
                 optional=True,
             ),
             requirements.IntRequirement(
-                name="dump-size",
-                description="Amount of bytes to dump for each dirty region/page found - Default 64 bytes",
+                name="hexdump-size",
+                description="Amount of bytes to show for each region/page found - Default 64 bytes",
                 optional=True,
                 default=64,
             ),
             requirements.BooleanRequirement(
-                name="dump-page",
-                description="Dump each dirty page and content - Default off",
+                name="show-all-dirty-pages",
+                description="Show all dirty pages in a VMA if at least one dirty page is found - Default off",
                 optional=True,
                 default=False,
             ),
+            requirements.BooleanRequirement(
+                name="dump-regions",
+                description="Dump each suspicious memory region in output. All dirty pages will be dumped if --show-all-dirty-pages is enabled.",
+                optional=True,
+                default=False,
+            ),
+            requirements.IntRequirement(
+                name="dump-maxsize",
+                description="Maximum size for dumped memory regions "
+                "(all the bigger regions will be ignored) - Default 1 GB",
+                default=cls.MAX_DUMPSIZE_DEFAULT,
+                optional=True,
+            ),
         ]
 
+    def _get_dirty_pages(self, proc_layer, vma) -> Iterable[Tuple[int, int]]:
+        """Get dirty pages inside the specified VMA.
+
+        Yields:
+            page address and page size
+        """
+        page_addr = vma.vm_start
+        while page_addr < vma.vm_end:
+            try:
+                # We don't want to use the layer's page size by default to handle
+                # large pages (PUD, PMD...) correctly.
+                _, page_size, _ = proc_layer._translate(page_addr)
+                if proc_layer.is_dirty(page_addr):
+                    yield page_addr, page_size
+            except (
+                AttributeError,
+                exceptions.PagedInvalidAddressException,
+                exceptions.InvalidAddressException,
+            ):
+                page_size = proc_layer.page_size
+
+            page_addr += page_size
+
+    def _is_suspicious(self, proc_layer, vma) -> Optional[Tuple[int, MaliciousFlags]]:
+        """Determine if a VMA is suspicious based on any of the following criterias:
+        - RWX
+        - RX
+        - X + DIRTY
+
+        Returns:
+            (suspicious page address, suspicious page size, malicious flag) or None
+        """
+        flags_str = vma.get_protection()
+
+        if flags_str == "rwx":
+            return vma.vm_start, vma.vm_end - vma.vm_start, MaliciousFlags.RWX
+        elif flags_str == "r-x" and vma.vm_file.dereference().vol.offset == 0:
+            return vma.vm_start, vma.vm_end - vma.vm_start, MaliciousFlags.RX
+        elif "x" in flags_str:
+            for page_addr, page_size in self._get_dirty_pages(proc_layer, vma):
+                vollog.warning(
+                    f"Found dirty page at {page_addr:#x} inside executable region {vma.vm_start:#x}-{vma.vm_end:#x}!"
+                )
+                # We do not attempt to find other dirty+exec pages once we have found one
+                return page_addr, page_size, MaliciousFlags.X_DIRTY
+
+        return None
+
     def _list_injections(
         self, task
-    ) -> Tuple[interfaces.objects.ObjectInterface, Optional[str], bytes]:
+    ) -> Iterable[
+        Tuple[interfaces.objects.ObjectInterface, Optional[str], bytes, int, int]
+    ]:
         """Generate memory regions for a process that may contain injected
         code."""
-
         proc_layer_name = task.add_process_layer()
         if not proc_layer_name:
-            return None
+            return
 
         proc_layer = self.context.layers[proc_layer_name]
-
-        dump_size = self.config["dump-size"]
-
+        data_size = self.config["hexdump-size"]
         # 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["dump-page"]
-
         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 vma_name == "[vdso]":
+                continue
 
-            # 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
+            suspicious_result = self._is_suspicious(proc_layer, vma)
+            if suspicious_result is None:
+                continue
 
-            if vma.is_suspicious(proc_layer) and vma_name != "[vdso]":
-                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
+            region_start, region_size, suspicious_flag = suspicious_result
+            # If _is_suspicious returns MaliciousFlags.X_DIRTY, this means at least one page
+            # in the region is dirty. If --show-all-dirty-pages is set, then we show
+            # all the dirty pages.
+            if (
+                suspicious_flag == MaliciousFlags.X_DIRTY
+                and self.config["show-all-dirty-pages"]
+            ):
+                # Dump each dirty page
+                for dirty_page_addr, dirty_page_size in self._get_dirty_pages(
+                    proc_layer, vma
+                ):
+                    name = f"{vma_name}, dirty page address: {dirty_page_addr:#x}"
+                    data = proc_layer.read(dirty_page_addr, data_size, pad=True)
+                    yield vma, name, data, dirty_page_addr, dirty_page_size
+                continue
+
+            name = vma_name
+            if suspicious_flag == MaliciousFlags.X_DIRTY:
+                name = f"{vma_name}, dirty page address: {region_start:#x}"
+
+            data = proc_layer.read(vma.vm_start, data_size, pad=True)
+            yield vma, name, data, region_start, region_size
 
     def _generator(self, tasks):
         # determine if we're on a 32 or 64 bit kernel
@@ -109,15 +184,30 @@ class Malfind(interfaces.plugins.PluginInterface):
         for task in tasks:
             process_name = utility.array_to_string(task.comm)
 
-            for vma, vma_name, data, offset in self._list_injections(task):
+            for vma, vma_name, data, region_start, region_size in self._list_injections(
+                task
+            ):
                 if is_32bit_arch:
                     architecture = "intel"
                 else:
                     architecture = "intel64"
 
-                disasm = renderers.Disassembly(
-                    data, vma.vm_start + offset, architecture
-                )
+                disasm = renderers.Disassembly(data, region_start, architecture)
+
+                file_output = "Disabled"
+                if self.config["dump-regions"]:
+                    file_handle = proc.Maps.vma_dump(
+                        self.context,
+                        task,
+                        region_start,
+                        region_start + region_size,
+                        self.open,
+                        self.config["dump-maxsize"],
+                    )
+
+                    if file_handle:
+                        file_handle.close()
+                        file_output = file_handle.preferred_filename
 
                 yield (
                     0,
@@ -130,6 +220,7 @@ class Malfind(interfaces.plugins.PluginInterface):
                         vma.get_protection(),
                         format_hints.HexBytes(data),
                         disasm,
+                        file_output,
                     ),
                 )
 
@@ -146,6 +237,7 @@ class Malfind(interfaces.plugins.PluginInterface):
                 ("Protection", str),
                 ("Hexdump", format_hints.HexBytes),
                 ("Disasm", renderers.Disassembly),
+                ("File output", str),
             ],
             self._generator(
                 pslist.PsList.list_tasks(

volatility3/framework/plugins/renderers/__init__.py

@@ -0,0 +1,8 @@
+# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+"""All core renderer plugins.
+
+These modules should only be imported from volatility3.plugins NOT
+volatility3.framework.plugins
+"""

volatility3/framework/plugins/renderers/parquet_renderer.py

@@ -9,12 +9,13 @@ from typing import (
     Dict,
     List,
     Optional,
-    Tuple,
     TextIO,
+    Tuple,
 )
+
+from volatility3.cli import text_renderer
 from volatility3.framework import interfaces, renderers
 from volatility3.framework.renderers import format_hints
-from volatility3.cli import text_renderer
 
 vollog = logging.getLogger(__name__)
 
@@ -27,192 +28,191 @@ try:
 except ImportError:
     vollog.debug("Arrow/Parquet libraries not found")
 
+if ARROW_PRESENT:
 
-class ArrowRenderer(text_renderer.CLIRenderer):
-    """Renderer that outputs Arrow IPC format data."""
+    class ArrowRenderer(text_renderer.CLIRenderer):
+        """Renderer that outputs Arrow IPC format data."""
 
-    name = "arrow"
-    structured_output = True
-    _version = (1, 0, 0)
+        name = "arrow"
+        structured_output = True
+        _version = (1, 0, 0)
 
-    def __init__(
-        self, options: Optional[List[interfaces.renderers.RenderOption]] = None
-    ) -> None:
-        super().__init__(options)
+        def __init__(
+            self, options: Optional[List[interfaces.renderers.RenderOption]] = None
+        ) -> None:
+            super().__init__(options)
 
-        if not ARROW_PRESENT:
-            raise RuntimeError("Arrow output format requires the pyarrow package")
+            self._to_arrow_type = {
+                renderers.Disassembly: pa.utf8,
+                bool: pa.bool_,
+                int: pa.int64,
+                float: pa.float64,
+                str: pa.utf8,
+                datetime.datetime: lambda: pa.timestamp("ms"),
+                format_hints.Bin: pa.uint64,
+                format_hints.Hex: pa.uint64,
+                format_hints.MultiTypeData: pa.binary,
+                format_hints.HexBytes: pa.binary,
+                renderers.LayerData: pa.binary,
+                bytes: pa.binary,
+            }
 
-        self._to_arrow_type = {
-            renderers.Disassembly: pa.utf8,
-            bool: pa.bool_,
-            int: pa.int64,
-            float: pa.float64,
-            str: pa.utf8,
-            datetime.datetime: lambda: pa.timestamp("ms"),
-            format_hints.Bin: pa.uint64,
-            format_hints.Hex: pa.uint64,
-            format_hints.MultiTypeData: pa.binary,
-            format_hints.HexBytes: pa.binary,
-            renderers.LayerData: pa.binary,
-            bytes: pa.binary,
-        }
+            # indicates if the output from the plugin is nested, e.g., pstree
+            # which would then need to be flattened
+            self._is_tree_result = False
+            self._node_id_counter = 0
 
-        # indicates if the output from the plugin is nested, e.g., pstree
-        # which would then need to be flattened
-        self._is_tree_result = False
-        self._node_id_counter = 0
+        def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
+            return []
 
-    def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
-        return []
+        def to_arrow_schema(self, grid: interfaces.renderers.TreeGrid) -> "pa.Schema":
+            fields = []
+            for column in grid.columns:
+                arrow_type = self._to_arrow_type[column.type]
+                fields.append(pa.field(column.name, arrow_type()))
 
-    def to_arrow_schema(self, grid: interfaces.renderers.TreeGrid) -> "pa.Schema":
-        fields = []
-        for column in grid.columns:
-            arrow_type = self._to_arrow_type[column.type]
-            fields.append(pa.field(column.name, arrow_type()))
+            # if the output is nested, e.g., windows.pstree
+            if self._is_tree_result:
+                fields.append(pa.field("_vol_id", pa.uint64()))
+                fields.append(pa.field("_vol_parent_id", pa.uint64()))
 
-        # if the output is nested, e.g., windows.pstree
-        if self._is_tree_result:
-            fields.append(pa.field("_vol_id", pa.uint64()))
-            fields.append(pa.field("_vol_parent_id", pa.uint64()))
+            return pa.schema(fields)
 
-        return pa.schema(fields)
+        def _flatten_tree_structure(self, nested: List[Dict]) -> List[Dict]:
+            """
+            Flattens a list of nested dicts using the `__children` key.
 
-    def _flatten_tree_structure(self, nested: List[Dict]) -> List[Dict]:
-        """
-        Flattens a list of nested dicts using the `__children` key.
+            Each node gets a `_vol_id` and a `_vol_parent_id` to preserve
+            the original tree structure in a flat format suitable for tabular output.
 
-        Each node gets a `_vol_id` and a `_vol_parent_id` to preserve
-        the original tree structure in a flat format suitable for tabular output.
+            Args:
+                nested: A list of dicts with optional `__children` lists (tree nodes).
 
-        Args:
-            nested: A list of dicts with optional `__children` lists (tree nodes).
+            Returns:
+                A flat list of dicts with `_vol_id` and `_vol_parent_id`.
+            """
+            rows = []
+            self._node_id_counter = 0
 
-        Returns:
-            A flat list of dicts with `_vol_id` and `_vol_parent_id`.
-        """
-        rows = []
-        self._node_id_counter = 0
+            def _process_node(node: Dict, parent_id: Optional[int]):
+                current_id = self._node_id_counter
+                self._node_id_counter += 1
 
-        def _process_node(node: Dict, parent_id: Optional[int]):
-            current_id = self._node_id_counter
-            self._node_id_counter += 1
+                entry = {k: v for k, v in node.items() if k != "__children"}
+                entry["_vol_id"] = current_id
+                entry["_vol_parent_id"] = parent_id
+                rows.append(entry)
 
-            entry = {k: v for k, v in node.items() if k != "__children"}
-            entry["_vol_id"] = current_id
-            entry["_vol_parent_id"] = parent_id
-            rows.append(entry)
+                for child in node.get("__children", []):
+                    _process_node(child, current_id)
 
-            for child in node.get("__children", []):
-                _process_node(child, current_id)
+            for root in nested:
+                _process_node(root, None)
 
-        for root in nested:
-            _process_node(root, None)
+            return rows
 
-        return rows
+        def output_result(self, schema: "pa.Schema", outfd: TextIO, result):
+            """Outputs the JSON data to a file in a particular format"""
 
-    def output_result(self, schema: "pa.Schema", outfd: TextIO, result):
-        """Outputs the JSON data to a file in a particular format"""
+            if self._is_tree_result:
+                result = self._flatten_tree_structure(result)
 
-        if self._is_tree_result:
-            result = self._flatten_tree_structure(result)
+            t = pa.Table.from_pylist(result, schema=schema)
+            self.write_table(t, outfd)
 
-        t = pa.Table.from_pylist(result, schema=schema)
-        self.write_table(t, outfd)
+        def write_table(self, t: "pa.Table", outfd: TextIO) -> None:
+            buf = pa.BufferOutputStream()
 
-    def write_table(self, t: "pa.Table", outfd: TextIO) -> None:
-        buf = pa.BufferOutputStream()
+            writer = pa.ipc.new_stream(buf, t.schema)
+            writer.write_table(t)
+            writer.close()
 
-        writer = pa.ipc.new_stream(buf, t.schema)
-        writer.write_table(t)
-        writer.close()
+            # Get the buffer bytes and write to output
+            buf_bytes = buf.getvalue().to_pybytes()
+            outfd.buffer.write(buf_bytes)
 
-        # Get the buffer bytes and write to output
-        buf_bytes = buf.getvalue().to_pybytes()
-        outfd.buffer.write(buf_bytes)
+        def render(self, grid: interfaces.renderers.TreeGrid):
+            outfd = sys.stdout
+            final_output: Tuple[
+                Dict[str, List[interfaces.renderers.TreeNode]],
+                List[interfaces.renderers.TreeNode],
+            ] = ({}, [])
 
-    def render(self, grid: interfaces.renderers.TreeGrid):
-        outfd = sys.stdout
-        final_output: Tuple[
-            Dict[str, List[interfaces.renderers.TreeNode]],
-            List[interfaces.renderers.TreeNode],
-        ] = ({}, [])
+            ignore_columns = self.ignored_columns(grid)
 
-        ignore_columns = self.ignored_columns(grid)
+            def visitor(
+                node: interfaces.renderers.TreeNode,
+                accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]],
+            ) -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]:
+                # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
+                acc_map, final_tree = accumulator
+                node_dict: Dict[str, Any] = {"__children": []}
+                line = []
+                for column_index, column in enumerate(grid.columns):
+                    if column in ignore_columns:
+                        continue
 
-        def visitor(
-            node: interfaces.renderers.TreeNode,
-            accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]],
-        ) -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]:
-            # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
-            acc_map, final_tree = accumulator
-            node_dict: Dict[str, Any] = {"__children": []}
-            line = []
-            for column_index, column in enumerate(grid.columns):
-                if column in ignore_columns:
-                    continue
+                    data = list(node.values)[column_index]
 
-                data = list(node.values)[column_index]
+                    if isinstance(data, interfaces.renderers.BaseAbsentValue):
+                        data = None
 
-                if isinstance(data, interfaces.renderers.BaseAbsentValue):
-                    data = None
+                    if isinstance(data, renderers.Disassembly):
+                        data = text_renderer.display_disassembly(data)
 
-                if isinstance(data, renderers.Disassembly):
-                    data = text_renderer.display_disassembly(data)
+                    if isinstance(data, renderers.LayerData):
+                        data = text_renderer.LayerDataRenderer().render_bytes(data)[0]
 
-                if isinstance(data, renderers.LayerData):
-                    data = text_renderer.LayerDataRenderer().render_bytes(data)[0]
+                    node_dict[column.name] = data
+                    line.append(data)
 
-                node_dict[column.name] = data
-                line.append(data)
+                if self.filter and self.filter.filter(line):
+                    return accumulator
 
-            if self.filter and self.filter.filter(line):
-                return accumulator
+                if node.parent:
+                    acc_map[node.parent.path]["__children"].append(node_dict)
+                    self._is_tree_result = True
+                else:
+                    final_tree.append(node_dict)
+                acc_map[node.path] = node_dict
 
-            if node.parent:
-                acc_map[node.parent.path]["__children"].append(node_dict)
-                self._is_tree_result = True
+                return (acc_map, final_tree)
+
+            if not grid.populated:
+                grid.populate(visitor, final_output)
             else:
-                final_tree.append(node_dict)
-            acc_map[node.path] = node_dict
+                grid.visit(
+                    node=None, function=visitor, initial_accumulator=final_output
+                )
 
-            return (acc_map, final_tree)
+            schema = self.to_arrow_schema(grid)
+            self.output_result(schema, outfd, final_output[1])
 
-        if not grid.populated:
-            grid.populate(visitor, final_output)
-        else:
-            grid.visit(node=None, function=visitor, initial_accumulator=final_output)
+    class ParquetRenderer(ArrowRenderer):
+        """Renderer that outputs Parquet format data."""
 
-        schema = self.to_arrow_schema(grid)
-        self.output_result(schema, outfd, final_output[1])
+        name = "parquet"
+        structured_output = True
+        _version = (1, 0, 0)
 
+        def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
+            return []
 
-class ParquetRenderer(ArrowRenderer):
-    """Renderer that outputs Parquet format data."""
+        def write_table(self, table: "pa.Table", outfd: TextIO) -> None:
+            """
+            Writes a table to stdout using the Parquet format.
 
-    name = "parquet"
-    structured_output = True
-    _version = (1, 0, 0)
+            Args:
+                t: The Arrow table to write
+                outfd: The output file descriptor
 
-    def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
-        return []
+            Returns:
+                Nothing
+            """
+            # Write DataFrame to a temporary file-like object
+            buf = pa.BufferOutputStream()
+            pq.write_table(table, buf, compression="snappy")
 
-    def write_table(self, table: "pa.Table", outfd: TextIO) -> None:
-        """
-        Writes a table to stdout using the Parquet format.
-
-        Args:
-            t: The Arrow table to write
-            outfd: The output file descriptor
-
-        Returns:
-            Nothing
-        """
-        # Write DataFrame to a temporary file-like object
-        buf = pa.BufferOutputStream()
-        pq.write_table(table, buf, compression="snappy")
-
-        # Get the buffer as a bytes object
-        buf_bytes = buf.getvalue().to_pybytes()
-        outfd.buffer.write(buf_bytes)
+            # Get the buffer as a bytes object
+            buf_bytes = buf.getvalue().to_pybytes()
+            outfd.buffer.write(buf_bytes)

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

@@ -1360,7 +1360,7 @@ class vm_area_struct(objects.StructType):
                     break
         return malicious_pages
 
-    # used by malfind
+    # previously used by malfind
     def is_suspicious(self, proclayer=None):
         ret = False