diff --git a/volatility3/framework/plugins/windows/deskscan.py b/volatility3/framework/plugins/windows/deskscan.py new file mode 100644 index 000000000..baba3d8a0 --- /dev/null +++ b/volatility3/framework/plugins/windows/deskscan.py @@ -0,0 +1,81 @@ +# This file is Copyright 2025 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, Iterable, Tuple + +from volatility3.framework import interfaces +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +from volatility3.plugins.windows import desktops, windowstations + +vollog = logging.getLogger(__name__) + + +class DeskScan(desktops.Desktops): + """Scans for the Desktop instances of each Window Station""" + + _required_framework_version = (2, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.implementation = self.scan_desktops + + @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="desktops", plugin=desktops.Desktops, version=(1, 0, 0) + ), + requirements.PluginRequirement( + name="windowstations", + plugin=windowstations.WindowStations, + version=(1, 0, 0), + ), + ] + + @classmethod + def scan_desktops( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + kernel_module_name: str, + ) -> Iterable[Tuple[int, str, int, str, str, int]]: + """ + Yields the information about each desktop and desktop thread needed for analysis + + The tuple yielded includes the: + Virtual address of the desktop + The window station name + The session id + Desktop name + Process name + Process ID (PID) + """ + kernel = context.modules[kernel_module_name] + + for desktop in windowstations.WindowStations.scan_gui_object( + context, config_path, kernel_module_name, b"Desk", "tagDESKTOP" + ): + desktop_name = desktop.get_name(kernel.symbol_table_name) + if not desktop_name: + continue + + winsta = desktop.get_window_station() + if not winsta: + continue + + winsta_name, session_id = winsta.get_info(kernel.symbol_table_name) + if not winsta_name or session_id is None: + continue + + for _thread, process_name, process_pid in desktop.get_threads(): + yield format_hints.Hex( + desktop.vol.offset + ), winsta_name, session_id, desktop_name, process_name, process_pid diff --git a/volatility3/framework/plugins/windows/desktops.py b/volatility3/framework/plugins/windows/desktops.py new file mode 100644 index 000000000..1085ff36d --- /dev/null +++ b/volatility3/framework/plugins/windows/desktops.py @@ -0,0 +1,91 @@ +# This file is Copyright 2025 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, Iterable + +from volatility3.framework import interfaces, renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +from volatility3.plugins.windows import windowstations + +vollog = logging.getLogger(__name__) + + +class Desktops(interfaces.plugins.PluginInterface): + """Enumerates the Desktop instances of each Window Station""" + + _required_framework_version = (2, 0, 0) + _version = (1, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.implementation = self.list_desktops + + @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="windowstations", + plugin=windowstations.WindowStations, + version=(1, 0, 0), + ), + ] + + @classmethod + def list_desktops( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + kernel_module_name: str, + ) -> Iterable[interfaces.objects.ObjectInterface]: + """ + Uses `scan_window_stations` to find each window station + For each found, enumerates its desktops followed by the + threads of each desktop. + """ + kernel = context.modules[kernel_module_name] + + for ( + winsta, + station_name, + session_id, + ) in windowstations.WindowStations.scan_window_stations( + context, config_path, kernel_module_name + ): + # for each window station, walk its list of desktops + for desktop, desktop_name in winsta.desktops(kernel.symbol_table_name): + # for each desktop, walk its threads + for _thread, process_name, process_pid in desktop.get_threads(): + yield format_hints.Hex( + desktop.vol.offset + ), station_name, session_id, desktop_name, process_name, process_pid + + def _generator(self): + kernel_name = self.config["kernel"] + + # call the implementation for finding desktops + # yield the information, which will include the owning window station and process + for desktop_info in self.implementation( + self.context, self.config_path, kernel_name + ): + yield 0, desktop_info + + def run(self): + return renderers.TreeGrid( + [ + ("Offset", format_hints.Hex), + ("Window Station", str), + ("Session", int), + ("Desktop", str), + ("Process", str), + ("PID", int), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/windows/gui.py b/volatility3/framework/plugins/windows/gui.py deleted file mode 100644 index 1cce5b7f3..000000000 --- a/volatility3/framework/plugins/windows/gui.py +++ /dev/null @@ -1,114 +0,0 @@ -# This file is Copyright 2020 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 -import os -from itertools import count -from typing import List, Tuple - -from volatility3.framework import interfaces, renderers, symbols -from volatility3.framework.configuration import requirements -from volatility3.framework.renderers import format_hints -from volatility3.framework.symbols import intermed -from volatility3.framework.symbols.windows import versions - -# from volatility3.plugins.windows import pslist, vadinfo, modules - -vollog = logging.getLogger(__name__) - - -class WinGUI(interfaces.plugins.PluginInterface): - """Parses information about Windows GUI Objects""" - - _required_framework_version = (2, 0, 0) - - # These checks must be completed from newest -> oldest OS version. - _win_version_file_map: List[Tuple[versions.OsDistinguisher, str]] = [ - (versions.is_win10_19577_or_later, "gui-win10-19577-x64"), - (versions.is_win10_19041_or_later, "gui-win10-19041-x64"), - (versions.is_win10_18362_or_later, "gui-win10-18362-x64"), - (versions.is_win10_17763_or_later, "gui-win10-17763-x64"), - (versions.is_win10_17134_or_later, "gui-win10-17134-x64"), - (versions.is_win10_16299_or_later, "gui-win10-16299-x64"), - (versions.is_win10_15063_or_later, "gui-win10-15063-x64"), - (versions.is_win10_10586_or_later, "gui-win10-10586-x64"), - (versions.is_windows_8_or_later, "gui-win8-x64"), - (versions.is_windows_7_sp1, "gui-win7sp1-x64"), - (versions.is_windows_7_sp0, "gui-win7sp0-x64"), - ] - - @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"], - ), - ] - - @staticmethod - def create_gui_table( - context: interfaces.context.ContextInterface, - symbol_table: str, - config_path: str, - ) -> str: - """Creates a symbol table for windows GUI types - - Args: - context: The context to retrieve required elements (layers, symbol tables) from - symbol_table: The name of an existing symbol table containing the kernel symbols - config_path: The configuration path within the context of the symbol table to create - - Returns: - The name of the constructed GUI table - """ - native_types = context.symbol_space[symbol_table].natives - - if not symbols.symbol_table_is_64bit(context, symbol_table): - raise NotImplementedError( - "This plugin only supports x64 versions of Windows" - ) - - table_mapping = {"nt_symbols": symbol_table} - - try: - symbol_filename = next( - filename - for version_check, filename in WinGUI._win_version_file_map - if version_check(context=context, symbol_table=symbol_table) - ) - except StopIteration: - raise NotImplementedError("This version of Windows is not supported!") - - vollog.debug(f"Using GUI table {symbol_filename}") - - return intermed.IntermediateSymbolTable.create( - context, - config_path, - os.path.join("windows", "gui"), - symbol_filename, - native_types=native_types, - table_mapping=table_mapping, - ) - - def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - - gui_table = self.create_gui_table( - self.context, kernel.symbol_table_name, self.config_path - ) - - c = count() - for _ in range(10): - yield ( - 0, - (next(c), tuple()), - ) - - def run(self): - return renderers.TreeGrid( - [], - self._generator(), - ) diff --git a/volatility3/framework/plugins/windows/poolscanner.py b/volatility3/framework/plugins/windows/poolscanner.py index 7446768c7..75de8bb95 100644 --- a/volatility3/framework/plugins/windows/poolscanner.py +++ b/volatility3/framework/plugins/windows/poolscanner.py @@ -182,6 +182,36 @@ class PoolScanner(plugins.PluginInterface): ), ) + @staticmethod + def gui_poolscanner_constraints( + gui_table: str, tags_filter: Optional[List[bytes]] = None + ) -> List[PoolConstraint]: + """ + Constraints for objects managed by the GUI subsystem (win32k*.sys) + """ + builtins = [ + PoolConstraint( + b"Wind", + type_name=gui_table + constants.BANG + "tagWINDOWSTATION", + size=(0x90, None), + page_type=PoolType.PAGED, + object_type="WindowStation", + skip_type_test=True, + ), + PoolConstraint( + b"Desk", + type_name=gui_table + constants.BANG + "tagDESKTOP", + page_type=PoolType.PAGED, + object_type="Desktop", + skip_type_test=True, + ), + ] + + if not tags_filter: + return builtins + + return [constraint for constraint in builtins if constraint.tag in tags_filter] + @classmethod def builtin_constraints( cls, symbol_table: str, tags_filter: Optional[List[bytes]] = None diff --git a/volatility3/framework/plugins/windows/windowstations.py b/volatility3/framework/plugins/windows/windowstations.py new file mode 100644 index 000000000..cb6a52b88 --- /dev/null +++ b/volatility3/framework/plugins/windows/windowstations.py @@ -0,0 +1,234 @@ +# This file is Copyright 2025 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 +import os +from typing import List, Tuple, Iterator, Generator, Dict + +from volatility3.framework import interfaces, renderers, symbols, exceptions +from volatility3.framework.configuration import requirements +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 gui +from volatility3.plugins.windows import poolscanner, modules + +vollog = logging.getLogger(__name__) + + +class WindowStations(interfaces.plugins.PluginInterface): + """Scans for top level Windows Stations""" + + _required_framework_version = (2, 0, 0) + _version = (1, 0, 0) + + # These checks must be completed from newest -> oldest OS version. + _win_version_file_map: List[Tuple[versions.OsDistinguisher, str]] = [ + (versions.is_win10_19577_or_later, "gui-win10-19577-x64"), + (versions.is_win10_19041_or_later, "gui-win10-19041-x64"), + (versions.is_win10_18362_or_later, "gui-win10-18362-x64"), + (versions.is_win10_17763_or_later, "gui-win10-17763-x64"), + (versions.is_win10_17134_or_later, "gui-win10-17134-x64"), + (versions.is_win10_16299_or_later, "gui-win10-16299-x64"), + (versions.is_win10_15063_or_later, "gui-win10-15063-x64"), + (versions.is_win10_10586_or_later, "gui-win10-10586-x64"), + (versions.is_windows_8_or_later, "gui-win8-x64"), + (versions.is_windows_7_sp1, "gui-win7sp1-x64"), + (versions.is_windows_7_sp0, "gui-win7sp0-x64"), + ] + + @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"], + ), + ] + + @staticmethod + def create_gui_table( + context: interfaces.context.ContextInterface, + symbol_table: str, + config_path: str, + ) -> str: + """Creates a symbol table for windows GUI types + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + symbol_table: The name of an existing symbol table containing the kernel symbols + config_path: The configuration path within the context of the symbol table to create + + Returns: + The name of the constructed GUI table + """ + native_types = context.symbol_space[symbol_table].natives + + if not symbols.symbol_table_is_64bit(context, symbol_table): + raise NotImplementedError( + "This plugin only supports x64 versions of Windows" + ) + + table_mapping = {"nt_symbols": symbol_table} + + try: + symbol_filename = next( + filename + for version_check, filename in WindowStations._win_version_file_map + if version_check(context=context, symbol_table=symbol_table) + ) + except StopIteration: + raise NotImplementedError("This version of Windows is not supported!") + + vollog.debug(f"Using GUI table {symbol_filename}") + + return intermed.IntermediateSymbolTable.create( + context, + config_path, + os.path.join("windows", "gui"), + symbol_filename, + class_types=gui.class_types, + native_types=native_types, + table_mapping=table_mapping, + ) + + @classmethod + def get_session_map( + cls, + context: interfaces.context.ContextInterface, + module_name: str, + gui_table_name: str, + ) -> Dict[int, interfaces.context.ModuleInterface]: + """ + Walks each session layer and returns a dictionary that + maps session identifiers to a module in the session's layer + """ + session_map = modules.Modules.get_session_layers_map(context, module_name) + + for session_id, session_layer in session_map.items(): + session_module = context.module( + gui_table_name, layer_name=session_layer, offset=0 + ) + session_map[session_id] = session_module + + return session_map + + @classmethod + def scan_gui_object( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + kernel_module_name: str, + object_tag: bytes, + object_type: str, + ) -> Generator[interfaces.objects.ObjectInterface, None, None]: + """ + An API that generically scans for GUI (win32*.sys) objects allocated in the pools (which is nearly all of them) + + This function scans within the kernel space for the tags and then uses `get_session_map` to instantiate objects + in their correct session address space. + + Args: + context: + config_path: + kernel_module_name: + object_tag: The 4 byte pool header tag to search for + object_type: The data structure of the GUI object within the pool + """ + + kernel = context.modules[kernel_module_name] + + gui_table_name = cls.create_gui_table( + context, kernel.symbol_table_name, config_path + ) + + constraints = poolscanner.PoolScanner.gui_poolscanner_constraints( + gui_table_name, [object_tag] + ) + + session_map = cls.get_session_map(context, kernel_module_name, gui_table_name) + + for result in poolscanner.PoolScanner.generate_pool_scan_extended( + context, + kernel.layer_name, + kernel.symbol_table_name, + gui_table_name, + constraints, + ): + _constraint, mem_object, _header = result + + # enforce that objects are in a valid session + # this prevents smear and also ensures future pointer + # dereferences are performed in the correct address space (layer) + try: + session_id = mem_object.get_session_id() + except exceptions.InvalidAddressException: + continue + + if session_id is not None: + session_module = session_map.get(session_id, None) + if session_module: + # create the object its own address space (per-session) + yield session_module.object( + object_type=object_type, offset=mem_object.vol.offset + ) + + @classmethod + def scan_window_stations( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + kernel_module_name: str, + ) -> Iterator[Tuple["gui.tagWINDOWSTATION", str, int]]: + """ + Scans for window stations through `scan_gui_object` + Yields each window station along with its name and session_id + """ + + seen = set() + + kernel = context.modules[kernel_module_name] + + for scanned_winsta in cls.scan_gui_object( + context, config_path, kernel_module_name, b"Wind", "tagWINDOWSTATION" + ): + # walk the list of each station found through scanning + for winsta in scanned_winsta.traverse(): + if winsta.vol.offset in seen: + continue + seen.add(winsta.vol.offset) + + # stations need to have a name and be in a session + name, session_id = winsta.get_info(kernel.symbol_table_name) + if name and session_id is not None: + yield winsta, name, session_id + + def _generator(self): + """ + A wrapper around `scan_window_stations` + """ + for winsta, name, session_id in self.scan_window_stations( + self.context, self.config_path, self.config["kernel"] + ): + yield ( + 0, + ( + format_hints.Hex(winsta.vol.offset), + name, + session_id, + ), + ) + + # Volatility 2 reported whether the station is interactive or not, but I could not determine if its algorithm + # is currently valid. I also did not see where the old code paths still checked the same bit mask + def run(self): + return renderers.TreeGrid( + [ + ("Offset", format_hints.Hex), + ("Name", str), + ("SessionId", int), + ], + self._generator(), + ) diff --git a/volatility3/framework/symbols/windows/extensions/gui.py b/volatility3/framework/symbols/windows/extensions/gui.py new file mode 100644 index 000000000..92693dba5 --- /dev/null +++ b/volatility3/framework/symbols/windows/extensions/gui.py @@ -0,0 +1,127 @@ +# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +from typing import Optional, Tuple, Iterator + +from volatility3.framework import exceptions, constants, interfaces +from volatility3.framework import objects +from volatility3.framework.objects import utility +from volatility3.framework.symbols.windows.extensions import pool + + +class tagWINDOWSTATION(objects.StructType, pool.ExecutiveObject): + def is_valid(self) -> bool: + sid = self.get_session_id() + return sid is not None and 0 <= sid < 256 + + def get_session_id(self) -> Optional[int]: + try: + return self.dwSessionId + except exceptions.InvalidAddressException: + return None + + def traverse(self, max_stations: int = 15): + """ + Traverses the window stations referenced in the list of stations + """ + seen = set() + + # include the first window station + yield self + + while len(seen) < max_stations: + try: + winsta = self.rpwinstaNext.dereference() + except exceptions.InvalidAddressException: + break + + if winsta.vol.offset in seen: + break + + yield winsta + + seen.add(winsta.vol.offset) + + def get_info(self, kernel_symbol_table_name) -> Optional[Tuple[str, int]]: + try: + name = self.get_name(kernel_symbol_table_name) + session_id = self.get_session_id() + except exceptions.InvalidAddressException: + return None, None + + # attempt to avoid smear + if session_id is not None and session_id < 256 and name and len(name) > 1: + return name, session_id + + return None, None + + def desktops(self, symbol_table_name, max_desktops: int = 12): + seen = set() + + while len(seen) < max_desktops: + try: + desktop = self.rpdeskList.dereference() + name = desktop.get_name(symbol_table_name) + except exceptions.InvalidAddressException: + break + + if desktop.vol.offset in seen: + break + + yield desktop, name + + seen.add(desktop.vol.offset) + + +class tagDESKTOP(objects.StructType, pool.ExecutiveObject): + def is_valid(self) -> bool: + """ + Enforce a valid sid + owning window station + """ + sid = self.get_session_id() + + valid_sid = sid is not None and 0 <= sid < 256 + + if valid_sid: + return self.get_window_station() is not None + + return False + + def get_window_station(self) -> Optional["tagWINDOWSTATION"]: + try: + return self.rpwinstaParent.dereference() + except exceptions.InvalidAddressException: + return None + + def get_session_id(self) -> Optional[int]: + winsta = self.get_window_station() + if winsta: + return winsta.get_session_id() + + return None + + def get_threads( + self, + ) -> Iterator[Tuple[interfaces.objects.ObjectInterface, str, int]]: + """ + Returns the threads of each desktop along with owning process information + """ + symbol_table_name = self.vol.type_name.split(constants.BANG)[0] + + for thread in self.PtiList.to_list( + symbol_table_name + constants.BANG + "tagTHREADINFO", "PtiLink" + ): + try: + process_name = utility.array_to_string(thread.ppi.Process.ImageFileName) + process_pid = thread.ppi.Process.UniqueProcessId + except exceptions.InvalidAddressException: + continue + + yield thread, process_name, process_pid + + +class_types = { + "tagWINDOWSTATION": tagWINDOWSTATION, + "tagDESKTOP": tagDESKTOP, +}