From 8da046530fd7ccd3f8a75fbe4e0b63d213207117 Mon Sep 17 00:00:00 2001 From: atcuno Date: Thu, 18 Jul 2024 14:57:31 -0500 Subject: [PATCH] Address feedback --- .../framework/plugins/windows/svcdiff.py | 22 ++++--- .../framework/plugins/windows/svclist.py | 58 +++++++++++-------- .../framework/plugins/windows/svcscan.py | 43 ++++++++------ 3 files changed, 69 insertions(+), 54 deletions(-) diff --git a/volatility3/framework/plugins/windows/svcdiff.py b/volatility3/framework/plugins/windows/svcdiff.py index 809064946..c41a7b86c 100644 --- a/volatility3/framework/plugins/windows/svcdiff.py +++ b/volatility3/framework/plugins/windows/svcdiff.py @@ -20,8 +20,7 @@ from volatility3.framework.symbols.windows import versions vollog = logging.getLogger(__name__) - -class SvcDiff(svclist.SvcList, svcscan.SvcScan): +class SvcDiff(svcscan.SvcScan): """Compares services found through list walking versus scanning to find rootkits""" _required_framework_version = (2, 4, 0) @@ -39,22 +38,23 @@ class SvcDiff(svclist.SvcList, svcscan.SvcScan): name="svclist", component=svclist.SvcList, version=(1, 0, 0) ), requirements.VersionRequirement( - name="svcscan", component=svcscan.SvcScan, version=(2, 0, 0) + name="svcscan", component=svcscan.SvcScan, version=(3, 0, 0) ), ] def _generator(self): """ - Finds services by walking the services.exe list on supported Windows 10 versions + On Windows 10 version 15063+ 64bit Windows memory samples, walk the services list + and scan for services then report differences """ - kernel = self.context.modules[self.config["kernel"]] + kernel, service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info() if not symbols.symbol_table_is_64bit( self.context, kernel.symbol_table_name ) or not versions.is_win10_15063_or_later( context=self.context, symbol_table=kernel.symbol_table_name ): - vollog.info( + vollog.warning( "This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples" ) return @@ -63,18 +63,16 @@ class SvcDiff(svclist.SvcList, svcscan.SvcScan): from_list = set() records = {} - service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info() - # collect unique service names from scanning - for service in self.service_scan( - service_table_name, service_binary_dll_map, filter_func + for service in svcscan.SvcScan.service_scan( + self.context, kernel, service_table_name, service_binary_dll_map, filter_func ): from_scan.add(service[6]) records[service[6]] = service # collect services from listing walking - for service in self.service_list( - service_table_name, service_binary_dll_map, filter_func + for service in svclist.SvcList.service_list( + self.context, kernel, service_table_name, service_binary_dll_map, filter_func ): from_list.add(service[6]) diff --git a/volatility3/framework/plugins/windows/svclist.py b/volatility3/framework/plugins/windows/svclist.py index 1938dd182..832b3d129 100644 --- a/volatility3/framework/plugins/windows/svclist.py +++ b/volatility3/framework/plugins/windows/svclist.py @@ -4,7 +4,7 @@ import logging -from typing import List +from typing import List, Optional, Tuple from volatility3.framework import interfaces, exceptions, symbols from volatility3.framework.configuration import requirements @@ -20,16 +20,26 @@ class SvcList(svcscan.SvcScan): _version = (1, 0, 0) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._enumeration_method = self.service_list + @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: # Since we're calling the plugin, make sure we have the plugin's requirements return [ requirements.PluginRequirement( - name="svcscan", plugin=svcscan.SvcScan, version=(2, 0, 0) + name="svcscan", plugin=svcscan.SvcScan, version=(3, 0, 0) + ), + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], ), ] - def _get_exe_range(self, proc): + @classmethod + def _get_exe_range(cls, proc) -> Optional[Tuple[int, int]]: """ Returns a tuple of starting,ending address for the VAD containing services.exe @@ -45,21 +55,27 @@ class SvcList(svcscan.SvcScan): return None - def service_list(self, service_table_name, service_binary_dll_map, filter_func): - kernel = self.context.modules[self.config["kernel"]] - + @classmethod + def service_list( + cls, + context: interfaces.context.ContextInterface, + kernel, + service_table_name: str, + service_binary_dll_map, + filter_func, + ): if not symbols.symbol_table_is_64bit( - self.context, kernel.symbol_table_name + context, kernel.symbol_table_name ) or not versions.is_win10_15063_or_later( - context=self.context, symbol_table=kernel.symbol_table_name + context=context, symbol_table=kernel.symbol_table_name ): - vollog.info( + vollog.warning( "This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples" ) return for proc in pslist.PsList.list_processes( - context=self.context, + context=context, layer_name=kernel.layer_name, symbol_table=kernel.symbol_table_name, filter_func=filter_func, @@ -74,9 +90,9 @@ class SvcList(svcscan.SvcScan): ) continue - layer = self.context.layers[layer_name] + layer = context.layers[layer_name] - exe_range = self._get_exe_range(proc) + exe_range = cls._get_exe_range(proc) if not exe_range: vollog.warning( "Could not find the application executable VAD for services.exe. Unable to proceed." @@ -84,19 +100,15 @@ class SvcList(svcscan.SvcScan): continue for offset in layer.scan( - context=self.context, + context=context, scanner=scanners.BytesScanner(needle=b"Sc27"), sections=exe_range, ): - for record in self.enumerate_vista_or_later_header( - service_table_name, service_binary_dll_map, layer_name, offset + for record in cls.enumerate_vista_or_later_header( + context, + service_table_name, + service_binary_dll_map, + layer_name, + offset, ): yield record - - def _generator(self): - service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info() - - for record in self.service_list( - service_table_name, service_binary_dll_map, filter_func - ): - yield (0, record) diff --git a/volatility3/framework/plugins/windows/svcscan.py b/volatility3/framework/plugins/windows/svcscan.py index 07274f03d..4368b83ce 100644 --- a/volatility3/framework/plugins/windows/svcscan.py +++ b/volatility3/framework/plugins/windows/svcscan.py @@ -39,7 +39,11 @@ class SvcScan(interfaces.plugins.PluginInterface): """Scans for windows services.""" _required_framework_version = (2, 0, 0) - _version = (2, 0, 0) + _version = (3, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._enumeration_method = self.service_scan @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -232,13 +236,14 @@ class SvcScan(interfaces.plugins.PluginInterface): for service_key in services } + @classmethod def enumerate_vista_or_later_header( - self, service_table_name, service_binary_dll_map, proc_layer_name, offset + cls, context, service_table_name, service_binary_dll_map, proc_layer_name, offset ): if offset % 8: return - service_header = self.context.object( + service_header =context.object( service_table_name + constants.BANG + "_SERVICE_HEADER", offset=offset, layer_name=proc_layer_name, @@ -257,17 +262,16 @@ class SvcScan(interfaces.plugins.PluginInterface): renderers.UnreadableValue(), renderers.UnreadableValue() ), ) - yield self.get_record_tuple(service_record, service_info) + yield cls.get_record_tuple(service_record, service_info) - def service_scan(self, service_table_name, service_binary_dll_map, filter_func): - kernel = self.context.modules[self.config["kernel"]] - - relative_tag_offset = self.context.symbol_space.get_type( + @classmethod + def service_scan(cls, context: interfaces.context.ContextInterface, kernel, service_table_name: str, service_binary_dll_map, filter_func): + relative_tag_offset = context.symbol_space.get_type( service_table_name + constants.BANG + "_SERVICE_RECORD" ).relative_child_offset("Tag") is_vista_or_later = versions.is_vista_or_later( - context=self.context, symbol_table=kernel.symbol_table_name + context=context, symbol_table=kernel.symbol_table_name ) if is_vista_or_later: @@ -278,7 +282,7 @@ class SvcScan(interfaces.plugins.PluginInterface): seen = [] for task in pslist.PsList.list_processes( - context=self.context, + context=context, layer_name=kernel.layer_name, symbol_table=kernel.symbol_table_name, filter_func=filter_func, @@ -295,15 +299,15 @@ class SvcScan(interfaces.plugins.PluginInterface): ) continue - layer = self.context.layers[proc_layer_name] + layer = context.layers[proc_layer_name] for offset in layer.scan( - context=self.context, + context=context, scanner=scanners.BytesScanner(needle=service_tag), sections=vadyarascan.VadYaraScan.get_vad_maps(task), ): if not is_vista_or_later: - service_record = self.context.object( + service_record = context.object( service_table_name + constants.BANG + "_SERVICE_RECORD", offset=offset - relative_tag_offset, layer_name=proc_layer_name, @@ -318,9 +322,10 @@ class SvcScan(interfaces.plugins.PluginInterface): renderers.UnreadableValue(), renderers.UnreadableValue() ), ) - yield self.get_record_tuple(service_record, service_info) + yield cls.get_record_tuple(service_record, service_info) else: - for service_record in self.enumerate_vista_or_later_header( + for service_record in cls.enumerate_vista_or_later_header( + context, service_table_name, service_binary_dll_map, proc_layer_name, @@ -351,13 +356,13 @@ class SvcScan(interfaces.plugins.PluginInterface): filter_func = pslist.PsList.create_name_filter(["services.exe"]) - return service_table_name, service_binary_dll_map, filter_func + return kernel, service_table_name, service_binary_dll_map, filter_func def _generator(self): - service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info() + kernel, service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info() - for record in self.service_scan( - service_table_name, service_binary_dll_map, filter_func + for record in self._enumeration_method( + self.context, kernel, service_table_name, service_binary_dll_map, filter_func ): yield (0, record)