diff --git a/volatility3/framework/plugins/windows/modscan.py b/volatility3/framework/plugins/windows/modscan.py index 99fadac07..3a786d2ca 100644 --- a/volatility3/framework/plugins/windows/modscan.py +++ b/volatility3/framework/plugins/windows/modscan.py @@ -9,16 +9,20 @@ from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import intermed from volatility3.framework.symbols.windows.extensions import pe -from volatility3.plugins.windows import poolscanner, dlllist, pslist +from volatility3.plugins.windows import poolscanner, dlllist, pslist, modules vollog = logging.getLogger(__name__) -class ModScan(interfaces.plugins.PluginInterface): +class ModScan(modules.Modules): """Scans for modules present in a particular windows memory image.""" _required_framework_version = (2, 0, 0) - _version = (1, 0, 0) + _version = (2, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._enumeration_method = self.scan_modules @classmethod def get_requirements(cls): @@ -31,6 +35,9 @@ class ModScan(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0) ), + requirements.VersionRequirement( + name="modules", component=modules.Modules, version=(1, 1, 0) + ), requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(2, 0, 0) ), @@ -43,6 +50,17 @@ class ModScan(interfaces.plugins.PluginInterface): default=False, optional=True, ), + requirements.IntRequirement( + name="base", + description="Extract a single module with BASE address", + optional=True, + ), + requirements.StringRequirement( + name="name", + description="module name/sub string", + optional=True, + default=None, + ), ] @classmethod @@ -72,157 +90,3 @@ 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: 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): - kernel = self.context.modules[self.config["kernel"]] - - session_layers = list( - self.get_session_layers( - self.context, kernel.layer_name, kernel.symbol_table_name - ) - ) - pe_table_name = intermed.IntermediateSymbolTable.create( - self.context, self.config_path, "windows", "pe", class_types=pe.class_types - ) - - for mod in self.scan_modules( - self.context, kernel.layer_name, kernel.symbol_table_name - ): - try: - BaseDllName = mod.BaseDllName.get_string() - except exceptions.InvalidAddressException: - BaseDllName = "" - - try: - FullDllName = mod.FullDllName.get_string() - except exceptions.InvalidAddressException: - FullDllName = "" - - file_output = "Disabled" - if self.config["dump"]: - session_layer_name = self.find_session_layer( - self.context, session_layers, mod.DllBase - ) - file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}" - if session_layer_name: - file_handle = dlllist.DllList.dump_pe( - self.context, - pe_table_name, - mod, - self.open, - layer_name=session_layer_name, - ) - file_output = "Error outputting file" - if file_handle: - file_output = file_handle.preferred_filename - - yield ( - 0, - ( - format_hints.Hex(mod.vol.offset), - format_hints.Hex(mod.DllBase), - format_hints.Hex(mod.SizeOfImage), - BaseDllName, - FullDllName, - file_output, - ), - ) - - def run(self): - return renderers.TreeGrid( - [ - ("Offset", format_hints.Hex), - ("Base", format_hints.Hex), - ("Size", format_hints.Hex), - ("Name", str), - ("Path", str), - ("File output", str), - ], - self._generator(), - ) diff --git a/volatility3/framework/plugins/windows/modules.py b/volatility3/framework/plugins/windows/modules.py index 7dabf6954..8aea3dd45 100644 --- a/volatility3/framework/plugins/windows/modules.py +++ b/volatility3/framework/plugins/windows/modules.py @@ -22,6 +22,10 @@ class Modules(interfaces.plugins.PluginInterface): _required_framework_version = (2, 0, 0) _version = (1, 1, 0) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._enumeration_method = self.list_modules + @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: return [ @@ -42,6 +46,11 @@ class Modules(interfaces.plugins.PluginInterface): default=False, optional=True, ), + requirements.IntRequirement( + name="base", + description="Extract a single module with BASE address", + optional=True, + ), requirements.StringRequirement( name="name", description="module name/sub string", @@ -50,49 +59,74 @@ class Modules(interfaces.plugins.PluginInterface): ), ] + def dump_module(self, session_layers, pe_table_name, mod): + session_layer_name = self.find_session_layer( + self.context, session_layers, mod.DllBase + ) + file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}" + if session_layer_name: + file_handle = dlllist.DllList.dump_pe( + self.context, + pe_table_name, + mod, + self.open, + layer_name=session_layer_name, + ) + file_output = "Error outputting file" + if file_handle: + file_output = file_handle.preferred_filename + + return file_output + + def process_module(self, session_layers, pe_table_name, mod): + if self.config["base"] and self.config["base"] != mod.DllBase: + return None + + try: + BaseDllName = mod.BaseDllName.get_string() + except exceptions.InvalidAddressException: + BaseDllName = "" + + if self.config["name"] and self.config["name"] not in BaseDllName: + return None + + try: + FullDllName = mod.FullDllName.get_string() + except exceptions.InvalidAddressException: + FullDllName = "" + + file_output = "Disabled" + if self.config["dump"]: + file_output = self.dump_module(session_layers, pe_table_name, mod) + + return ( + format_hints.Hex(mod.vol.offset), + format_hints.Hex(mod.DllBase), + format_hints.Hex(mod.SizeOfImage), + BaseDllName, + FullDllName, + file_output, + ) + def _generator(self): kernel = self.context.modules[self.config["kernel"]] + pe_table_name = intermed.IntermediateSymbolTable.create( self.context, self.config_path, "windows", "pe", class_types=pe.class_types ) - for mod in self.list_modules( + session_layers = list( + self.get_session_layers( + self.context, kernel.layer_name, kernel.symbol_table_name + ) + ) + + for mod in self._enumeration_method( self.context, kernel.layer_name, kernel.symbol_table_name ): - try: - BaseDllName = mod.BaseDllName.get_string() - except exceptions.InvalidAddressException: - BaseDllName = "" - - try: - FullDllName = mod.FullDllName.get_string() - except exceptions.InvalidAddressException: - FullDllName = "" - - if self.config["name"] and self.config["name"] not in BaseDllName: - continue - - file_output = "Disabled" - if self.config["dump"]: - file_handle = dlllist.DllList.dump_pe( - self.context, pe_table_name, mod, self.open - ) - file_output = "Error outputting file" - if file_handle: - file_handle.close() - file_output = file_handle.preferred_filename - - yield ( - 0, - ( - format_hints.Hex(mod.vol.offset), - format_hints.Hex(mod.DllBase), - format_hints.Hex(mod.SizeOfImage), - BaseDllName, - FullDllName, - file_output, - ), - ) + record = self.process_module(session_layers, pe_table_name, mod) + if record: + yield (0, record) @classmethod def get_session_layers(