diff --git a/volatility/framework/plugins/windows/modscan.py b/volatility/framework/plugins/windows/modscan.py index 6c45ed101..45ca70ba7 100644 --- a/volatility/framework/plugins/windows/modscan.py +++ b/volatility/framework/plugins/windows/modscan.py @@ -1,15 +1,17 @@ # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # +import logging +from typing import Iterable, List, Generator -from typing import Iterable - -from volatility.framework import renderers, interfaces, exceptions +from volatility.framework import renderers, interfaces, exceptions, constants from volatility.framework.configuration import requirements from volatility.framework.renderers import format_hints from volatility.framework.symbols import intermed from volatility.framework.symbols.windows.extensions import pe -from volatility.plugins.windows import poolscanner, dlllist +from volatility.plugins.windows import poolscanner, dlllist, pslist + +vollog = logging.getLogger(__name__) class ModScan(interfaces.plugins.PluginInterface): @@ -27,6 +29,7 @@ class ModScan(interfaces.plugins.PluginInterface): requirements.VersionRequirement(name = 'poolerscanner', component = poolscanner.PoolScanner, version = (1, 0, 0)), + requirements.VersionRequirement(name = 'pslist', component = dlllist.DllList, version = (2, 0, 0)), requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (2, 0, 0)), requirements.BooleanRequirement(name = 'dump', description = "Extract listed modules", @@ -58,7 +61,82 @@ class ModScan(interfaces.plugins.PluginInterface): _constraint, mem_object, _header = result yield mem_object + @classmethod + def get_session_layers(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table: str, + pids: List[int] = None) -> Generator[str, None, None]: + """Build a cache of possible virtual layers, in priority starting with + the primary/kernel layer. Then keep one layer per session by cycling + through the process list. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + layer_name: The name of the layer on which to operate + symbol_table: The name of the table containing the kernel symbols + pids: A list of process identifiers to include exclusively or None for no filter + + Returns: + A list of session layer names + """ + seen_ids = [] # type: List[interfaces.objects.ObjectInterface] + filter_func = pslist.PsList.create_pid_filter(pids or []) + + for proc in pslist.PsList.list_processes(context = context, + layer_name = layer_name, + symbol_table = symbol_table, + filter_func = filter_func): + proc_id = "Unknown" + try: + proc_id = proc.UniqueProcessId + proc_layer_name = proc.add_process_layer() + + # create the session space object in the process' own layer. + # not all processes have a valid session pointer. + session_space = context.object(symbol_table + constants.BANG + "_MM_SESSION_SPACE", + layer_name = layer_name, + offset = proc.Session) + + if session_space.SessionId in seen_ids: + continue + + except exceptions.InvalidAddressException: + vollog.log( + constants.LOGLEVEL_VVV, + "Process {} does not have a valid Session or a layer could not be constructed for it".format( + proc_id)) + continue + + # save the layer if we haven't seen the session yet + seen_ids.append(session_space.SessionId) + yield proc_layer_name + + @classmethod + def find_session_layer(cls, context: interfaces.context.ContextInterface, session_layers: Iterable[str], + base_address: int): + """Given a base address and a list of layer names, find a layer that + can access the specified address. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + layer_name: The name of the layer on which to operate + symbol_table: The name of the table containing the kernel symbols + session_layers: A list of session layer names + base_address: The base address to identify the layers that can access it + + Returns: + Layer name or None if no layers that contain the base address can be found + """ + + for layer_name in session_layers: + if context.layers[layer_name].is_valid(base_address): + return layer_name + + return None + def _generator(self): + session_layers = list(self.get_session_layers(self.context, self.config['primary'], self.config['nt_symbols'])) pe_table_name = intermed.IntermediateSymbolTable.create(self.context, self.config_path, "windows", @@ -79,10 +157,15 @@ class ModScan(interfaces.plugins.PluginInterface): file_output = "Disabled" if self.config['dump']: - file_handler = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self.open) - file_output = "Error outputting file" - if file_handler: - file_output = file_handler.preferred_filename + + session_layer_name = self.find_session_layer(self.context, session_layers, mod.DllBase) + file_output = "Cannot find a viable session layer for {0:#x}".format(mod.DllBase) + if session_layer_name: + file_handler = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self.open, + layer_name = session_layer_name) + file_output = "Error outputting file" + if file_handler: + file_output = file_handler.preferred_filename yield (0, ( format_hints.Hex(mod.vol.offset),