Plugins: Windows.string speed enhancements by eve

Author: ikelos

volatility3/framework/plugins/windows/strings.py

@@ -2,6 +2,7 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
+from dataclasses import dataclass
 import logging
 import re
 from typing import Dict, Generator, List, Set, Tuple, Optional
@@ -15,10 +16,62 @@
 vollog = logging.getLogger(__name__)
 
 
+@dataclass
+class MappingNode:
+    def __init__(
+        self,
+        physical_addr_start,
+        physical_addr_end,
+        virtual_addr_start,
+        virtual_addr_end,
+        process_id,
+        region,
+    ) -> None:
+        self.physical_addr_start = physical_addr_start
+        self.physical_addr_end = physical_addr_end
+        self.virtual_addr_start = virtual_addr_start
+        self.virtual_addr_end = virtual_addr_end
+        self.process_id = process_id
+        self.region = region
+
+
+class MappingTree:
+    def __init__(self, root=None) -> None:
+        self.root = root
+        self.left = None
+        self.right = None
+
+    def add(self, node):
+        if isinstance(node, MappingNode):
+            if self.root == None:
+                self.root = node
+            elif node.physical_addr_start < self.root.physical_addr_start:
+                if self.left == None:
+                    self.left = MappingTree(node)
+                else:
+                    self.left.add(node)
+            else:
+                if self.right == None:
+                    self.right = MappingTree(node)
+                else:
+                    self.right.add(node)
+        else:
+            raise TypeError()
+
+    def at(self, point):
+        if self.root:
+            if self.root.physical_addr_start <= point <= self.root.physical_addr_end:
+                yield self.root
+            if point < self.root.physical_addr_start and self.left:
+                yield from self.left.at(point)
+            elif self.right:
+                yield from self.right.at(point)
+
+
 class Strings(interfaces.plugins.PluginInterface):
     """Reads output from the strings command and indicates which process(es) each string belongs to."""
 
-    _version = (1, 2, 0)
+    _version = (2, 0, 0)
     _required_framework_version = (2, 0, 0)
     strings_pattern = re.compile(rb"^(?:\W*)([0-9]+)(?:\W*)(\w[\w\W]+)\n?")
 
@@ -43,11 +96,16 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 name="strings_file", description="Strings file"
             ),
         ]
-        # TODO: Make URLRequirement that can accept a file address which the framework can open
 
     def run(self):
         return renderers.TreeGrid(
-            [("String", str), ("Physical Address", format_hints.Hex), ("Result", str)],
+            [
+                ("String", str),
+                ("Region", str),
+                ("PID", int),
+                ("Physical Address", format_hints.Hex),
+                ("Virtual Address", format_hints.Hex),
+            ],
             self._generator(),
         )
 
@@ -70,7 +128,7 @@ def _generator(self) -> Generator[Tuple, None, None]:
             line = strings_fp.readline()
         kernel = self.context.modules[self.config["kernel"]]
 
