From c4e7e50180a45dfb80b453b61e9b94860d5a01d3 Mon Sep 17 00:00:00 2001 From: atcuno Date: Wed, 19 Jun 2024 14:42:23 -0500 Subject: [PATCH] Add svclist and svcdiff plugins. Make svcscan more modular to support inheritance and cleaner code --- .../framework/plugins/windows/svcdiff.py | 74 ++++++++++++ .../framework/plugins/windows/svclist.py | 86 ++++++++++++++ .../framework/plugins/windows/svcscan.py | 111 +++++++++++------- .../framework/symbols/windows/versions.py | 9 ++ 4 files changed, 235 insertions(+), 45 deletions(-) create mode 100644 volatility3/framework/plugins/windows/svcdiff.py create mode 100644 volatility3/framework/plugins/windows/svclist.py diff --git a/volatility3/framework/plugins/windows/svcdiff.py b/volatility3/framework/plugins/windows/svcdiff.py new file mode 100644 index 000000000..03f0afcd3 --- /dev/null +++ b/volatility3/framework/plugins/windows/svcdiff.py @@ -0,0 +1,74 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +# This module attempts to locate skeleton-key like function hooks. +# It does this by locating the CSystems array through a variety of methods, +# and then validating the entry for RC4 HMAC (0x17 / 23) +# +# For a thorough walkthrough on how the R&D was performed to develop this plugin, +# please see our blogpost here: +# +# https://volatility-labs.blogspot.com/2021/10/memory-forensics-r-illustrated.html + +import logging + +from volatility3.framework import symbols +from volatility3.framework.configuration import requirements +from volatility3.plugins.windows import svclist, svcscan +from volatility3.framework.symbols.windows import versions + +vollog = logging.getLogger(__name__) + +class SvcDiff(svclist.SvcList, svcscan.SvcScan): + """Compares services found through list walking versus scanning to find rootkits""" + + _required_framework_version = (2, 4, 0) + + @classmethod + def get_requirements(cls): + # Since we're calling the plugin, make sure we have the plugin's requirements + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="svclist", component=svclist.SvcList, version=(1, 0, 0) + ), + requirements.VersionRequirement( + name="svcscan", component=svcscan.SvcScan, version=(2, 0, 0) + ), + ] + + def _generator(self): + """ + Finds services by walking the services.exe list on supported Windows 10 versions + """ + kernel = self.context.modules[self.config["kernel"]] + + 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("This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples") + return + + from_scan = set() + 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): + 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): + from_list.add(service[6]) + + # report services found from scanning but not list walking + for hidden_service in from_scan-from_list: + yield (0, records[hidden_service]) + diff --git a/volatility3/framework/plugins/windows/svclist.py b/volatility3/framework/plugins/windows/svclist.py new file mode 100644 index 000000000..b4541981d --- /dev/null +++ b/volatility3/framework/plugins/windows/svclist.py @@ -0,0 +1,86 @@ +# 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 List + +from volatility3.framework import interfaces, exceptions, symbols +from volatility3.framework.configuration import requirements +from volatility3.framework.symbols.windows import versions +from volatility3.plugins.windows import svcscan, pslist +from volatility3.framework.layers import scanners + +vollog = logging.getLogger(__name__) + + +class SvcList(svcscan.SvcScan): + """Lists services contained with the services.exe doubly linked list of services""" + + _version = (1, 0, 0) + + @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) + ), + ] + + def _get_exe_range(self, proc): + """ + Returns a tuple of starting,ending address for + the VAD containing services.exe + """ + + vad_root = proc.get_vad_root() + for vad in vad_root.traverse(): + filename = vad.get_file_name() + if isinstance(filename, str) and filename.lower().endswith("\\services.exe"): + return [(vad.get_start(), vad.get_size())] + + return None + + def service_list(self, service_table_name, service_binary_dll_map, filter_func): + kernel = self.context.modules[self.config["kernel"]] + + 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("This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples") + return + + for proc in pslist.PsList.list_processes( + context=self.context, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, + filter_func=filter_func, + ): + try: + layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException: + vollog.warning("Unable to access memory of services.exe running with PID: {}".format(proc.UniqueProcessId)) + continue + + layer = self.context.layers[layer_name] + + exe_range = self._get_exe_range(proc) + if not exe_range: + vollog.warning("Could not find the application executable VAD for services.exe. Unable to proceed.") + continue + + for offset in layer.scan( + context=self.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): + 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 10de46e2a..08d11a946 100644 --- a/volatility3/framework/plugins/windows/svcscan.py +++ b/volatility3/framework/plugins/windows/svcscan.py @@ -19,7 +19,7 @@ from volatility3.framework.layers import scanners 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 +from volatility3.framework.symbols.windows.extensions import services as services_types from volatility3.plugins.windows import poolscanner, pslist, vadyarascan from volatility3.plugins.windows.registry import hivelist @@ -140,7 +140,7 @@ class SvcScan(interfaces.plugins.PluginInterface): config_path, os.path.join("windows", "services"), symbol_filename, - class_types=services.class_types, + class_types=services_types.class_types, native_types=native_types, ) @@ -232,28 +232,44 @@ class SvcScan(interfaces.plugins.PluginInterface): for service_key in services } - def _generator(self): + def enumerate_vista_or_later_header( + self, + service_table_name, + service_binary_dll_map, + proc_layer_name, + offset + ): + if offset % 8: + return + + service_header = self.context.object( + service_table_name + constants.BANG + "_SERVICE_HEADER", + offset=offset, + layer_name=proc_layer_name, + ) + + if not service_header.is_valid(): + return + + # since we walk the s-list backwards, if we've seen + # an object, then we've also seen all objects that + # exist before it, thus we can break at that time. + for service_record in service_header.ServiceRecord.traverse(): + service_info = service_binary_dll_map.get( + service_record.get_name(), + ServiceBinaryInfo( + renderers.UnreadableValue(), renderers.UnreadableValue() + ), + ) + yield self.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"]] - service_table_name = self.create_service_table( - self.context, kernel.symbol_table_name, self.config_path - ) - - # Building the dictionary ahead of time is much better for performance - # vs looking up each service's DLL individually. - services_key = self._get_service_key(kernel) - service_binary_dll_map = ( - self._get_service_binary_map(services_key) - if services_key is not None - else {} - ) - relative_tag_offset = self.context.symbol_space.get_type( service_table_name + constants.BANG + "_SERVICE_RECORD" ).relative_child_offset("Tag") - filter_func = pslist.PsList.create_name_filter(["services.exe"]) - is_vista_or_later = versions.is_vista_or_later( context=self.context, symbol_table=kernel.symbol_table_name ) @@ -306,37 +322,42 @@ class SvcScan(interfaces.plugins.PluginInterface): renderers.UnreadableValue(), renderers.UnreadableValue() ), ) - yield ( - 0, - self.get_record_tuple(service_record, service_info), - ) + yield self.get_record_tuple(service_record, service_info) else: - service_header = self.context.object( - service_table_name + constants.BANG + "_SERVICE_HEADER", - offset=offset, - layer_name=proc_layer_name, - ) - - if not service_header.is_valid(): - continue - - # since we walk the s-list backwards, if we've seen - # an object, then we've also seen all objects that - # exist before it, thus we can break at that time. - for service_record in service_header.ServiceRecord.traverse(): + for service_record in self.enumerate_vista_or_later_header(service_table_name, service_binary_dll_map, proc_layer_name, offset): if service_record in seen: break seen.append(service_record) - service_info = service_binary_dll_map.get( - service_record.get_name(), - ServiceBinaryInfo( - renderers.UnreadableValue(), renderers.UnreadableValue() - ), - ) - yield ( - 0, - self.get_record_tuple(service_record, service_info), - ) + yield service_record + + + def get_prereq_info(self): + """ + Data structures and information needed to analyze service information + """ + kernel = self.context.modules[self.config["kernel"]] + + service_table_name = self.create_service_table( + self.context, kernel.symbol_table_name, self.config_path + ) + + services_key = self._get_service_key(kernel) + + service_binary_dll_map = ( + self._get_service_binary_map(services_key) + if services_key is not None + else {} + ) + + filter_func = pslist.PsList.create_name_filter(["services.exe"]) + + return 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() + + for record in self.service_scan(service_table_name, service_binary_dll_map, filter_func): + yield (0, record) def run(self): return renderers.TreeGrid( diff --git a/volatility3/framework/symbols/windows/versions.py b/volatility3/framework/symbols/windows/versions.py index e1e74afc0..d8964f575 100644 --- a/volatility3/framework/symbols/windows/versions.py +++ b/volatility3/framework/symbols/windows/versions.py @@ -141,6 +141,15 @@ is_win10_15063 = OsDistinguisher( ], ) +is_win10_15063_or_later = OsDistinguisher( + version_check=lambda x: x >= (10, 0, 15063), + fallback_checks=[ + ("ObHeaderCookie", None, True), + ("_HANDLE_TABLE", "HandleCount", False), + ("_EPROCESS", "KeepAliveCounter", False), + ], +) + is_win10_16299_or_later = OsDistinguisher( version_check=lambda x: x >= (10, 0, 16299), fallback_checks=[