WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-13 13:17:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #482 from volatilityfoundation/feature/better-strings-checking

Browse Source 
Feature/better strings checking
This commit is contained in:
ikelos
2021-03-21 00:13:37 +00:00
committed by GitHub
parent de78763e32 05ffab9108
commit 6702595ebe
1 changed files with 69 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/windows/strings.py
+69 -40
View File
@@ -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,8 +18,9 @@ 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?")
strings_pattern = re.compile(rb"^(?:\W*)([0-9]+)(?:\W*)(\w[\w\W]+)\n?")
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -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,27 +44,38 @@ class Strings(interfaces.plugins.PluginInterface):
def _generator(self) -> Generator[Tuple, None, None]:
"""Generates results from a strings file."""
revmap = self.generate_mapping(self.config['primary'])
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")
strings_size = path.getsize(strings_fp.file.name)
line = strings_fp.readline()
last_prog = 0
count = 0 # type: float
while line:
count += 1
try:
offset, string = self._parse_line(line)
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)))
string_list.append((offset, string))
except ValueError:
vollog.error("Strings file is in the wrong format")
return
vollog.error("Line in unrecognized format: line {}".format(count))
line = strings_fp.readline()
prog = strings_fp.tell() / strings_size * 100
revmap = self.generate_mapping(self.context,
self.config['primary'],
self.config['nt_symbols'],
progress_callback = self._progress_callback,
pid_list = self.config['pid'])
last_prog = line_count = 0 # type: float
num_strings = len(string_list)
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 = line_count / num_strings * 100
if round(prog, 1) > last_prog:
last_prog = round(prog, 1)
self._progress_callback(prog, "Matching strings in memory")
@@ -81,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
@@ -101,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