diff --git a/volatility3/framework/plugins/windows/orphan_kernel_threads.py b/volatility3/framework/plugins/windows/orphan_kernel_threads.py new file mode 100644 index 000000000..f4901dc8c --- /dev/null +++ b/volatility3/framework/plugins/windows/orphan_kernel_threads.py @@ -0,0 +1,96 @@ +# 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 +# + +import logging +from typing import List, Generator + +from volatility3.framework import interfaces, symbols +from volatility3.framework.configuration import requirements +from volatility3.plugins.windows import thrdscan, ssdt + +vollog = logging.getLogger(__name__) + + +class Threads(thrdscan.ThrdScan): + """Lists process threads""" + + _required_framework_version = (2, 4, 0) + _version = (1, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.implementation = self.list_orphan_kernel_threads + + @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.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.PluginRequirement( + name="thrdscan", plugin=thrdscan.ThrdScan, version=(1, 1, 0) + ), + requirements.PluginRequirement( + name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0) + ), + ] + + @classmethod + def list_orphan_kernel_threads( + cls, + context: interfaces.context.ContextInterface, + module_name: str, + ) -> Generator[interfaces.objects.ObjectInterface, None, None]: + """Yields thread objects of kernel threads that do not map to a module + + Args: + cls + context: the context to operate upon + module_name: name of the module to use for scanning + Returns: + A generator of thread objects of orphaned threads + """ + module = context.modules[module_name] + layer_name = module.layer_name + symbol_table = module.symbol_table_name + + collection = ssdt.SSDT.build_module_collection( + context, layer_name, symbol_table + ) + + # FIXME - use a proper constant once established + # used to filter out smeared pointers + if symbols.symbol_table_is_64bit(context, symbol_table): + kernel_start = 0xFFFFF80000000000 + else: + kernel_start = 0x80000000 + + for thread in thrdscan.ThrdScan.scan_threads(context, module_name): + # we don't want smeared or terminated threads + try: + proc = thread.owning_process() + except AttributeError: + continue + + # we only care about kernel threads, 4 = System + # previous methods for determining if a thread was a kernel thread + # such as bit fields and flags are not stable in Win10+ + # so we check if the thread is from the kernel itself or one its child + # kernel processes (MemCompression, Regsitry, ...) + if proc.UniqueProcessId != 4 and proc.InheritedFromUniqueProcessId != 4: + continue + + if thread.StartAddress < kernel_start: + continue + + module_symbols = list( + collection.get_module_symbols_by_absolute_location(thread.StartAddress) + ) + + # alert on threads that do not map to a module + if not module_symbols: + yield thread diff --git a/volatility3/framework/plugins/windows/thrdscan.py b/volatility3/framework/plugins/windows/thrdscan.py index 6e664e052..b812a15ff 100644 --- a/volatility3/framework/plugins/windows/thrdscan.py +++ b/volatility3/framework/plugins/windows/thrdscan.py @@ -22,8 +22,8 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) _version = (1, 1, 0) def __init__(self, *args, **kwargs): - self.implementation = self.scan_threads super().__init__(*args, **kwargs) + self.implementation = self.scan_threads @classmethod def get_requirements(cls): @@ -48,8 +48,7 @@ class ThrdScan(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 + module_name: Name of the module to use for scanning Returns: A list of _ETHREAD objects found by scanning memory for the "Thre" / "Thr\\xE5" pool signatures diff --git a/volatility3/framework/plugins/windows/threads.py b/volatility3/framework/plugins/windows/threads.py index 39f3b7d77..98a3169a5 100644 --- a/volatility3/framework/plugins/windows/threads.py +++ b/volatility3/framework/plugins/windows/threads.py @@ -19,8 +19,8 @@ class Threads(thrdscan.ThrdScan): _version = (1, 0, 0) def __init__(self, *args, **kwargs): - self.implementation = self.list_process_threads super().__init__(*args, **kwargs) + self.implementation = self.list_process_threads @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -50,7 +50,6 @@ class Threads(thrdscan.ThrdScan): Args: proc: _EPROCESS object from which to list the VADs - filter_func: Function to take a virtual address descriptor value and return True if it should be filtered out Returns: A list of threads based on the process and filtered based on the filter function @@ -64,22 +63,19 @@ class Threads(thrdscan.ThrdScan): seen.add(thread.vol.offset) yield thread - @classmethod - def filter_func(cls, config: interfaces.configuration.HierarchicalDict) -> Callable: - return pslist.PsList.create_pid_filter(config.get("pid", None)) - @classmethod def list_process_threads( cls, context: interfaces.context.ContextInterface, module_name: str, - filter_func: Callable, ) -> Iterable[interfaces.objects.ObjectInterface]: """Runs through all processes and lists threads for each process""" module = context.modules[module_name] layer_name = module.layer_name symbol_table_name = module.symbol_table_name + filter_func = pslist.PsList.create_pid_filter(context.config.get("pid", None)) + for proc in pslist.PsList.list_processes( context=context, layer_name=layer_name,