From 64ecd65d2c8ccb5e9284182104266e97d8fabce4 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Tue, 8 Apr 2025 14:37:29 -0500 Subject: [PATCH] Windows: Improve type-hints in thrdscan, extensions This improves type-hinting in the `ThrdScan` class and in the `ETHREAD` extension class through narrowing the return type of some methods from `interfaces.objects.ObjectInterface` to their actual return type, `extensions.ETHREAD`. Also creates a `NamedTuple` for holding thread info, which cleans up the type signature and makes the returned value easier for consumers to use. --- .../framework/plugins/windows/thrdscan.py | 77 ++++++++----------- .../symbols/windows/extensions/__init__.py | 10 ++- 2 files changed, 41 insertions(+), 46 deletions(-) diff --git a/volatility3/framework/plugins/windows/thrdscan.py b/volatility3/framework/plugins/windows/thrdscan.py index 0ac3d0c33..49103cc7e 100644 --- a/volatility3/framework/plugins/windows/thrdscan.py +++ b/volatility3/framework/plugins/windows/thrdscan.py @@ -1,15 +1,16 @@ ## ## plugin for testing addition of threads scan support to poolscanner.py ## -import logging import datetime -from typing import Callable, Iterable, Tuple, Optional, Dict +import logging +from typing import Callable, Dict, NamedTuple, Optional, Union, Tuple, Iterator -from volatility3.framework import renderers, interfaces, exceptions +from volatility3.framework import exceptions, interfaces, objects, renderers from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints -from volatility3.plugins.windows import poolscanner, pe_symbols +from volatility3.framework.symbols.windows import extensions as win_extensions from volatility3.plugins import timeliner +from volatility3.plugins.windows import pe_symbols, poolscanner vollog = logging.getLogger(__name__) @@ -21,6 +22,17 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) _required_framework_version = (2, 6, 0) _version = (2, 0, 0) + class ThreadInfo(NamedTuple): + offset: int + pid: objects.Pointer + tid: objects.Pointer + start_addr: objects.Pointer + start_path: Optional[str] + win32_start_addr: objects.Pointer + win32_start_path: Optional[str] + create_time: Union[datetime.datetime, interfaces.renderers.BaseAbsentValue] + exit_time: Union[datetime.datetime, interfaces.renderers.BaseAbsentValue] + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.implementation = self.scan_threads @@ -51,7 +63,7 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) cls, context: interfaces.context.ContextInterface, module_name: str, - ) -> Iterable[interfaces.objects.ObjectInterface]: + ) -> Iterator[win_extensions.ETHREAD]: """Scans for threads using the poolscanner module and constraints. Args: @@ -77,19 +89,9 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) @classmethod def gather_thread_info( cls, - ethread: interfaces.objects.ObjectInterface, - vads_cache: Dict[int, pe_symbols.ranges_type] = None, - ) -> Tuple[ - int, - int, - int, - int, - Optional[str], - int, - Optional[str], - Optional[datetime.datetime], - Optional[datetime.datetime], - ]: + ethread: win_extensions.ETHREAD, + vads_cache: Optional[Dict[int, pe_symbols.ranges_type]] = None, + ) -> Optional[ThreadInfo]: try: thread_offset = ethread.vol.offset owner_proc_pid = ethread.Cid.UniqueProcess @@ -135,19 +137,19 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) start_path = None win32start_path = None - return ( - format_hints.Hex(thread_offset), + return cls.ThreadInfo( + thread_offset, owner_proc_pid, thread_tid, - format_hints.Hex(thread_start_addr), + thread_start_addr, start_path, - format_hints.Hex(thread_win32start_addr), + thread_win32start_addr, win32start_path, thread_create_time, thread_exit_time, ) - def _generator(self, filter_func: Callable): + def _generator(self, filter_func: Callable) -> Iterator[Tuple[int, Tuple]]: kernel_name = self.config["kernel"] vads_cache: Dict[int, pe_symbols.ranges_type] = {} @@ -156,27 +158,16 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface) info = self.gather_thread_info(ethread, vads_cache) if info: - ( - offset, - pid, - tid, - start_addr, - start_path, - win32start_addr, - win32start_path, - create_time, - exit_time, - ) = info yield 0, ( - offset, - pid, - tid, - start_addr, - start_path or renderers.NotAvailableValue(), - win32start_addr, - win32start_path or renderers.NotAvailableValue(), - create_time, - exit_time, + format_hints.Hex(info.offset), + info.pid, + info.tid, + format_hints.Hex(info.start_addr), + info.start_path or renderers.NotAvailableValue(), + format_hints.Hex(info.win32_start_addr), + info.win32_start_path or renderers.NotAvailableValue(), + info.create_time, + info.exit_time, ) def generate_timeline(self): diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 9fe250ba5..1972d6a50 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -568,16 +568,20 @@ class ETHREAD(objects.StructType, pool.ExecutiveObject): # passed all validations return True - def get_create_time(self): + def get_create_time( + self, + ) -> Union[datetime.datetime, interfaces.renderers.BaseAbsentValue]: # For Windows XPs if self.has_member("ThreadsProcess"): return conversion.wintime_to_datetime(self.CreateTime.QuadPart >> 3) return conversion.wintime_to_datetime(self.CreateTime.QuadPart) - def get_exit_time(self): + def get_exit_time( + self, + ) -> Union[datetime.datetime, interfaces.renderers.BaseAbsentValue]: return conversion.wintime_to_datetime(self.ExitTime.QuadPart) - def owning_process(self) -> interfaces.objects.ObjectInterface: + def owning_process(self) -> "EPROCESS": """Return the EPROCESS that owns this thread.""" # For Windows XPs