mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-13 13:17:38 +02:00
Windows GUI Plugins: Adds three plugins
This adds the windowstations, desktops, and deskscan plugins, and removes the gui.py plugin file that was stubbed out in the introductory work for the effort. It also adds a new method to the poolscanner class, but does not bump the poolscanner version number since there is already a major version number bump going into this PR. Co-authored-by: Andrew Case <andrew@dfir.org>
This commit is contained in:
Andrew Case
committed by
David McDonald
David McDonald
parent
885bf187e2
commit
dad5a75aaa
@@ -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
|
||||
@@ -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(),
|
||||
)
|
||||
@@ -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(),
|
||||
)
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
)
|
||||
@@ -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,
|
||||
}
|
||||
Reference in New Issue
Block a user