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.

Author: dgmcdona

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 @@ def scan_threads(
         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 @@ def scan_threads(
     @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 @@ def gather_thread_info(
             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 @@ def _generator(self, filter_func: Callable):
             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):

volatility3/framework/symbols/windows/extensions/__init__.py

@@ -568,16 +568,20 @@ def is_valid(self) -> bool:
         # 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