-        revmap = self.generate_mapping(
+        revmap_tree = self.generate_mapping(
             self.context,
             kernel.layer_name,
             kernel.symbol_table_name,
@@ -81,26 +139,39 @@ def _generator(self) -> Generator[Tuple, None, None]:
         last_prog: float = 0
         line_count: float = 0
         num_strings = len(string_list)
-        for offset, string in string_list:
+
+        for phys_offset, string in string_list:
             line_count += 1
-            try:
-                revmap_list = [
-                    name + ":" + hex(offset) for (name, offset) in revmap[offset >> 12]
-                ]
-            except (IndexError, KeyError):
-                revmap_list = ["FREE MEMORY"]
-            yield (
-                0,
-                (
-                    str(string, "latin-1"),
-                    format_hints.Hex(offset),
-                    ", ".join(revmap_list),
-                ),
-            )
-            prog = line_count / num_strings * 100
-            if round(prog, 1) > last_prog:
-                last_prog = round(prog, 1)
-                self._progress_callback(prog, "Matching strings in memory")
+
+            matched_region = False
+            for node in revmap_tree.at(phys_offset):
+                matched_region = True
+
+                region_offset = phys_offset - node.physical_addr_start
+                offset = node.virtual_addr_start + region_offset
+                yield (
+                    0,
+                    (
+                        str(string.strip(), "latin-1"),
+                        node.region,
+                        node.process_id,
+                        format_hints.Hex(phys_offset),
+                        format_hints.Hex(offset),
+                    ),
+                )
+
+            if not matched_region:
+                # no maps found for this offset
+                yield (
+                    0,
+                    (
+                        str(string.strip(), "latin-1"),
+                        "Unallocated",
+                        -1,
+                        format_hints.Hex(phys_offset),
+                        format_hints.Hex(0x00),
+                    ),
+                )
 
     def _parse_line(self, line: bytes) -> Tuple[int, bytes]:
         """Parses a single line from a strings file.
@@ -118,83 +189,91 @@ def _parse_line(self, line: bytes) -> Tuple[int, bytes]:
         offset, string = match.group(1, 2)
         return int(offset), string
 
-    @classmethod
     def generate_mapping(
         cls,
         context: interfaces.context.ContextInterface,
         layer_name: str,
         symbol_table: str,
         progress_callback: constants.ProgressCallback = None,
         pid_list: Optional[List[int]] = None,
-    ) -> Dict[int, Set[Tuple[str, int]]]:
-        """Creates a reverse mapping between virtual addresses and physical
-        addresses.
-
-        Args:
-            context: the context for the method to run against
-            layer_name: the layer to map against the string lines
-            symbol_table: the name of the symbol table for the provided layer
-            progress_callback: an optional callable to display progress
-            pid_list: a lit of process IDs to consider when generating the reverse map
-
-        Returns:
-            A mapping of virtual offsets to strings and physical offsets
-        """
+    ):
         filter = pslist.PsList.create_pid_filter(pid_list)
+        revmap_tree = MappingTree()
 
+        # start with kernel mappings
         layer = context.layers[layer_name]
-        reverse_map: Dict[int, Set[Tuple[str, int]]] = dict()
+        min_kernel_addr = 2 ** (layer._maxvirtaddr - 1)
         if isinstance(layer, intel.Intel):
             # We don't care about errors, we just wanted chunks that map correctly
-            for mapval in layer.mapping(0x0, layer.maximum_address, ignore_errors=True):
-                offset, _, mapped_offset, mapped_size, maplayer = mapval
-                for val in range(mapped_offset, mapped_offset + mapped_size, 0x1000):
-                    cur_set = reverse_map.get(val >> 12, set())
-                    cur_set.add(("kernel", offset))
-                    reverse_map[val >> 12] = cur_set
+            for mapval in layer.mapping(
+                min_kernel_addr, layer.maximum_address, ignore_errors=True
+            ):
+                (
+                    virt_offset,
+                    virt_size,
+                    phy_offset,
+                    phy_mapping_size,
+                    _phy_layer_name,
+                ) = mapval
+
+                node = MappingNode(
+                    phy_offset,
+                    phy_offset + phy_mapping_size,
+                    virt_offset,
+                    virt_offset + virt_size,
+                    -1,
+                    "Kernel",
+                )
+                revmap_tree.add(node)
+
                 if progress_callback:
                     progress_callback(
-                        (offset * 100) / layer.maximum_address,
-                        "Creating reverse kernel map",
+                        (virt_offset * 100) / layer.maximum_address,
+                        f"Creating custom tree mapping for kernel",
                     )
 
-            # TODO: Include kernel modules
+        # now process normal processes, ignoring kernel addrs
+        for process in pslist.PsList.list_processes(context, layer_name, symbol_table):
+            if not filter(process):
+                proc_id = "Unknown"
+                try:
+                    proc_id = process.UniqueProcessId
+                    proc_layer_name = process.add_process_layer()
+                except exceptions.InvalidAddressException as excp:
+                    vollog.debug(
+                        "Process {}: invalid address {} in layer {}".format(
+                            proc_id, excp.invalid_address, excp.layer_name
+                        )
+                    )
+                    continue
 
-            for process in pslist.PsList.list_processes(
-                context, layer_name, symbol_table
-            ):
-                if not filter(process):
-                    proc_id = "Unknown"
-                    try:
-                        proc_id = process.UniqueProcessId
-                        proc_layer_name = process.add_process_layer()
-                    except exceptions.InvalidAddressException as excp:
-                        vollog.debug(
-                            "Process {}: invalid address {} in layer {}".format(
-                                proc_id, excp.invalid_address, excp.layer_name
-                            )
+                proc_layer = context.layers[proc_layer_name]
+                max_proc_addr = (2 ** (proc_layer._maxvirtaddr - 1)) - 1
+                if isinstance(proc_layer, linear.LinearlyMappedLayer):
+                    for mapval in proc_layer.mapping(
+                        0, max_proc_addr, ignore_errors=True
+                    ):
+                        (
+                            virt_offset,
+                            virt_size,
+                            phy_offset,
+                            phy_mapping_size,
+                            _phy_layer_name,
+                        ) = mapval
+
+                        node = MappingNode(
+                            phy_offset,
+                            phy_offset + phy_mapping_size,
+                            virt_offset,
+                            virt_offset + virt_size,
+                            proc_id,
+                            "Process",
                         )
-                        continue
-
-                    proc_layer = context.layers[proc_layer_name]
-                    if isinstance(proc_layer, linear.LinearlyMappedLayer):
-                        for mapval in proc_layer.mapping(
-                            0x0, proc_layer.maximum_address, ignore_errors=True
-                        ):
-                            mapped_offset, _, offset, mapped_size, maplayer = mapval
-                            for val in range(
-                                mapped_offset, mapped_offset + mapped_size, 0x1000
-                            ):
-                                cur_set = reverse_map.get(mapped_offset >> 12, set())
-                                cur_set.add(
-                                    (f"Process {process.UniqueProcessId}", offset)
-                                )
-                                reverse_map[mapped_offset >> 12] = cur_set
-                            # FIXME: make the progress for all processes, rather than per-process
-                            if progress_callback:
-                                progress_callback(
-                                    (offset * 100) / layer.maximum_address,
-                                    f"Creating mapping for task {process.UniqueProcessId}",
-                                )
-
-        return reverse_map
+                        revmap_tree.add(node)
+
+                        if progress_callback:
+                            progress_callback(
+                                (virt_offset * 100) / max_proc_addr,
+                                f"Creating custom tree mapping for task {proc_id}",
+                            )
+        return revmap_tree