diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 43d6308c7..779b698ad 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -139,38 +139,65 @@ class PoolScanner(plugins.PluginInterface): and not PoolScanner.is_windows_8_or_later(context, layer_name, symbol_table)) def _generator(self): + for result in self.generate_pool_scan(self.context, + self.config["primary"], + self.config["nt_symbols"]): + constraint, mem_object, header = result + + # generate some type-specific info for sanity checking + if constraint.object_type == "Process": + name = mem_object.ImageFileName.cast( + "string", max_length=mem_object.ImageFileName.vol.count, errors="replace") + elif constraint.object_type == "File": + try: + name = mem_object.FileName.String + except exceptions.PagedInvalidAddressException: + vollog.log(constants.LOGLEVEL_VVV, "Skipping file at {0:#x}".format(mem_object.vol.offset)) + continue + else: + name = renderers.NotApplicableValue() + + yield (0, (constraint.type_name, format_hints.Hex(header.vol.offset), header.vol.layer_name, name)) + + @classmethod + def generate_pool_scan(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table: str) \ + -> Generator[Tuple[PoolConstraint, interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface], None, None]: + constraints = [ # atom tables PoolConstraint( b'AtmT', - type_name = self.config["nt_symbols"] + constants.BANG + "_RTL_ATOM_TABLE", + type_name = symbol_table + constants.BANG + "_RTL_ATOM_TABLE", size = (200, None), page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # processes on windows before windows 8 PoolConstraint( b'Pro\xe3', - type_name = self.config["nt_symbols"] + constants.BANG + "_EPROCESS", + type_name = symbol_table + constants.BANG + "_EPROCESS", object_type = "Process", size = (600, None), page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # processes on windows starting with windows 8 PoolConstraint( b'Proc', - type_name = self.config["nt_symbols"] + constants.BANG + "_EPROCESS", + type_name = symbol_table + constants.BANG + "_EPROCESS", object_type = "Process", size = (600, None), page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # files on windows before windows 8 PoolConstraint( b'Fil\xe5', - type_name = self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT", + type_name = symbol_table + constants.BANG + "_FILE_OBJECT", object_type = "File", size = (150, None), page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # files on windows starting with windows 8 PoolConstraint( b'File', - type_name = self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT", + type_name = symbol_table + constants.BANG + "_FILE_OBJECT", object_type = "File", size = (150, None), page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), @@ -178,24 +205,24 @@ class PoolScanner(plugins.PluginInterface): # get the object type map type_map = handles.Handles.list_objects( - context = self.context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"]) + context = context, layer_name = layer_name, symbol_table = symbol_table) cookie = handles.Handles.find_cookie( - context = self.context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"]) + context = context, layer_name = layer_name, symbol_table = symbol_table) - is_windows_10 = self.is_windows_10(context = self._context, symbol_table = self.config["nt_symbols"]) - is_windows_8_or_later = self.is_windows_8_or_later( - context = self._context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"]) + is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table) + is_windows_8_or_later = cls.is_windows_8_or_later( + context = context, layer_name = layer_name, symbol_table = symbol_table) # start off with the primary virtual layer - scan_layer = self.config['primary'] + scan_layer = layer_name # switch to a non-virtual layer if necessary if not is_windows_10: - scan_layer = self.context.memory[scan_layer].config['memory_layer'] + scan_layer = context.memory[scan_layer].config['memory_layer'] - for constraint, header in self.pool_scan( - self._context, scan_layer, self.config['nt_symbols'], constraints, alignment = 8): + for constraint, header in cls.pool_scan( + context, scan_layer, symbol_table, constraints, alignment = 8): mem_object = header.get_object( type_name = constraint.type_name, @@ -209,20 +236,7 @@ class PoolScanner(plugins.PluginInterface): vollog.log(constants.LOGLEVEL_VVV, "Cannot create an instance of {}".format(constraint.type_name)) continue - # generate some type-specific info for sanity checking - if constraint.object_type == "Process": - name = mem_object.ImageFileName.cast( - "string", max_length = mem_object.ImageFileName.vol.count, errors = "replace") - elif constraint.object_type == "File": - try: - name = mem_object.FileName.String - except exceptions.PagedInvalidAddressException: - vollog.log(constants.LOGLEVEL_VVV, "Skipping file at {0:#x}".format(mem_object.vol.offset)) - continue - else: - name = renderers.NotApplicableValue() - - yield (0, (constraint.type_name, format_hints.Hex(header.vol.offset), header.vol.layer_name, name)) + yield constraint, mem_object, header @classmethod def pool_scan(cls, diff --git a/volatility/framework/plugins/windows/psscan.py b/volatility/framework/plugins/windows/psscan.py new file mode 100644 index 000000000..f0b5ab182 --- /dev/null +++ b/volatility/framework/plugins/windows/psscan.py @@ -0,0 +1,81 @@ +# This file was contributed to the Volatility Framework Version 3. +# Copyright (C) 2018 Volatility Foundation. +# +# THE LICENSED WORK IS PROVIDED UNDER THE TERMS OF THE Volatility Contributors +# Public License V1.0("LICENSE") AS FIRST COMPLETED BY: Volatility Foundation, +# Inc. ANY USE, PUBLIC DISPLAY, PUBLIC PERFORMANCE, REPRODUCTION OR DISTRIBUTION +# OF, OR PREPARATION OF SUBSEQUENT WORKS, DERIVATIVE WORKS OR DERIVED WORKS BASED +# ON, THE LICENSED WORK CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS LICENSE AND ITS +# TERMS, WHETHER OR NOT SUCH RECIPIENT READS THE TERMS OF THE LICENSE. "LICENSED +# WORK,” “RECIPIENT" AND “DISTRIBUTOR" ARE DEFINED IN THE LICENSE. A COPY OF THE +# LICENSE IS LOCATED IN THE TEXT FILE ENTITLED "LICENSE.txt" ACCOMPANYING THE +# CONTENTS OF THIS FILE. IF A COPY OF THE LICENSE DOES NOT ACCOMPANY THIS FILE, A +# COPY OF THE LICENSE MAY ALSO BE OBTAINED AT THE FOLLOWING WEB SITE: +# https://www.volatilityfoundation.org/license/vcpl_v1.0 +# +# Software distributed under the License is distributed on an "AS IS" basis, +# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the +# specific language governing rights and limitations under the License. +# + +import datetime +from typing import Iterable + +import volatility.framework.interfaces.plugins as plugins +from volatility.framework import renderers, interfaces +from volatility.framework.configuration import requirements +from volatility.framework.renderers import format_hints +from volatility.plugins import timeliner +import volatility.plugins.windows.poolscanner as poolscanner + +class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface): + """Scans for processes present in a particular windows memory image""" + + @classmethod + def get_requirements(cls): + return [ + requirements.TranslationLayerRequirement( + name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]), + requirements.SymbolRequirement(name = "nt_symbols", description = "Windows kernel symbols"), + ] + + @classmethod + def scan_processes(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table: str) -> \ + Iterable[interfaces.objects.ObjectInterface]: + """Scans for processes using the poolscanner module and constraints""" + + for result in poolscanner.PoolScanner.generate_pool_scan(context, + layer_name, + symbol_table): + + constraint, mem_object, _header = result + if constraint.object_type == "Process": + yield mem_object + + def _generator(self): + for proc in self.scan_processes( + self.context, + self.config['primary'], + self.config['nt_symbols']): + + yield (0, (proc.UniqueProcessId, proc.InheritedFromUniqueProcessId, + proc.ImageFileName.cast("string", max_length = proc.ImageFileName.vol.count, errors = 'replace'), + format_hints.Hex(proc.vol.offset), proc.ActiveThreads, proc.get_handle_count(), proc.get_session_id(), + proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time())) + + def generate_timeline(self): + for row in self._generator(): + _depth, row_data = row + description = "Process: {} ({})".format(row_data[2], row_data[3]) + yield (description, timeliner.TimeLinerType.CREATED, row_data[8]) + yield (description, timeliner.TimeLinerType.MODIFIED, row_data[9]) + + def run(self): + return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str), + ("Offset", format_hints.Hex), ("Threads", int), + ("Handles", int), ("SessionId", int), ("Wow64", bool), + ("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime)], + self._generator())