diff --git a/volatility3/framework/plugins/windows/strings.py b/volatility3/framework/plugins/windows/strings.py index 5d0d3f873..a44e59672 100644 --- a/volatility3/framework/plugins/windows/strings.py +++ b/volatility3/framework/plugins/windows/strings.py @@ -4,10 +4,9 @@ import logging import re -from os import path -from typing import Dict, Generator, List, Set, Tuple +from typing import Dict, Generator, List, Set, Tuple, Optional -from volatility3.framework import interfaces, renderers, exceptions +from volatility3.framework import interfaces, renderers, exceptions, constants from volatility3.framework.configuration import requirements from volatility3.framework.layers import intel, resources, linear from volatility3.framework.renderers import format_hints @@ -19,6 +18,7 @@ vollog = logging.getLogger(__name__) class Strings(interfaces.plugins.PluginInterface): """Reads output from the strings command and indicates which process(es) each string belongs to.""" + _version = (1, 0, 0) _required_framework_version = (1, 0, 0) strings_pattern = re.compile(rb"^(?:\W*)([0-9]+)(?:\W*)(\w[\w\W]+)\n?") @@ -30,6 +30,10 @@ class Strings(interfaces.plugins.PluginInterface): description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]), requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"), + requirements.ListRequirement(name = 'pid', + element_type = int, + description = "Process ID to include (all other processes are excluded)", + optional = True), requirements.URIRequirement(name = "strings_file", description = "Strings file") ] # TODO: Make URLRequirement that can accept a file address which the framework can open @@ -40,40 +44,38 @@ class Strings(interfaces.plugins.PluginInterface): def _generator(self) -> Generator[Tuple, None, None]: """Generates results from a strings file.""" - string_list = {} # type: Dict[int, List[bytes]] + string_list = [] # type: List[Tuple[int,bytes]] # Test strings file format is accurate accessor = resources.ResourceAccessor() strings_fp = accessor.open(self.config['strings_file'], "rb") line = strings_fp.readline() - count = 0 + count = 0 # type: float while line: count += 1 try: offset, string = self._parse_line(line) - string_list[offset] = string_list.get(offset, []) + [string] + string_list.append((offset, string)) except ValueError: vollog.error("Line in unrecognized format: line {}".format(count)) - return line = strings_fp.readline() - # TODO: Check the strings file *before* doing the expensive computation - revmap = self.generate_mapping(self.config['primary']) + revmap = self.generate_mapping(self.context, + self.config['primary'], + self.config['nt_symbols'], + progress_callback = self._progress_callback, + pid_list = self.config['pid']) - strings_fp = accessor.open(self.config['strings_file'], "rb") - strings_size = path.getsize(strings_fp.file.name) - - last_prog = count = 0 # type: float + last_prog = line_count = 0 # type: float num_strings = len(string_list) - for offset in string_list: - string = b"; ".join(string_list[offset]) - count += 1 + for 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 = count / num_strings * 100 + 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") @@ -94,17 +96,29 @@ class Strings(interfaces.plugins.PluginInterface): offset, string = match.group(1, 2) return int(offset), string - def generate_mapping(self, layer_name: str) -> Dict[int, Set[Tuple[str, int]]]: + @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 """ - layer = self._context.layers[layer_name] + filter = pslist.PsList.create_pid_filter(pid_list) + + layer = context.layers[layer_name] reverse_map = dict() # type: Dict[int, Set[Tuple[str, int]]] if isinstance(layer, intel.Intel): # We don't care about errors, we just wanted chunks that map correctly @@ -114,31 +128,33 @@ class Strings(interfaces.plugins.PluginInterface): cur_set = reverse_map.get(mapped_offset >> 12, set()) cur_set.add(("kernel", offset)) reverse_map[mapped_offset >> 12] = cur_set - self._progress_callback((offset * 100) / layer.maximum_address, "Creating reverse kernel map") + if progress_callback: + progress_callback((offset * 100) / layer.maximum_address, "Creating reverse kernel map") # TODO: Include kernel modules - for process in pslist.PsList.list_processes(self.context, self.config['primary'], - self.config['nt_symbols']): - 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)) + continue - proc_layer = self.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(("Process {}".format(process.UniqueProcessId), offset)) - reverse_map[mapped_offset >> 12] = cur_set - # FIXME: make the progress for all processes, rather than per-process - self._progress_callback((offset * 100) / layer.maximum_address, - "Creating mapping for task {}".format(process.UniqueProcessId)) + 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(("Process {}".format(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, + "Creating mapping for task {}".format(process.UniqueProcessId)) return reverse_map