From f69b00b48c07328b618efaad6aa1e14cb3a5c3eb Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sat, 8 Mar 2025 17:21:27 +0000 Subject: [PATCH] Convert poolscanning APIs to new coding standards --- .../framework/plugins/windows/bigpools.py | 34 ++--- .../framework/plugins/windows/callbacks.py | 126 +++++++----------- .../framework/plugins/windows/cmdscan.py | 17 +-- .../framework/plugins/windows/consoles.py | 50 +++---- .../framework/plugins/windows/dlllist.py | 13 +- .../framework/plugins/windows/driverscan.py | 12 +- .../framework/plugins/windows/filescan.py | 22 ++- .../framework/plugins/windows/handles.py | 5 +- .../plugins/windows/indirect_system_calls.py | 2 +- volatility3/framework/plugins/windows/info.py | 70 +++++----- .../framework/plugins/windows/modscan.py | 4 +- .../framework/plugins/windows/mutantscan.py | 21 ++- .../framework/plugins/windows/netscan.py | 52 ++++---- .../framework/plugins/windows/netstat.py | 6 +- .../framework/plugins/windows/poolscanner.py | 43 +++--- .../framework/plugins/windows/psscan.py | 58 ++++---- .../framework/plugins/windows/psxview.py | 32 ++--- .../plugins/windows/registry/hivescan.py | 9 +- .../framework/plugins/windows/svcscan.py | 5 +- .../framework/plugins/windows/symlinkscan.py | 20 +-- .../framework/plugins/windows/thrdscan.py | 10 +- .../plugins/windows/windowstations.py | 9 +- 22 files changed, 262 insertions(+), 358 deletions(-) diff --git a/volatility3/framework/plugins/windows/bigpools.py b/volatility3/framework/plugins/windows/bigpools.py index 9da702ae0..fabdd4306 100644 --- a/volatility3/framework/plugins/windows/bigpools.py +++ b/volatility3/framework/plugins/windows/bigpools.py @@ -21,7 +21,7 @@ class BigPools(interfaces.plugins.PluginInterface): """List big page pools.""" _required_framework_version = (2, 0, 0) - _version = (1, 1, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -50,8 +50,7 @@ class BigPools(interfaces.plugins.PluginInterface): def list_big_pools( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, tags: Optional[list] = None, show_free: bool = False, ): @@ -59,19 +58,13 @@ class BigPools(interfaces.plugins.PluginInterface): 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 + kernel_module_name: The name of the module for the kernel tags: An optional list of pool tags to filter big page pool tags by Yields: A big page pool object """ - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] big_page_table_offset = ntkrnlmp.get_symbol("PoolBigPageTable").address big_page_table = ntkrnlmp.object( @@ -87,8 +80,10 @@ class BigPools(interfaces.plugins.PluginInterface): big_page_table_type = ntkrnlmp.get_type("_POOL_TRACKER_BIG_PAGES") except exceptions.SymbolError: # We have to manually load a symbol table - is_vista_or_later = versions.is_vista_or_later(context, symbol_table) - is_win10 = versions.is_win10(context, symbol_table) + is_vista_or_later = versions.is_vista_or_later( + context, ntkrnlmp.symbol_table_name + ) + is_win10 = versions.is_win10(context, ntkrnlmp.symbol_table_name) if is_win10: big_pools_json_filename = "bigpools-win10" elif is_vista_or_later: @@ -96,7 +91,7 @@ class BigPools(interfaces.plugins.PluginInterface): else: big_pools_json_filename = "bigpools" - if symbols.symbol_table_is_64bit(context, symbol_table): + if symbols.symbol_table_is_64bit(context, ntkrnlmp.symbol_table_name): big_pools_json_filename += "-x64" else: big_pools_json_filename += "-x86" @@ -104,16 +99,17 @@ class BigPools(interfaces.plugins.PluginInterface): new_table_name = intermed.IntermediateSymbolTable.create( context=context, config_path=configuration.path_join( - context.symbol_space[symbol_table].config_path, "bigpools" + context.symbol_space[ntkrnlmp.symbol_table_name].config_path, + "bigpools", ), sub_path=os.path.join("windows", "bigpools"), filename=big_pools_json_filename, - table_mapping={"nt_symbols": symbol_table}, + table_mapping={"nt_symbols": ntkrnlmp.symbol_table_name}, class_types={ "_POOL_TRACKER_BIG_PAGES": extensions.pool.POOL_TRACKER_BIG_PAGES }, ) - module = context.module(new_table_name, layer_name, offset=0) + module = context.module(new_table_name, ntkrnlmp.layer_name, offset=0) big_page_table_type = module.get_type("_POOL_TRACKER_BIG_PAGES") big_pools = ntkrnlmp.object( @@ -136,12 +132,10 @@ class BigPools(interfaces.plugins.PluginInterface): tags = [tag for tag in self.config["tags"].split(",")] else: tags = None - kernel = self.context.modules[self.config["kernel"]] for big_pool in self.list_big_pools( context=self.context, - layer_name=kernel.layer_name, - symbol_table=kernel.symbol_table_name, + kernel_module_name=self.config["kernel"], tags=tags, show_free=self.config.get("show-free"), ): diff --git a/volatility3/framework/plugins/windows/callbacks.py b/volatility3/framework/plugins/windows/callbacks.py index 08e343ec7..c832df66a 100644 --- a/volatility3/framework/plugins/windows/callbacks.py +++ b/volatility3/framework/plugins/windows/callbacks.py @@ -28,7 +28,7 @@ class Callbacks(interfaces.plugins.PluginInterface): """Lists kernel callbacks and notification routines.""" _required_framework_version = (2, 0, 0) - _version = (2, 0, 1) + _version = (3, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -42,7 +42,7 @@ class Callbacks(interfaces.plugins.PluginInterface): name="ssdt", plugin=ssdt.SSDT, version=(2, 0, 0) ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), requirements.PluginRequirement( name="driverirp", plugin=driverirp.DriverIrp, version=(1, 0, 0) @@ -211,8 +211,7 @@ class Callbacks(interfaces.plugins.PluginInterface): def scan( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, + kernel_module_name: str, callback_symbol_table: str, ) -> Iterable[ Tuple[ @@ -225,18 +224,21 @@ class Callbacks(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols + kernel_module_name: Name of the module for the kernel callback_symbol_table: The name of the table containing the callback object symbols (_SHUTDOWN_PACKET etc.) Returns: A list of callback objects found by scanning the `layer_name` layer for callback pool signatures """ + kernel = context.modules[kernel_module_name] + is_vista_or_later = versions.is_vista_or_later( - context=context, symbol_table=nt_symbol_table + context=context, symbol_table=kernel.symbol_table_name ) - type_map = handles.Handles.get_type_map(context, layer_name, nt_symbol_table) + type_map = handles.Handles.get_type_map( + context, kernel.layer_name, kernel.symbol_table_name + ) constraints = cls.create_callback_scan_constraints( context, callback_symbol_table, is_vista_or_later @@ -247,7 +249,7 @@ class Callbacks(interfaces.plugins.PluginInterface): mem_object, _header, ) in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, nt_symbol_table, constraints + context, kernel_module_name, constraints ): try: if isinstance(mem_object, callbacks._SHUTDOWN_PACKET): @@ -347,31 +349,24 @@ class Callbacks(interfaces.plugins.PluginInterface): def list_notify_routines( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[Tuple[str, int, Optional[str]]]: """Lists all kernel notification routines. 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 + kernel_module_name: The name of the module of the kernel callback_table_name: The name of the table containing the callback symbols Yields: A name, location and optional detail string """ - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] is_vista_or_later = versions.is_vista_or_later( - context=context, symbol_table=symbol_table + context=context, symbol_table=ntkrnlmp.symbol_table_name ) full_type_name = callback_table_name + constants.BANG + "_GENERIC_CALLBACK" @@ -416,20 +411,14 @@ class Callbacks(interfaces.plugins.PluginInterface): def _list_registry_callbacks_legacy( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[Tuple[str, int, None]]: """ Lists all registry callbacks from the old format via the CmpCallBackVector. """ - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] full_type_name = ( callback_table_name + constants.BANG + "_EX_CALLBACK_ROUTINE_BLOCK" ) @@ -467,20 +456,13 @@ class Callbacks(interfaces.plugins.PluginInterface): def _list_registry_callbacks_new( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[Tuple[str, int, Optional[str]]]: """ Lists all registry callbacks via the CallbackListHead. """ - - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] full_type_name = callback_table_name + constants.BANG + "_CM_CALLBACK_ENTRY" symbol_offset = ntkrnlmp.get_symbol("CallbackListHead").address @@ -504,40 +486,33 @@ class Callbacks(interfaces.plugins.PluginInterface): def list_registry_callbacks( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[Tuple[str, int, Optional[str]]]: """Lists all registry callbacks. 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 + kernel_module_name: The name of the module of the kernel callback_table_name: The name of the table containing the callback symbols Yields: A name, location and optional detail string """ - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] if ntkrnlmp.has_symbol("CmpCallBackVector") and ntkrnlmp.has_symbol( "CmpCallBackCount" ): yield from cls._list_registry_callbacks_legacy( - context, layer_name, symbol_table, callback_table_name + context, kernel_module_name, callback_table_name ) elif ntkrnlmp.has_symbol("CallbackListHead") and ntkrnlmp.has_symbol( "CmpCallBackCount" ): yield from cls._list_registry_callbacks_new( - context, layer_name, symbol_table, callback_table_name + context, kernel_module_name, callback_table_name ) else: symbols_to_check = [ @@ -552,14 +527,11 @@ class Callbacks(interfaces.plugins.PluginInterface): symbol_status = "exists" vollog.debug(f"symbol {symbol_name} {symbol_status}.") - return None - @classmethod def list_bugcheck_reason_callbacks( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[ Tuple[ @@ -572,20 +544,14 @@ class Callbacks(interfaces.plugins.PluginInterface): 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 + kernel_module_name: The name of the module of the kernel callback_table_name: The name of the table containing the callback symbols Yields: A name, location and optional detail string """ - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] try: list_offset = ntkrnlmp.get_symbol( @@ -599,11 +565,15 @@ class Callbacks(interfaces.plugins.PluginInterface): callback_table_name + constants.BANG + "_KBUGCHECK_REASON_CALLBACK_RECORD" ) callback_record = context.object( - object_type=full_type_name, offset=kvo + list_offset, layer_name=layer_name + object_type=full_type_name, + offset=ntkrnlmp.offset + list_offset, + layer_name=ntkrnlmp.layer_name, ) for callback in callback_record.Entry: - if not context.layers[layer_name].is_valid(callback.CallbackRoutine, 64): + if not context.layers[ntkrnlmp.layer_name].is_valid( + callback.CallbackRoutine, 64 + ): continue try: @@ -626,8 +596,7 @@ class Callbacks(interfaces.plugins.PluginInterface): def list_bugcheck_callbacks( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, callback_table_name: str, ) -> Iterable[ Tuple[ @@ -640,20 +609,13 @@ class Callbacks(interfaces.plugins.PluginInterface): 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 + kernel_module_name: The name of the module of the kernel callback_table_name: The name of the table containing the callback symbols Yields: A name, location and optional detail string """ - - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) + ntkrnlmp = context.modules[kernel_module_name] try: list_offset = ntkrnlmp.get_symbol("KeBugCheckCallbackListHead").address @@ -665,17 +627,20 @@ class Callbacks(interfaces.plugins.PluginInterface): callback_table_name + constants.BANG + "_KBUGCHECK_CALLBACK_RECORD" ) callback_record = context.object( - full_type_name, offset=kvo + list_offset, layer_name=layer_name + full_type_name, + offset=ntkrnlmp.offset + list_offset, + layer_name=ntkrnlmp.layer_name, ) for callback in callback_record.Entry: - if not context.layers[layer_name].is_valid(callback.CallbackRoutine, 64): + if not context.layers[ntkrnlmp.layer_name].is_valid( + callback.CallbackRoutine, 64 + ): continue try: - component = context.object( - symbol_table + constants.BANG + "string", - layer_name=layer_name, + component = ntkrnlmp.object( + "string", offset=callback.Component, max_length=64, errors="replace", @@ -708,8 +673,7 @@ class Callbacks(interfaces.plugins.PluginInterface): for callback_method in callback_methods: for callback_type, callback_address, callback_detail in callback_method( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], callback_symbol_table, ): if callback_detail is None: diff --git a/volatility3/framework/plugins/windows/cmdscan.py b/volatility3/framework/plugins/windows/cmdscan.py index be955374b..b7eab79fb 100644 --- a/volatility3/framework/plugins/windows/cmdscan.py +++ b/volatility3/framework/plugins/windows/cmdscan.py @@ -24,7 +24,7 @@ class CmdScan(interfaces.plugins.PluginInterface): """Looks for Windows Command History lists""" _required_framework_version = (2, 4, 0) - _version = (1, 0, 0) + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -39,7 +39,7 @@ class CmdScan(interfaces.plugins.PluginInterface): name="pslist", component=pslist.PsList, version=(3, 0, 0) ), requirements.PluginRequirement( - name="consoles", plugin=consoles.Consoles, version=(2, 0, 0) + name="consoles", plugin=consoles.Consoles, version=(3, 0, 0) ), requirements.BooleanRequirement( name="no_registry", @@ -83,9 +83,8 @@ class CmdScan(interfaces.plugins.PluginInterface): def get_command_history( cls, context: interfaces.context.ContextInterface, - kernel_layer_name: str, - kernel_symbol_table_name: str, config_path: str, + kernel_module_name: str, procs: Generator[interfaces.objects.ObjectInterface, None, None], max_history: Set[int], ) -> Tuple[ @@ -97,8 +96,6 @@ class CmdScan(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - kernel_layer_name: The name of the layer on which to operate - kernel_symbol_table_name: The name of the table containing the kernel symbols config_path: The config path where to find symbol files procs: List of process objects max_history: An initial set of CommandHistorySize values @@ -135,9 +132,8 @@ class CmdScan(interfaces.plugins.PluginInterface): if conhost_symbol_table is None: conhost_symbol_table = consoles.Consoles.create_conhost_symbol_table( context, - kernel_layer_name, - kernel_symbol_table_name, config_path, + kernel_module_name, proc_layer_name, conhostexe_base, ) @@ -279,8 +275,6 @@ class CmdScan(interfaces.plugins.PluginInterface): procs: the process list filtered to conhost.exe instances """ - kernel = self.context.modules[self.config["kernel"]] - max_history = set(self.config.get("max_history", [50])) no_registry = self.config.get("no_registry") @@ -302,9 +296,8 @@ class CmdScan(interfaces.plugins.PluginInterface): command_history_properties, ) in self.get_command_history( self.context, - kernel.layer_name, - kernel.symbol_table_name, self.config_path, + self.config["kernel"], procs, max_history, ): diff --git a/volatility3/framework/plugins/windows/consoles.py b/volatility3/framework/plugins/windows/consoles.py index ff5875354..8236d87be 100644 --- a/volatility3/framework/plugins/windows/consoles.py +++ b/volatility3/framework/plugins/windows/consoles.py @@ -31,7 +31,7 @@ class Consoles(interfaces.plugins.PluginInterface): _required_framework_version = (2, 4, 0) # 2.0.0 - change the signature of `get_console_settings_from_registry` - _version = (2, 0, 0) + _version = (3, 0, 0) @classmethod def get_requirements(cls): @@ -128,9 +128,8 @@ class Consoles(interfaces.plugins.PluginInterface): def determine_conhost_version( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, config_path: str, + kernel_module_name: str, conhost_layer_name: str, conhost_base: int, ) -> Tuple[Optional[str], Dict[str, Type]]: @@ -139,9 +138,8 @@ class Consoles(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols config_path: The config path where to find symbol files + kernel_module_name: The name of the module for the kernel conhost_layer_name: The name of the conhot process memory layer conhost_base: the base address of conhost.exe @@ -149,8 +147,10 @@ class Consoles(interfaces.plugins.PluginInterface): The filename of the symbol table to use and the associated class types. """ + kernel = context.modules[kernel_module_name] + is_64bit = symbols.symbol_table_is_64bit( - context=context, symbol_table_name=nt_symbol_table + context=context, symbol_table_name=kernel.symbol_table_name ) if is_64bit: @@ -158,9 +158,9 @@ class Consoles(interfaces.plugins.PluginInterface): else: arch = "x86" - vers = info.Info.get_version_structure(context, layer_name, nt_symbol_table) + vers = info.Info.get_version_structure(context, kernel_module_name) - kuser = info.Info.get_kuser_structure(context, layer_name, nt_symbol_table) + kuser = info.Info.get_kuser_structure(context, kernel_module_name) try: vers_minor_version = int(vers.MinorVersion) @@ -247,7 +247,7 @@ class Consoles(interfaces.plugins.PluginInterface): ) except (exceptions.InvalidAddressException, TypeError, AttributeError): # the following is IntelLayer specific and might need to be adapted to other architectures. - physical_layer_name = context.layers[layer_name].config.get( + physical_layer_name = context.layers[kernel.layer_name].config.get( "memory_layer", None ) if physical_layer_name: @@ -318,9 +318,8 @@ class Consoles(interfaces.plugins.PluginInterface): def create_conhost_symbol_table( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, config_path: str, + kernel_module_name: str, conhost_layer_name: str, conhost_base: int, ) -> str: @@ -328,20 +327,20 @@ class Consoles(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols config_path: The config path where to find symbol files + kernel_module_name: The name of the module of the kernel Returns: The name of the constructed symbol table """ - table_mapping = {"nt_symbols": nt_symbol_table} + kernel = context.modules[kernel_module_name] + + table_mapping = {"nt_symbols": kernel.symbol_table_name} symbol_filename, class_types = cls.determine_conhost_version( context, - layer_name, - nt_symbol_table, config_path, + kernel_module_name, conhost_layer_name, conhost_base, ) @@ -366,9 +365,8 @@ class Consoles(interfaces.plugins.PluginInterface): def get_console_info( cls, context: interfaces.context.ContextInterface, - kernel_layer_name: str, - kernel_table_name: str, config_path: str, + kernel_module_name: str, procs: Generator[interfaces.objects.ObjectInterface, None, None], max_history: Set[int], max_buffers: Set[int], @@ -385,9 +383,8 @@ class Consoles(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - kernel_layer_name: The name of the layer on which to operate - kernel_table_name: The name of the table containing the kernel symbols config_path: The config path where to find symbol files + kernel_module_name: The name of the module for the kernel procs: list of process objects max_history: an initial set of CommandHistorySize values max_buffers: an initial list of HistoryBufferMax values @@ -427,9 +424,8 @@ class Consoles(interfaces.plugins.PluginInterface): if conhost_symbol_table is None: conhost_symbol_table = cls.create_conhost_symbol_table( context, - kernel_layer_name, - kernel_table_name, config_path, + kernel_module_name, proc_layer_name, conhostexe_base, ) @@ -486,7 +482,7 @@ class Consoles(interfaces.plugins.PluginInterface): console_properties.append( { "level": 1, - "name": "_CONSOLE_INFORMATION.ScreenX", + "kernel_module_nameme": "_CONSOLE_INFORMATION.ScreenX", "address": console_info.ScreenX.vol.offset, "data": console_info.ScreenX, } @@ -810,8 +806,7 @@ class Consoles(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from config_path: The config path where to find symbol files - kernel_layer_name: The name of the layer on which to operate - kernel_symbol_table_name: The name of the table containing the kernel symbols + kernel_module_name: The name of the module for the kernel max_history: an initial set of CommandHistorySize values max_buffers: an initial list of HistoryBufferMax values @@ -853,8 +848,6 @@ class Consoles(interfaces.plugins.PluginInterface): procs: the process list filtered to conhost.exe instances """ - kernel = self.context.modules[self.config["kernel"]] - max_history = set(self.config.get("max_history", [50])) max_buffers = set(self.config.get("max_buffers", [4])) no_registry = self.config.get("no_registry") @@ -874,9 +867,8 @@ class Consoles(interfaces.plugins.PluginInterface): proc = None for proc, console_info, console_properties in self.get_console_info( self.context, - kernel.layer_name, - kernel.symbol_table_name, self.config_path, + self.config["kernel"], procs, max_history, max_buffers, diff --git a/volatility3/framework/plugins/windows/dlllist.py b/volatility3/framework/plugins/windows/dlllist.py index 7efc8a7eb..b1c6f2f05 100644 --- a/volatility3/framework/plugins/windows/dlllist.py +++ b/volatility3/framework/plugins/windows/dlllist.py @@ -37,13 +37,13 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): name="pslist", component=pslist.PsList, version=(3, 0, 0) ), requirements.VersionRequirement( - name="psscan", component=psscan.PsScan, version=(1, 1, 0) + name="psscan", component=psscan.PsScan, version=(2, 0, 0) ), requirements.VersionRequirement( name="pedump", component=pedump.PEDump, version=(2, 0, 0) ), requirements.VersionRequirement( - name="info", component=info.Info, version=(1, 0, 0) + name="info", component=info.Info, version=(2, 0, 0) ), requirements.ListRequirement( name="pid", @@ -85,11 +85,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): self.context, self.config_path, "windows", "pe", class_types=pe.class_types ) - kernel = self.context.modules[self.config["kernel"]] - - kuser = info.Info.get_kuser_structure( - self.context, kernel.layer_name, kernel.symbol_table_name - ) + kuser = info.Info.get_kuser_structure(self.context, self.config["kernel"]) nt_major_version = int(kuser.NtMajorVersion) nt_minor_version = int(kuser.NtMinorVersion) @@ -209,8 +205,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): if self.config["offset"]: procs = psscan.PsScan.scan_processes( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], filter_func=psscan.PsScan.create_offset_filter( self.context, kernel.layer_name, diff --git a/volatility3/framework/plugins/windows/driverscan.py b/volatility3/framework/plugins/windows/driverscan.py index f179ff548..86db8d72b 100644 --- a/volatility3/framework/plugins/windows/driverscan.py +++ b/volatility3/framework/plugins/windows/driverscan.py @@ -25,7 +25,7 @@ class DriverScan(interfaces.plugins.PluginInterface): architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), ] @@ -48,15 +48,11 @@ class DriverScan(interfaces.plugins.PluginInterface): kernel = context.modules[kernel_module_name] - symbol_table_name = kernel.symbol_table_name - layer_name = kernel.layer_name - constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table_name, [b"Dri\xf6", b"Driv"] + kernel.symbol_table_name, [b"Dri\xf6", b"Driv"] ) - module = context.module(symbol_table_name, layer_name, 0) - driver_start_offset = module.get_type("_DRIVER_OBJECT").relative_child_offset( + driver_start_offset = kernel.get_type("_DRIVER_OBJECT").relative_child_offset( "DriverStart" ) @@ -65,7 +61,7 @@ class DriverScan(interfaces.plugins.PluginInterface): ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table_name, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result diff --git a/volatility3/framework/plugins/windows/filescan.py b/volatility3/framework/plugins/windows/filescan.py index 82566361d..abec2a92e 100644 --- a/volatility3/framework/plugins/windows/filescan.py +++ b/volatility3/framework/plugins/windows/filescan.py @@ -14,7 +14,7 @@ class FileScan(interfaces.plugins.PluginInterface): """Scans for file objects present in a particular windows memory image.""" _required_framework_version = (2, 0, 0) - _version = (1, 0, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -25,7 +25,7 @@ class FileScan(interfaces.plugins.PluginInterface): architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), ] @@ -33,36 +33,32 @@ class FileScan(interfaces.plugins.PluginInterface): def scan_files( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> Iterable[interfaces.objects.ObjectInterface]: """Scans for file objects using the poolscanner module and constraints. 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 + kernel_module_name: The name of the module for the kernel Returns: A list of File objects as found from the `layer_name` layer based on File pool signatures """ + kernel = context.modules[kernel_module_name] + constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table, [b"Fil\xe5", b"File"] + kernel.symbol_table_name, [b"Fil\xe5", b"File"] ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result yield mem_object def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - - for fileobj in self.scan_files( - self.context, kernel.layer_name, kernel.symbol_table_name - ): + for fileobj in self.scan_files(self.context, self.config["kernel"]): try: file_name = fileobj.FileName.String except exceptions.InvalidAddressException: diff --git a/volatility3/framework/plugins/windows/handles.py b/volatility3/framework/plugins/windows/handles.py index 6ab465439..e39c23a30 100644 --- a/volatility3/framework/plugins/windows/handles.py +++ b/volatility3/framework/plugins/windows/handles.py @@ -39,7 +39,7 @@ class Handles(interfaces.plugins.PluginInterface): name="pslist", plugin=pslist.PsList, version=(3, 0, 0) ), requirements.VersionRequirement( - name="psscan", component=psscan.PsScan, version=(1, 1, 0) + name="psscan", component=psscan.PsScan, version=(2, 0, 0) ), requirements.ListRequirement( name="pid", @@ -376,8 +376,7 @@ class Handles(interfaces.plugins.PluginInterface): if self.config["offset"]: procs = psscan.PsScan.scan_processes( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], filter_func=psscan.PsScan.create_offset_filter( self.context, kernel.layer_name, diff --git a/volatility3/framework/plugins/windows/indirect_system_calls.py b/volatility3/framework/plugins/windows/indirect_system_calls.py index c241a9b67..9f3fc4359 100644 --- a/volatility3/framework/plugins/windows/indirect_system_calls.py +++ b/volatility3/framework/plugins/windows/indirect_system_calls.py @@ -52,7 +52,7 @@ class IndirectSystemCalls(direct_system_calls.DirectSystemCalls): requirements.PluginRequirement( name="direct_system_calls", plugin=direct_system_calls.DirectSystemCalls, - version=(1, 0, 0), + version=(2, 0, 0), ), ] diff --git a/volatility3/framework/plugins/windows/info.py b/volatility3/framework/plugins/windows/info.py index e20af8114..a2e438c3f 100644 --- a/volatility3/framework/plugins/windows/info.py +++ b/volatility3/framework/plugins/windows/info.py @@ -17,7 +17,7 @@ class Info(plugins.PluginInterface): """Show OS & kernel details of the memory sample being analyzed.""" _required_framework_version = (2, 0, 0) - _version = (1, 0, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -56,6 +56,9 @@ class Info(plugins.PluginInterface): # FileLayer won't have dependencies pass + # FIXME - this needs to be deprecated. This is exactly the same + # as getting it from context.modules + # Deprecation warning will go once the API is overhauled @classmethod def get_kernel_module( cls, @@ -80,13 +83,12 @@ class Info(plugins.PluginInterface): cls, context: interfaces.context.ContextInterface, config_path: str, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> interfaces.objects.ObjectInterface: """Returns the KDDEBUGGER_DATA64 structure for a kernel""" - ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table) + ntkrnlmp = context.modules[kernel_module_name] - native_types = context.symbol_space[symbol_table].natives + native_types = context.symbol_space[ntkrnlmp.symbol_table_name].natives kdbg_offset = ntkrnlmp.get_symbol("KdDebuggerDataBlock").address @@ -102,7 +104,7 @@ class Info(plugins.PluginInterface): kdbg_obj = context.object( kdbg_table_name + constants.BANG + "_KDDEBUGGER_DATA64", offset=ntkrnlmp.offset + kdbg_offset, - layer_name=layer_name, + layer_name=ntkrnlmp.layer_name, ) return kdbg_obj @@ -111,16 +113,15 @@ class Info(plugins.PluginInterface): def get_kuser_structure( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> interfaces.objects.ObjectInterface: """Returns the _KUSER_SHARED_DATA structure for a kernel""" - virtual_layer = context.layers[layer_name] + ntkrnlmp = context.modules[kernel_module_name] + + virtual_layer = context.layers[ntkrnlmp.layer_name] if not isinstance(virtual_layer, layers.intel.Intel): raise TypeError("Virtual Layer is not an intel layer") - ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table) - # this is a hard-coded address in the Windows OS if virtual_layer.bits_per_register == 32: kuser_addr = 0xFFDF0000 @@ -129,7 +130,6 @@ class Info(plugins.PluginInterface): kuser = ntkrnlmp.object( object_type="_KUSER_SHARED_DATA", - layer_name=layer_name, offset=kuser_addr, absolute=True, ) @@ -140,17 +140,15 @@ class Info(plugins.PluginInterface): def get_version_structure( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> interfaces.objects.ObjectInterface: """Returns the KdVersionBlock information from a kernel""" - ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table) + ntkrnlmp = context.modules[kernel_module_name] vers_offset = ntkrnlmp.get_symbol("KdVersionBlock").address vers = ntkrnlmp.object( object_type="_DBGKD_GET_VERSION64", - layer_name=layer_name, offset=vers_offset, ) @@ -193,35 +191,38 @@ class Info(plugins.PluginInterface): def _generator(self): kernel = self.context.modules[self.config["kernel"]] - layer_name = kernel.layer_name - symbol_table = kernel.symbol_table_name - layer = self.context.layers[layer_name] - table = self.context.symbol_space[symbol_table] + kernel_layer = self.context.layers[kernel.layer_name] + symbol_table = self.context.symbol_space[kernel.symbol_table_name] kdbg = self.get_kdbg_structure( - self.context, self.config_path, layer_name, symbol_table + self.context, + self.config_path, + self.config["kernel"], ) - yield (0, ("Kernel Base", hex(layer.config["kernel_virtual_offset"]))) - yield (0, ("DTB", hex(layer.config["page_map_offset"]))) - yield (0, ("Symbols", table.config["isf_url"])) + yield (0, ("Kernel Base", hex(kernel_layer.config["kernel_virtual_offset"]))) + yield (0, ("DTB", hex(kernel_layer.config["page_map_offset"]))) + yield (0, ("Symbols", symbol_table.config["isf_url"])) yield ( 0, ( "Is64Bit", str( symbols.symbol_table_is_64bit( - context=self.context, symbol_table_name=symbol_table + context=self.context, symbol_table_name=kernel.symbol_table_name ) ), ), ) yield ( 0, - ("IsPAE", str(self.context.layers[layer_name].metadata.get("pae", False))), + ( + "IsPAE", + str(self.context.layers[kernel.layer_name].metadata.get("pae", False)), + ), ) - for i, layer in self.get_depends(self.context, layer_name): + for i, layer in self.get_depends(self.context, kernel.layer_name): yield (0, (layer.name, f"{i} {layer.__class__.__name__}")) if kdbg.Header.OwnerTag == 0x4742444B: @@ -229,23 +230,22 @@ class Info(plugins.PluginInterface): yield (0, ("NTBuildLab", kdbg.get_build_lab())) yield (0, ("CSDVersion", str(kdbg.get_csdversion()))) - vers = self.get_version_structure(self.context, layer_name, symbol_table) + vers = self.get_version_structure(self.context, self.config["kernel"]) yield (0, ("KdVersionBlock", hex(vers.vol.offset))) yield (0, ("Major/Minor", f"{vers.MajorVersion}.{vers.MinorVersion}")) yield (0, ("MachineType", str(vers.MachineType))) - ntkrnlmp = self.get_kernel_module(self.context, layer_name, symbol_table) + cpu_count_offset = kernel.get_symbol("KeNumberProcessors").address - cpu_count_offset = ntkrnlmp.get_symbol("KeNumberProcessors").address - - cpu_count = ntkrnlmp.object( - object_type="unsigned int", layer_name=layer_name, offset=cpu_count_offset + cpu_count = kernel.object( + object_type="unsigned int", + offset=cpu_count_offset, ) yield (0, ("KeNumberProcessors", str(cpu_count))) - kuser = self.get_kuser_structure(self.context, layer_name, symbol_table) + kuser = self.get_kuser_structure(self.context, self.config["kernel"]) yield (0, ("SystemTime", str(kuser.SystemTime.get_time()))) yield ( @@ -266,7 +266,7 @@ class Info(plugins.PluginInterface): # yield (0, ("SafeBootMode", "True" if kuser.SafeBootMode else "False")) nt_header = self.get_ntheader_structure( - self.context, self.config_path, layer_name + self.context, self.config_path, kernel.layer_name ) yield ( diff --git a/volatility3/framework/plugins/windows/modscan.py b/volatility3/framework/plugins/windows/modscan.py index ab1383ddf..7330b2cb4 100644 --- a/volatility3/framework/plugins/windows/modscan.py +++ b/volatility3/framework/plugins/windows/modscan.py @@ -32,7 +32,7 @@ class ModScan(modules.Modules): architectures=["Intel32", "Intel64"], ), requirements.VersionRequirement( - name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0) ), requirements.VersionRequirement( name="modules", component=modules.Modules, version=(3, 0, 0) @@ -81,7 +81,7 @@ class ModScan(modules.Modules): ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, kernel.layer_name, kernel.symbol_table_name, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result yield mem_object diff --git a/volatility3/framework/plugins/windows/mutantscan.py b/volatility3/framework/plugins/windows/mutantscan.py index 64d3b5470..ca6e26157 100644 --- a/volatility3/framework/plugins/windows/mutantscan.py +++ b/volatility3/framework/plugins/windows/mutantscan.py @@ -14,6 +14,7 @@ class MutantScan(interfaces.plugins.PluginInterface): """Scans for mutexes present in a particular windows memory image.""" _required_framework_version = (2, 0, 0) + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -24,7 +25,7 @@ class MutantScan(interfaces.plugins.PluginInterface): architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), ] @@ -32,36 +33,32 @@ class MutantScan(interfaces.plugins.PluginInterface): def scan_mutants( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> Iterable[interfaces.objects.ObjectInterface]: """Scans for mutants using the poolscanner module and constraints. 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 + kernel_module_name: The name of the module for the kernel Returns: A list of Mutant objects found by scanning memory for the Mutant pool signatures """ + kernel = context.modules[kernel_module_name] + constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table, [b"Mut\xe1", b"Muta"] + kernel.symbol_table_name, [b"Mut\xe1", b"Muta"] ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result yield mem_object def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - - for mutant in self.scan_mutants( - self.context, kernel.layer_name, kernel.symbol_table_name - ): + for mutant in self.scan_mutants(self.context, self.config["kernel"]): try: name = mutant.get_name() except (ValueError, exceptions.InvalidAddressException): diff --git a/volatility3/framework/plugins/windows/netscan.py b/volatility3/framework/plugins/windows/netscan.py index 6f98547d7..0c8523cce 100644 --- a/volatility3/framework/plugins/windows/netscan.py +++ b/volatility3/framework/plugins/windows/netscan.py @@ -23,7 +23,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): """Scans for network objects present in a particular windows memory image.""" _required_framework_version = (2, 0, 0) - _version = (1, 0, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -34,10 +34,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): architectures=["Intel32", "Intel64"], ), requirements.VersionRequirement( - name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0) ), requirements.VersionRequirement( - name="info", component=info.Info, version=(1, 0, 0) + name="info", component=info.Info, version=(2, 0, 0) ), requirements.VersionRequirement( name="verinfo", component=verinfo.VerInfo, version=(1, 0, 0) @@ -117,15 +117,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def determine_tcpip_version( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, + kernel_module_name: str, ) -> Tuple[str, Type]: """Tries to determine which symbol filename to use for the image's tcpip driver. The logic is partially taken from the info plugin. Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols + kernel_module_name: Name of the module for the kernel Returns: The filename of the symbol table to use. @@ -137,12 +135,14 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # therefore we determine the version based on the kernel version as testing # with several windows versions has showed this to work out correctly. + kernel = context.modules[kernel_module_name] + is_64bit = symbols.symbol_table_is_64bit( - context=context, symbol_table_name=nt_symbol_table + context=context, symbol_table_name=kernel.symbol_table_name ) is_18363_or_later = versions.is_win10_18363_or_later( - context=context, symbol_table=nt_symbol_table + context=context, symbol_table=kernel.symbol_table_name ) if is_64bit: @@ -150,9 +150,9 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): else: arch = "x86" - vers = info.Info.get_version_structure(context, layer_name, nt_symbol_table) + vers = info.Info.get_version_structure(context, kernel_module_name) - kuser = info.Info.get_kuser_structure(context, layer_name, nt_symbol_table) + kuser = info.Info.get_kuser_structure(context, kernel_module_name) try: vers_minor_version = int(vers.MinorVersion) @@ -259,7 +259,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): "Requiring further version inspection due to OS version by checking tcpip.sys's FileVersion header" ) # the following is IntelLayer specific and might need to be adapted to other architectures. - physical_layer_name = context.layers[layer_name].config.get( + physical_layer_name = context.layers[kernel.layer_name].config.get( "memory_layer", None ) if physical_layer_name: @@ -322,27 +322,26 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def create_netscan_symbol_table( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, + kernel_module_name: str, config_path: str, ) -> str: """Creates a symbol table for TCP Listeners and TCP/UDP Endpoints. Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols + kernel_module_name: Name of the module for the kernel config_path: The config path where to find symbol files Returns: The name of the constructed symbol table """ - table_mapping = {"nt_symbols": nt_symbol_table} + kernel = context.modules[kernel_module_name] + + table_mapping = {"nt_symbols": kernel.symbol_table_name} symbol_filename, class_types = cls.determine_tcpip_version( context, - layer_name, - nt_symbol_table, + kernel_module_name, ) return intermed.IntermediateSymbolTable.create( @@ -358,16 +357,14 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def scan( cls, context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, + kernel_module_name: str, netscan_symbol_table: str, ) -> Iterable[interfaces.objects.ObjectInterface]: """Scans for network objects using the poolscanner module and constraints. Args: context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - nt_symbol_table: The name of the table containing the kernel symbols + kernel_module_name: The name of the module for the kernel netscan_symbol_table: The name of the table containing the network object symbols (_TCP_LISTENER etc.) Returns: @@ -377,7 +374,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): constraints = cls.create_netscan_constraints(context, netscan_symbol_table) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, nt_symbol_table, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result yield mem_object @@ -385,16 +382,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def _generator(self, show_corrupt_results: Optional[bool] = None): """Generates the network objects for use in rendering.""" - kernel = self.context.modules[self.config["kernel"]] - netscan_symbol_table = self.create_netscan_symbol_table( - self.context, kernel.layer_name, kernel.symbol_table_name, self.config_path + self.context, self.config["kernel"], self.config_path ) for netw_obj in self.scan( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], netscan_symbol_table, ): vollog.debug( diff --git a/volatility3/framework/plugins/windows/netstat.py b/volatility3/framework/plugins/windows/netstat.py index 5daa6cc79..aaab4494c 100644 --- a/volatility3/framework/plugins/windows/netstat.py +++ b/volatility3/framework/plugins/windows/netstat.py @@ -34,7 +34,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): architectures=["Intel32", "Intel64"], ), requirements.VersionRequirement( - name="netscan", component=netscan.NetScan, version=(1, 0, 0) + name="netscan", component=netscan.NetScan, version=(2, 0, 0) ), requirements.VersionRequirement( name="modules", component=modules.Modules, version=(3, 0, 0) @@ -43,7 +43,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): name="pdbutil", component=pdbutil.PDBUtility, version=(1, 0, 0) ), requirements.VersionRequirement( - name="info", component=info.Info, version=(1, 0, 0) + name="info", component=info.Info, version=(2, 0, 0) ), requirements.VersionRequirement( name="verinfo", component=verinfo.VerInfo, version=(1, 0, 0) @@ -629,7 +629,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): kernel = self.context.modules[self.config["kernel"]] netscan_symbol_table = netscan.NetScan.create_netscan_symbol_table( - self.context, kernel.layer_name, kernel.symbol_table_name, self.config_path + self.context, self.config["kernel"], self.config_path ) tcpip_module = self.get_tcpip_module(self.context, self.config["kernel"]) diff --git a/volatility3/framework/plugins/windows/poolscanner.py b/volatility3/framework/plugins/windows/poolscanner.py index af19cd035..c78c47c43 100644 --- a/volatility3/framework/plugins/windows/poolscanner.py +++ b/volatility3/framework/plugins/windows/poolscanner.py @@ -129,7 +129,7 @@ class PoolScanner(plugins.PluginInterface): """A generic pool scanner plugin.""" _required_framework_version = (2, 0, 0) - _version = (1, 1, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -151,7 +151,7 @@ class PoolScanner(plugins.PluginInterface): constraints = self.builtin_constraints(symbol_table) for constraint, mem_object, header in self.generate_pool_scan( - self.context, kernel.layer_name, symbol_table, constraints + self.context, self.config["kernel"], constraints ): # generate some type-specific info for sanity checking if constraint.object_type == "Process": @@ -365,8 +365,7 @@ class PoolScanner(plugins.PluginInterface): def generate_pool_scan_extended( cls, context: interfaces.context.ContextInterface, - kernel_layer_name: str, - kernel_symbol_table_name: str, + kernel_module_name: str, object_symbol_table_name: str, constraints: List[PoolConstraint], ) -> Generator[ @@ -384,41 +383,42 @@ class PoolScanner(plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from - kernel_layer_name: The name of the base kernel layer - kernel_symbol_table_name: The name of the table containing the kernel symbols + kernel_module_name: The name of the module for the kernel object_symbol_table_name: The name of the symbol table for the object being scanned for constraints: List of pool constraints used to limit the scan results Returns: Iterable of tuples, containing the constraint that matched, the object from memory, the object header used to determine the object """ + kernel = context.modules[kernel_module_name] + # get the object type map type_map = handles.Handles.get_type_map( context=context, - layer_name=kernel_layer_name, - symbol_table=kernel_symbol_table_name, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, ) cookie = handles.Handles.find_cookie( context=context, - layer_name=kernel_layer_name, - symbol_table=kernel_symbol_table_name, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, ) - is_windows_10 = versions.is_windows_10(context, kernel_symbol_table_name) + is_windows_10 = versions.is_windows_10(context, kernel.symbol_table_name) is_windows_8_or_later = versions.is_windows_8_or_later( - context, kernel_symbol_table_name + context, kernel.symbol_table_name ) # start off with the primary virtual layer - scan_layer = kernel_layer_name + scan_layer = kernel.layer_name # switch to a non-virtual layer if necessary if not is_windows_10: scan_layer = context.layers[scan_layer].config["memory_layer"] if symbols.symbol_table_is_64bit( - context=context, symbol_table_name=kernel_symbol_table_name + context=context, symbol_table_name=kernel.symbol_table_name ): alignment = 0x10 else: @@ -433,12 +433,11 @@ class PoolScanner(plugins.PluginInterface): alignment=alignment, ): - # construct the object in its own layer, using its own types mem_objects = header.get_object( constraint=constraint, use_top_down=is_windows_8_or_later, - native_layer_name=kernel_layer_name, - kernel_symbol_table=kernel_symbol_table_name, + native_layer_name=kernel.layer_name, + kernel_symbol_table=kernel.symbol_table_name, ) for mem_object in mem_objects: @@ -472,8 +471,7 @@ class PoolScanner(plugins.PluginInterface): def generate_pool_scan( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, constraints: List[PoolConstraint], ) -> Generator[ Tuple[ @@ -489,17 +487,18 @@ class PoolScanner(plugins.PluginInterface): 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 + kernel_module_name: The name of the module for the kernel constraints: List of pool constraints used to limit the scan results Returns: Iterable of tuples, containing the constraint that matched, the object from memory, the object header used to determine the object """ + kernel = context.modules[kernel_module_name] + # repeat the symbol table to match the original `generate_pool_scan` behaviour yield from cls.generate_pool_scan_extended( - context, layer_name, symbol_table, symbol_table, constraints + context, kernel_module_name, kernel.symbol_table_name, constraints ) @classmethod diff --git a/volatility3/framework/plugins/windows/psscan.py b/volatility3/framework/plugins/windows/psscan.py index a19423029..45f935ceb 100644 --- a/volatility3/framework/plugins/windows/psscan.py +++ b/volatility3/framework/plugins/windows/psscan.py @@ -23,7 +23,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): """Scans for processes present in a particular windows memory image.""" _required_framework_version = (2, 3, 1) - _version = (1, 1, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -37,7 +37,10 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): name="pslist", plugin=pslist.PsList, version=(3, 0, 0) ), requirements.VersionRequirement( - name="info", component=info.Info, version=(1, 0, 0) + name="info", component=info.Info, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0) ), requirements.ListRequirement( name="pid", @@ -141,8 +144,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def scan_processes( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, filter_func: Callable[ [interfaces.objects.ObjectInterface], bool ] = lambda _: False, @@ -151,19 +153,20 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): 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 + kernel_module_name: The name of the module for the kernel Returns: A list of processes found by scanning the `layer_name` layer for process pool signatures """ + kernel = context.modules[kernel_module_name] + constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table, [b"Pro\xe3", b"Proc"] + kernel.symbol_table_name, [b"Pro\xe3", b"Proc"] ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result if not filter_func(mem_object): @@ -173,16 +176,14 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def virtual_process_from_physical( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, proc: interfaces.objects.ObjectInterface, ) -> Optional[interfaces.objects.ObjectInterface]: """Returns a virtual process from a physical addressed one 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 + kernel_module_name: The name of the module inside the kernel proc: the process object with physical address Returns: @@ -190,16 +191,10 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): """ - version = cls.get_osversion(context, layer_name, symbol_table) + ntkrnlmp = context.modules[kernel_module_name] + + version = cls.get_osversion(context, kernel_module_name) - # If it's WinXP->8.1 we have now a physical process address. - # We'll use the first thread to bounce back to the virtual process - kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None) - if not kvo: - raise ValueError( - "Intel layer does not have an associated kernel virtual offset, failing" - ) - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) tleoffset = ntkrnlmp.get_type("_ETHREAD").relative_child_offset( "ThreadListEntry" ) @@ -208,7 +203,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # If (and only if) we're dealing with 64-bit Windows 7 SP1 # then add the other commonly seen member offset to the list - bits = context.layers[layer_name].bits_per_register + bits = context.layers[ntkrnlmp.layer_name].bits_per_register if version == (6, 1, 7601) and bits == 64: offsets.append(tleoffset + 8) @@ -225,7 +220,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # Sanity check the bounce. # This compares the original offset with the new one (translated from virtual layer) (_, _, ph_offset, _, _) = list( - context.layers[layer_name].mapping( + context.layers[ntkrnlmp.layer_name].mapping( offset=virtual_process.vol.offset, length=0 ) )[0] @@ -237,23 +232,20 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def get_osversion( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> Tuple[int, int, int]: """Returns the complete OS version (MAJ,MIN,BUILD) 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 - + kernel_module_name: The name of the module for the kernel Returns: A tuple with (MAJ,MIN,BUILD) """ - kuser = info.Info.get_kuser_structure(context, layer_name, symbol_table) + kuser = info.Info.get_kuser_structure(context, kernel_module_name) nt_major_version = int(kuser.NtMajorVersion) nt_minor_version = int(kuser.NtMinorVersion) - vers = info.Info.get_version_structure(context, layer_name, symbol_table) + vers = info.Info.get_version_structure(context, kernel_module_name) build = vers.MinorVersion return (nt_major_version, nt_minor_version, build) @@ -268,8 +260,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): for proc in self.scan_processes( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], filter_func=pslist.PsList.create_pid_filter(self.config.get("pid", None)), ): file_output = "Disabled" @@ -281,8 +272,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): try: vproc = self.virtual_process_from_physical( self.context, - kernel.layer_name, - kernel.symbol_table_name, + self.config["kernel"], proc, ) except exceptions.PagedInvalidAddressException: diff --git a/volatility3/framework/plugins/windows/psxview.py b/volatility3/framework/plugins/windows/psxview.py index 89ef897cb..3377163f7 100644 --- a/volatility3/framework/plugins/windows/psxview.py +++ b/volatility3/framework/plugins/windows/psxview.py @@ -11,7 +11,6 @@ from volatility3.framework.renderers import TreeGrid, format_hints from volatility3.framework.symbols.windows import extensions from volatility3.plugins.windows import ( handles, - info, pslist, psscan, thrdscan, @@ -49,14 +48,11 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter description="Windows kernel", architectures=["Intel32", "Intel64"], ), - requirements.VersionRequirement( - name="info", component=info.Info, version=(1, 0, 0) - ), requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(3, 0, 0) ), requirements.VersionRequirement( - name="psscan", component=psscan.PsScan, version=(1, 0, 0) + name="psscan", component=psscan.PsScan, version=(2, 0, 0) ), requirements.VersionRequirement( name="thrdscan", component=thrdscan.ThrdScan, version=(1, 0, 0) @@ -114,10 +110,10 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter return self._proc_list_to_dict(tasks) def _check_psscan( - self, layer_name: str, symbol_table: str + self, ) -> Dict[int, extensions.EPROCESS]: res = psscan.PsScan.scan_processes( - context=self.context, layer_name=layer_name, symbol_table=symbol_table + context=self.context, kernel_module_name=self.config["kernel"] ) return self._proc_list_to_dict(res) @@ -144,20 +140,24 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter return self._proc_list_to_dict(ret) def _check_csrss_handles( - self, tasks: Iterable[extensions.EPROCESS], layer_name: str, symbol_table: str + self, tasks: Iterable[extensions.EPROCESS] ) -> Dict[int, extensions.EPROCESS]: ret: List[extensions.EPROCESS] = [] + kernel = self.context.modules[self.config["kernel"]] + handles_plugin = handles.Handles( context=self.context, config_path=self.config_path ) - type_map = handles_plugin.get_type_map(self.context, layer_name, symbol_table) + type_map = handles_plugin.get_type_map( + self.context, kernel.layer_name, kernel.symbol_table_name + ) cookie = handles_plugin.find_cookie( context=self.context, - layer_name=layer_name, - symbol_table=symbol_table, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, ) for p in tasks: @@ -179,8 +179,6 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter return self._proc_list_to_dict(ret) def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - kdbg_list_processes = list( pslist.PsList.list_processes( context=self.context, kernel_module_name=self.config["kernel"] @@ -191,13 +189,9 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter processes: Dict[str, Dict[int, extensions.EPROCESS]] = {} processes["pslist"] = self._check_pslist(kdbg_list_processes) - processes["psscan"] = self._check_psscan( - kernel.layer_name, kernel.symbol_table_name - ) + processes["psscan"] = self._check_psscan() processes["thrdscan"] = self._check_thrdscan() - processes["csrss"] = self._check_csrss_handles( - kdbg_list_processes, kernel.layer_name, kernel.symbol_table_name - ) + processes["csrss"] = self._check_csrss_handles(kdbg_list_processes) # Unique set of all offsets from all sources offsets = set(chain(*(mapping.keys() for mapping in processes.values()))) diff --git a/volatility3/framework/plugins/windows/registry/hivescan.py b/volatility3/framework/plugins/windows/registry/hivescan.py index d91eeafc2..a28ab19de 100644 --- a/volatility3/framework/plugins/windows/registry/hivescan.py +++ b/volatility3/framework/plugins/windows/registry/hivescan.py @@ -26,10 +26,10 @@ class HiveScan(interfaces.plugins.PluginInterface): architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), requirements.PluginRequirement( - name="bigpools", plugin=bigpools.BigPools, version=(1, 0, 0) + name="bigpools", plugin=bigpools.BigPools, version=(2, 0, 0) ), ] @@ -62,8 +62,7 @@ class HiveScan(interfaces.plugins.PluginInterface): for pool in bigpools.BigPools.list_big_pools( context, - layer_name=kernel.layer_name, - symbol_table=kernel.symbol_table_name, + kernel_module_name=kernel_name, tags=["CM10"], ): cmhive = ntkrnlmp.object( @@ -77,7 +76,7 @@ class HiveScan(interfaces.plugins.PluginInterface): ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, kernel.layer_name, kernel.symbol_table_name, constraints + context, kernel_name, constraints ): _constraint, mem_object, _header = result yield mem_object diff --git a/volatility3/framework/plugins/windows/svcscan.py b/volatility3/framework/plugins/windows/svcscan.py index 653995f90..915850574 100644 --- a/volatility3/framework/plugins/windows/svcscan.py +++ b/volatility3/framework/plugins/windows/svcscan.py @@ -20,7 +20,7 @@ from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import intermed from volatility3.framework.symbols.windows import versions from volatility3.framework.symbols.windows.extensions import services as services_types -from volatility3.plugins.windows import poolscanner, pslist +from volatility3.plugins.windows import pslist from volatility3.plugins.windows.registry import hivelist vollog = logging.getLogger(__name__) @@ -53,9 +53,6 @@ class SvcScan(interfaces.plugins.PluginInterface): requirements.PluginRequirement( name="pslist", plugin=pslist.PsList, version=(3, 0, 0) ), - requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) - ), requirements.PluginRequirement( name="hivelist", plugin=hivelist.HiveList, version=(2, 0, 0) ), diff --git a/volatility3/framework/plugins/windows/symlinkscan.py b/volatility3/framework/plugins/windows/symlinkscan.py index 89fdf142e..459129843 100644 --- a/volatility3/framework/plugins/windows/symlinkscan.py +++ b/volatility3/framework/plugins/windows/symlinkscan.py @@ -17,6 +17,8 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa _required_framework_version = (2, 0, 0) + _version = (2, 0, 0) + @classmethod def get_requirements(cls): return [ @@ -25,14 +27,16 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa description="Windows kernel", architectures=["Intel32", "Intel64"], ), + requirements.VersionRequirement( + name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0) + ), ] @classmethod def scan_symlinks( cls, context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, + kernel_module_name: str, ) -> Iterable[interfaces.objects.ObjectInterface]: """Scans for links using the poolscanner module and constraints. @@ -45,22 +49,20 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa A list of symlink objects found by scanning memory for the Symlink pool signatures """ + kernel = context.modules[kernel_module_name] + constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table, [b"Sym\xe2", b"Symb"] + kernel.symbol_table_name, [b"Sym\xe2", b"Symb"] ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table, constraints + context, kernel_module_name, constraints ): _constraint, mem_object, _header = result yield mem_object def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - - for link in self.scan_symlinks( - self.context, kernel.layer_name, kernel.symbol_table_name - ): + for link in self.scan_symlinks(self.context, self.config["kernel"]): try: from_name = link.get_link_name() except (ValueError, exceptions.InvalidAddressException): diff --git a/volatility3/framework/plugins/windows/thrdscan.py b/volatility3/framework/plugins/windows/thrdscan.py index c0963e754..b43593401 100644 --- a/volatility3/framework/plugins/windows/thrdscan.py +++ b/volatility3/framework/plugins/windows/thrdscan.py @@ -34,7 +34,7 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0) + name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0) ), ] @@ -54,16 +54,14 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) A list of _ETHREAD objects found by scanning memory for the "Thre" / "Thr\\xE5" pool signatures """ - module = context.modules[module_name] - layer_name = module.layer_name - symbol_table = module.symbol_table_name + kernel = context.modules[module_name] constraints = poolscanner.PoolScanner.builtin_constraints( - symbol_table, [b"Thr\xe5", b"Thre"] + kernel.symbol_table_name, [b"Thr\xe5", b"Thre"] ) for result in poolscanner.PoolScanner.generate_pool_scan( - context, layer_name, symbol_table, constraints + context, module_name, constraints ): _constraint, mem_object, _header = result yield mem_object diff --git a/volatility3/framework/plugins/windows/windowstations.py b/volatility3/framework/plugins/windows/windowstations.py index a077ef2fc..9d666e3dd 100644 --- a/volatility3/framework/plugins/windows/windowstations.py +++ b/volatility3/framework/plugins/windows/windowstations.py @@ -46,6 +46,12 @@ class WindowStations(interfaces.plugins.PluginInterface): description="Windows kernel", architectures=["Intel32", "Intel64"], ), + requirements.VersionRequirement( + name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="modules", component=modules.Modules, version=(3, 0, 0) + ), ] @staticmethod @@ -152,8 +158,7 @@ class WindowStations(interfaces.plugins.PluginInterface): for result in poolscanner.PoolScanner.generate_pool_scan_extended( context=context, - kernel_layer_name=kernel.layer_name, - kernel_symbol_table_name=kernel.symbol_table_name, + kernel_module_name=kernel_module_name, object_symbol_table_name=gui_table_name, constraints=constraints, ):