WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-12 12:47:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Convert poolscanning APIs to new coding standards

Browse Source
This commit is contained in:
Andrew Case
2025-03-08 17:21:27 +00:00
parent 5347bb8168
commit f69b00b48c
22 changed files with 262 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/windows/bigpools.py
+14 -20
View File
@@ -21,7 +21,7 @@ class BigPools(interfaces.plugins.PluginInterface):
"""List big page pools."""
_required_framework_version = (2, 0, 0)
_version = (1, 1, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -50,8 +50,7 @@ class BigPools(interfaces.plugins.PluginInterface):
def list_big_pools(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
tags: Optional[list] = None,
show_free: bool = False,
):
@@ -59,19 +58,13 @@ class BigPools(interfaces.plugins.PluginInterface):
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
kernel_module_name: The name of the module for the kernel
tags: An optional list of pool tags to filter big page pool tags by
Yields:
A big page pool object
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
big_page_table_offset = ntkrnlmp.get_symbol("PoolBigPageTable").address
big_page_table = ntkrnlmp.object(
@@ -87,8 +80,10 @@ class BigPools(interfaces.plugins.PluginInterface):
big_page_table_type = ntkrnlmp.get_type("_POOL_TRACKER_BIG_PAGES")
except exceptions.SymbolError:
# We have to manually load a symbol table
is_vista_or_later = versions.is_vista_or_later(context, symbol_table)
is_win10 = versions.is_win10(context, symbol_table)
is_vista_or_later = versions.is_vista_or_later(
context, ntkrnlmp.symbol_table_name
)
is_win10 = versions.is_win10(context, ntkrnlmp.symbol_table_name)
if is_win10:
big_pools_json_filename = "bigpools-win10"
elif is_vista_or_later:
@@ -96,7 +91,7 @@ class BigPools(interfaces.plugins.PluginInterface):
else:
big_pools_json_filename = "bigpools"
if symbols.symbol_table_is_64bit(context, symbol_table):
if symbols.symbol_table_is_64bit(context, ntkrnlmp.symbol_table_name):
big_pools_json_filename += "-x64"
else:
big_pools_json_filename += "-x86"
@@ -104,16 +99,17 @@ class BigPools(interfaces.plugins.PluginInterface):
new_table_name = intermed.IntermediateSymbolTable.create(
context=context,
config_path=configuration.path_join(
context.symbol_space[symbol_table].config_path, "bigpools"
context.symbol_space[ntkrnlmp.symbol_table_name].config_path,
"bigpools",
),
sub_path=os.path.join("windows", "bigpools"),
filename=big_pools_json_filename,
table_mapping={"nt_symbols": symbol_table},
table_mapping={"nt_symbols": ntkrnlmp.symbol_table_name},
class_types={
"_POOL_TRACKER_BIG_PAGES": extensions.pool.POOL_TRACKER_BIG_PAGES
},
)
module = context.module(new_table_name, layer_name, offset=0)
module = context.module(new_table_name, ntkrnlmp.layer_name, offset=0)
big_page_table_type = module.get_type("_POOL_TRACKER_BIG_PAGES")
big_pools = ntkrnlmp.object(
@@ -136,12 +132,10 @@ class BigPools(interfaces.plugins.PluginInterface):
tags = [tag for tag in self.config["tags"].split(",")]
else:
tags = None
kernel = self.context.modules[self.config["kernel"]]
for big_pool in self.list_big_pools(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
kernel_module_name=self.config["kernel"],
tags=tags,
show_free=self.config.get("show-free"),
):
volatility3/framework/plugins/windows/callbacks.py
+45 -81
View File
@@ -28,7 +28,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
"""Lists kernel callbacks and notification routines."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 1)
_version = (3, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -42,7 +42,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
name="ssdt", plugin=ssdt.SSDT, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="driverirp", plugin=driverirp.DriverIrp, version=(1, 0, 0)
@@ -211,8 +211,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
def scan(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
kernel_module_name: str,
callback_symbol_table: str,
) -> Iterable[
Tuple[
@@ -225,18 +224,21 @@ class Callbacks(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
kernel_module_name: Name of the module for the kernel
callback_symbol_table: The name of the table containing the callback object symbols (_SHUTDOWN_PACKET etc.)
Returns:
A list of callback objects found by scanning the `layer_name` layer for callback pool signatures
"""
kernel = context.modules[kernel_module_name]
is_vista_or_later = versions.is_vista_or_later(
context=context, symbol_table=nt_symbol_table
context=context, symbol_table=kernel.symbol_table_name
)
type_map = handles.Handles.get_type_map(context, layer_name, nt_symbol_table)
type_map = handles.Handles.get_type_map(
context, kernel.layer_name, kernel.symbol_table_name
)
constraints = cls.create_callback_scan_constraints(
context, callback_symbol_table, is_vista_or_later
@@ -247,7 +249,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
mem_object,
_header,
) in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, nt_symbol_table, constraints
context, kernel_module_name, constraints
):
try:
if isinstance(mem_object, callbacks._SHUTDOWN_PACKET):
@@ -347,31 +349,24 @@ class Callbacks(interfaces.plugins.PluginInterface):
def list_notify_routines(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, Optional[str]]]:
"""Lists all kernel notification routines.
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
kernel_module_name: The name of the module of the kernel
callback_table_name: The name of the table containing the callback symbols
Yields:
A name, location and optional detail string
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
is_vista_or_later = versions.is_vista_or_later(
context=context, symbol_table=symbol_table
context=context, symbol_table=ntkrnlmp.symbol_table_name
)
full_type_name = callback_table_name + constants.BANG + "_GENERIC_CALLBACK"
@@ -416,20 +411,14 @@ class Callbacks(interfaces.plugins.PluginInterface):
def _list_registry_callbacks_legacy(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, None]]:
"""
Lists all registry callbacks from the old format via the CmpCallBackVector.
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
full_type_name = (
callback_table_name + constants.BANG + "_EX_CALLBACK_ROUTINE_BLOCK"
)
@@ -467,20 +456,13 @@ class Callbacks(interfaces.plugins.PluginInterface):
def _list_registry_callbacks_new(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, Optional[str]]]:
"""
Lists all registry callbacks via the CallbackListHead.
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
full_type_name = callback_table_name + constants.BANG + "_CM_CALLBACK_ENTRY"
symbol_offset = ntkrnlmp.get_symbol("CallbackListHead").address
@@ -504,40 +486,33 @@ class Callbacks(interfaces.plugins.PluginInterface):
def list_registry_callbacks(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, Optional[str]]]:
"""Lists all registry callbacks.
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
kernel_module_name: The name of the module of the kernel
callback_table_name: The name of the table containing the callback symbols
Yields:
A name, location and optional detail string
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
if ntkrnlmp.has_symbol("CmpCallBackVector") and ntkrnlmp.has_symbol(
"CmpCallBackCount"
):
yield from cls._list_registry_callbacks_legacy(
context, layer_name, symbol_table, callback_table_name
context, kernel_module_name, callback_table_name
)
elif ntkrnlmp.has_symbol("CallbackListHead") and ntkrnlmp.has_symbol(
"CmpCallBackCount"
):
yield from cls._list_registry_callbacks_new(
context, layer_name, symbol_table, callback_table_name
context, kernel_module_name, callback_table_name
)
else:
symbols_to_check = [
@@ -552,14 +527,11 @@ class Callbacks(interfaces.plugins.PluginInterface):
symbol_status = "exists"
vollog.debug(f"symbol {symbol_name} {symbol_status}.")
return None
@classmethod
def list_bugcheck_reason_callbacks(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[
Tuple[
@@ -572,20 +544,14 @@ class Callbacks(interfaces.plugins.PluginInterface):
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
kernel_module_name: The name of the module of the kernel
callback_table_name: The name of the table containing the callback symbols
Yields:
A name, location and optional detail string
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
try:
list_offset = ntkrnlmp.get_symbol(
@@ -599,11 +565,15 @@ class Callbacks(interfaces.plugins.PluginInterface):
callback_table_name + constants.BANG + "_KBUGCHECK_REASON_CALLBACK_RECORD"
)
callback_record = context.object(
object_type=full_type_name, offset=kvo + list_offset, layer_name=layer_name
object_type=full_type_name,
offset=ntkrnlmp.offset + list_offset,
layer_name=ntkrnlmp.layer_name,
)
for callback in callback_record.Entry:
if not context.layers[layer_name].is_valid(callback.CallbackRoutine, 64):
if not context.layers[ntkrnlmp.layer_name].is_valid(
callback.CallbackRoutine, 64
):
continue
try:
@@ -626,8 +596,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
def list_bugcheck_callbacks(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
callback_table_name: str,
) -> Iterable[
Tuple[
@@ -640,20 +609,13 @@ class Callbacks(interfaces.plugins.PluginInterface):
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
kernel_module_name: The name of the module of the kernel
callback_table_name: The name of the table containing the callback symbols
Yields:
A name, location and optional detail string
"""
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.modules[kernel_module_name]
try:
list_offset = ntkrnlmp.get_symbol("KeBugCheckCallbackListHead").address
@@ -665,17 +627,20 @@ class Callbacks(interfaces.plugins.PluginInterface):
callback_table_name + constants.BANG + "_KBUGCHECK_CALLBACK_RECORD"
)
callback_record = context.object(
full_type_name, offset=kvo + list_offset, layer_name=layer_name
full_type_name,
offset=ntkrnlmp.offset + list_offset,
layer_name=ntkrnlmp.layer_name,
)
for callback in callback_record.Entry:
if not context.layers[layer_name].is_valid(callback.CallbackRoutine, 64):
if not context.layers[ntkrnlmp.layer_name].is_valid(
callback.CallbackRoutine, 64
):
continue
try:
component = context.object(
symbol_table + constants.BANG + "string",
layer_name=layer_name,
component = ntkrnlmp.object(
"string",
offset=callback.Component,
max_length=64,
errors="replace",
@@ -708,8 +673,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
for callback_method in callback_methods:
for callback_type, callback_address, callback_detail in callback_method(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
callback_symbol_table,
):
if callback_detail is None:
volatility3/framework/plugins/windows/cmdscan.py
+5 -12
View File
@@ -24,7 +24,7 @@ class CmdScan(interfaces.plugins.PluginInterface):
"""Looks for Windows Command History lists"""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
@@ -39,7 +39,7 @@ class CmdScan(interfaces.plugins.PluginInterface):
name="pslist", component=pslist.PsList, version=(3, 0, 0)
),
requirements.PluginRequirement(
name="consoles", plugin=consoles.Consoles, version=(2, 0, 0)
name="consoles", plugin=consoles.Consoles, version=(3, 0, 0)
),
requirements.BooleanRequirement(
name="no_registry",
@@ -83,9 +83,8 @@ class CmdScan(interfaces.plugins.PluginInterface):
def get_command_history(
cls,
context: interfaces.context.ContextInterface,
kernel_layer_name: str,
kernel_symbol_table_name: str,
config_path: str,
kernel_module_name: str,
procs: Generator[interfaces.objects.ObjectInterface, None, None],
max_history: Set[int],
) -> Tuple[
@@ -97,8 +96,6 @@ class CmdScan(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
kernel_layer_name: The name of the layer on which to operate
kernel_symbol_table_name: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
procs: List of process objects
max_history: An initial set of CommandHistorySize values
@@ -135,9 +132,8 @@ class CmdScan(interfaces.plugins.PluginInterface):
if conhost_symbol_table is None:
conhost_symbol_table = consoles.Consoles.create_conhost_symbol_table(
context,
kernel_layer_name,
kernel_symbol_table_name,
config_path,
kernel_module_name,
proc_layer_name,
conhostexe_base,
)
@@ -279,8 +275,6 @@ class CmdScan(interfaces.plugins.PluginInterface):
procs: the process list filtered to conhost.exe instances
"""
kernel = self.context.modules[self.config["kernel"]]
max_history = set(self.config.get("max_history", [50]))
no_registry = self.config.get("no_registry")
@@ -302,9 +296,8 @@ class CmdScan(interfaces.plugins.PluginInterface):
command_history_properties,
) in self.get_command_history(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config_path,
self.config["kernel"],
procs,
max_history,
):
volatility3/framework/plugins/windows/consoles.py
+21 -29
View File
@@ -31,7 +31,7 @@ class Consoles(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 4, 0)
# 2.0.0 - change the signature of `get_console_settings_from_registry`
_version = (2, 0, 0)
_version = (3, 0, 0)
@classmethod
def get_requirements(cls):
@@ -128,9 +128,8 @@ class Consoles(interfaces.plugins.PluginInterface):
def determine_conhost_version(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
config_path: str,
kernel_module_name: str,
conhost_layer_name: str,
conhost_base: int,
) -> Tuple[Optional[str], Dict[str, Type]]:
@@ -139,9 +138,8 @@ class Consoles(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
kernel_module_name: The name of the module for the kernel
conhost_layer_name: The name of the conhot process memory layer
conhost_base: the base address of conhost.exe
@@ -149,8 +147,10 @@ class Consoles(interfaces.plugins.PluginInterface):
The filename of the symbol table to use and the associated class types.
"""
kernel = context.modules[kernel_module_name]
is_64bit = symbols.symbol_table_is_64bit(
context=context, symbol_table_name=nt_symbol_table
context=context, symbol_table_name=kernel.symbol_table_name
)
if is_64bit:
@@ -158,9 +158,9 @@ class Consoles(interfaces.plugins.PluginInterface):
else:
arch = "x86"
vers = info.Info.get_version_structure(context, layer_name, nt_symbol_table)
vers = info.Info.get_version_structure(context, kernel_module_name)
kuser = info.Info.get_kuser_structure(context, layer_name, nt_symbol_table)
kuser = info.Info.get_kuser_structure(context, kernel_module_name)
try:
vers_minor_version = int(vers.MinorVersion)
@@ -247,7 +247,7 @@ class Consoles(interfaces.plugins.PluginInterface):
)
except (exceptions.InvalidAddressException, TypeError, AttributeError):
# the following is IntelLayer specific and might need to be adapted to other architectures.
physical_layer_name = context.layers[layer_name].config.get(
physical_layer_name = context.layers[kernel.layer_name].config.get(
"memory_layer", None
)
if physical_layer_name:
@@ -318,9 +318,8 @@ class Consoles(interfaces.plugins.PluginInterface):
def create_conhost_symbol_table(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
config_path: str,
kernel_module_name: str,
conhost_layer_name: str,
conhost_base: int,
) -> str:
@@ -328,20 +327,20 @@ class Consoles(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
kernel_module_name: The name of the module of the kernel
Returns:
The name of the constructed symbol table
"""
table_mapping = {"nt_symbols": nt_symbol_table}
kernel = context.modules[kernel_module_name]
table_mapping = {"nt_symbols": kernel.symbol_table_name}
symbol_filename, class_types = cls.determine_conhost_version(
context,
layer_name,
nt_symbol_table,
config_path,
kernel_module_name,
conhost_layer_name,
conhost_base,
)
@@ -366,9 +365,8 @@ class Consoles(interfaces.plugins.PluginInterface):
def get_console_info(
cls,
context: interfaces.context.ContextInterface,
kernel_layer_name: str,
kernel_table_name: str,
config_path: str,
kernel_module_name: str,
procs: Generator[interfaces.objects.ObjectInterface, None, None],
max_history: Set[int],
max_buffers: Set[int],
@@ -385,9 +383,8 @@ class Consoles(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
kernel_layer_name: The name of the layer on which to operate
kernel_table_name: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
kernel_module_name: The name of the module for the kernel
procs: list of process objects
max_history: an initial set of CommandHistorySize values
max_buffers: an initial list of HistoryBufferMax values
@@ -427,9 +424,8 @@ class Consoles(interfaces.plugins.PluginInterface):
if conhost_symbol_table is None:
conhost_symbol_table = cls.create_conhost_symbol_table(
context,
kernel_layer_name,
kernel_table_name,
config_path,
kernel_module_name,
proc_layer_name,
conhostexe_base,
)
@@ -486,7 +482,7 @@ class Consoles(interfaces.plugins.PluginInterface):
console_properties.append(
{
"level": 1,
"name": "_CONSOLE_INFORMATION.ScreenX",
"kernel_module_nameme": "_CONSOLE_INFORMATION.ScreenX",
"address": console_info.ScreenX.vol.offset,
"data": console_info.ScreenX,
}
@@ -810,8 +806,7 @@ class Consoles(interfaces.plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
config_path: The config path where to find symbol files
kernel_layer_name: The name of the layer on which to operate
kernel_symbol_table_name: The name of the table containing the kernel symbols
kernel_module_name: The name of the module for the kernel
max_history: an initial set of CommandHistorySize values
max_buffers: an initial list of HistoryBufferMax values
@@ -853,8 +848,6 @@ class Consoles(interfaces.plugins.PluginInterface):
procs: the process list filtered to conhost.exe instances
"""
kernel = self.context.modules[self.config["kernel"]]
max_history = set(self.config.get("max_history", [50]))
max_buffers = set(self.config.get("max_buffers", [4]))
no_registry = self.config.get("no_registry")
@@ -874,9 +867,8 @@ class Consoles(interfaces.plugins.PluginInterface):
proc = None
for proc, console_info, console_properties in self.get_console_info(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config_path,
self.config["kernel"],
procs,
max_history,
max_buffers,
volatility3/framework/plugins/windows/dlllist.py
+4 -9
View File
@@ -37,13 +37,13 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
name="pslist", component=pslist.PsList, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="psscan", component=psscan.PsScan, version=(1, 1, 0)
name="psscan", component=psscan.PsScan, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="pedump", component=pedump.PEDump, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
name="info", component=info.Info, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -85,11 +85,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
self.context, self.config_path, "windows", "pe", class_types=pe.class_types
)
kernel = self.context.modules[self.config["kernel"]]
kuser = info.Info.get_kuser_structure(
self.context, kernel.layer_name, kernel.symbol_table_name
)
kuser = info.Info.get_kuser_structure(self.context, self.config["kernel"])
nt_major_version = int(kuser.NtMajorVersion)
nt_minor_version = int(kuser.NtMinorVersion)
@@ -209,8 +205,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
if self.config["offset"]:
procs = psscan.PsScan.scan_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
filter_func=psscan.PsScan.create_offset_filter(
self.context,
kernel.layer_name,
volatility3/framework/plugins/windows/driverscan.py
+4 -8
View File
@@ -25,7 +25,7 @@ class DriverScan(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
]
@@ -48,15 +48,11 @@ class DriverScan(interfaces.plugins.PluginInterface):
kernel = context.modules[kernel_module_name]
symbol_table_name = kernel.symbol_table_name
layer_name = kernel.layer_name
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table_name, [b"Dri\xf6", b"Driv"]
kernel.symbol_table_name, [b"Dri\xf6", b"Driv"]
)
module = context.module(symbol_table_name, layer_name, 0)
driver_start_offset = module.get_type("_DRIVER_OBJECT").relative_child_offset(
driver_start_offset = kernel.get_type("_DRIVER_OBJECT").relative_child_offset(
"DriverStart"
)
@@ -65,7 +61,7 @@ class DriverScan(interfaces.plugins.PluginInterface):
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table_name, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
volatility3/framework/plugins/windows/filescan.py
+9 -13
View File
@@ -14,7 +14,7 @@ class FileScan(interfaces.plugins.PluginInterface):
"""Scans for file objects present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
@@ -25,7 +25,7 @@ class FileScan(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
]
@@ -33,36 +33,32 @@ class FileScan(interfaces.plugins.PluginInterface):
def scan_files(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Scans for file objects using the poolscanner module and constraints.
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
kernel_module_name: The name of the module for the kernel
Returns:
A list of File objects as found from the `layer_name` layer based on File pool signatures
"""
kernel = context.modules[kernel_module_name]
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table, [b"Fil\xe5", b"File"]
kernel.symbol_table_name, [b"Fil\xe5", b"File"]
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
for fileobj in self.scan_files(
self.context, kernel.layer_name, kernel.symbol_table_name
):
for fileobj in self.scan_files(self.context, self.config["kernel"]):
try:
file_name = fileobj.FileName.String
except exceptions.InvalidAddressException:
volatility3/framework/plugins/windows/handles.py
+2 -3
View File
@@ -39,7 +39,7 @@ class Handles(interfaces.plugins.PluginInterface):
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="psscan", component=psscan.PsScan, version=(1, 1, 0)
name="psscan", component=psscan.PsScan, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -376,8 +376,7 @@ class Handles(interfaces.plugins.PluginInterface):
if self.config["offset"]:
procs = psscan.PsScan.scan_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
filter_func=psscan.PsScan.create_offset_filter(
self.context,
kernel.layer_name,
volatility3/framework/plugins/windows/indirect_system_calls.py
+1 -1
View File
@@ -52,7 +52,7 @@ class IndirectSystemCalls(direct_system_calls.DirectSystemCalls):
requirements.PluginRequirement(
name="direct_system_calls",
plugin=direct_system_calls.DirectSystemCalls,
version=(1, 0, 0),
version=(2, 0, 0),
),
]
volatility3/framework/plugins/windows/info.py
+35 -35
View File
@@ -17,7 +17,7 @@ class Info(plugins.PluginInterface):
"""Show OS & kernel details of the memory sample being analyzed."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -56,6 +56,9 @@ class Info(plugins.PluginInterface):
# FileLayer won't have dependencies
pass
# FIXME - this needs to be deprecated. This is exactly the same
# as getting it from context.modules
# Deprecation warning will go once the API is overhauled
@classmethod
def get_kernel_module(
cls,
@@ -80,13 +83,12 @@ class Info(plugins.PluginInterface):
cls,
context: interfaces.context.ContextInterface,
config_path: str,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> interfaces.objects.ObjectInterface:
"""Returns the KDDEBUGGER_DATA64 structure for a kernel"""
ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table)
ntkrnlmp = context.modules[kernel_module_name]
native_types = context.symbol_space[symbol_table].natives
native_types = context.symbol_space[ntkrnlmp.symbol_table_name].natives
kdbg_offset = ntkrnlmp.get_symbol("KdDebuggerDataBlock").address
@@ -102,7 +104,7 @@ class Info(plugins.PluginInterface):
kdbg_obj = context.object(
kdbg_table_name + constants.BANG + "_KDDEBUGGER_DATA64",
offset=ntkrnlmp.offset + kdbg_offset,
layer_name=layer_name,
layer_name=ntkrnlmp.layer_name,
)
return kdbg_obj
@@ -111,16 +113,15 @@ class Info(plugins.PluginInterface):
def get_kuser_structure(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> interfaces.objects.ObjectInterface:
"""Returns the _KUSER_SHARED_DATA structure for a kernel"""
virtual_layer = context.layers[layer_name]
ntkrnlmp = context.modules[kernel_module_name]
virtual_layer = context.layers[ntkrnlmp.layer_name]
if not isinstance(virtual_layer, layers.intel.Intel):
raise TypeError("Virtual Layer is not an intel layer")
ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table)
# this is a hard-coded address in the Windows OS
if virtual_layer.bits_per_register == 32:
kuser_addr = 0xFFDF0000
@@ -129,7 +130,6 @@ class Info(plugins.PluginInterface):
kuser = ntkrnlmp.object(
object_type="_KUSER_SHARED_DATA",
layer_name=layer_name,
offset=kuser_addr,
absolute=True,
)
@@ -140,17 +140,15 @@ class Info(plugins.PluginInterface):
def get_version_structure(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> interfaces.objects.ObjectInterface:
"""Returns the KdVersionBlock information from a kernel"""
ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table)
ntkrnlmp = context.modules[kernel_module_name]
vers_offset = ntkrnlmp.get_symbol("KdVersionBlock").address
vers = ntkrnlmp.object(
object_type="_DBGKD_GET_VERSION64",
layer_name=layer_name,
offset=vers_offset,
)
@@ -193,35 +191,38 @@ class Info(plugins.PluginInterface):
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
layer_name = kernel.layer_name
symbol_table = kernel.symbol_table_name
layer = self.context.layers[layer_name]
table = self.context.symbol_space[symbol_table]
kernel_layer = self.context.layers[kernel.layer_name]
symbol_table = self.context.symbol_space[kernel.symbol_table_name]
kdbg = self.get_kdbg_structure(
self.context, self.config_path, layer_name, symbol_table
self.context,
self.config_path,
self.config["kernel"],
)
yield (0, ("Kernel Base", hex(layer.config["kernel_virtual_offset"])))
yield (0, ("DTB", hex(layer.config["page_map_offset"])))
yield (0, ("Symbols", table.config["isf_url"]))
yield (0, ("Kernel Base", hex(kernel_layer.config["kernel_virtual_offset"])))
yield (0, ("DTB", hex(kernel_layer.config["page_map_offset"])))
yield (0, ("Symbols", symbol_table.config["isf_url"]))
yield (
0,
(
"Is64Bit",
str(
symbols.symbol_table_is_64bit(
context=self.context, symbol_table_name=symbol_table
context=self.context, symbol_table_name=kernel.symbol_table_name
)
),
),
)
yield (
0,
("IsPAE", str(self.context.layers[layer_name].metadata.get("pae", False))),
(
"IsPAE",
str(self.context.layers[kernel.layer_name].metadata.get("pae", False)),
),
)
for i, layer in self.get_depends(self.context, layer_name):
for i, layer in self.get_depends(self.context, kernel.layer_name):
yield (0, (layer.name, f"{i} {layer.__class__.__name__}"))
if kdbg.Header.OwnerTag == 0x4742444B:
@@ -229,23 +230,22 @@ class Info(plugins.PluginInterface):
yield (0, ("NTBuildLab", kdbg.get_build_lab()))
yield (0, ("CSDVersion", str(kdbg.get_csdversion())))
vers = self.get_version_structure(self.context, layer_name, symbol_table)
vers = self.get_version_structure(self.context, self.config["kernel"])
yield (0, ("KdVersionBlock", hex(vers.vol.offset)))
yield (0, ("Major/Minor", f"{vers.MajorVersion}.{vers.MinorVersion}"))
yield (0, ("MachineType", str(vers.MachineType)))
ntkrnlmp = self.get_kernel_module(self.context, layer_name, symbol_table)
cpu_count_offset = kernel.get_symbol("KeNumberProcessors").address
cpu_count_offset = ntkrnlmp.get_symbol("KeNumberProcessors").address
cpu_count = ntkrnlmp.object(
object_type="unsigned int", layer_name=layer_name, offset=cpu_count_offset
cpu_count = kernel.object(
object_type="unsigned int",
offset=cpu_count_offset,
)
yield (0, ("KeNumberProcessors", str(cpu_count)))
kuser = self.get_kuser_structure(self.context, layer_name, symbol_table)
kuser = self.get_kuser_structure(self.context, self.config["kernel"])
yield (0, ("SystemTime", str(kuser.SystemTime.get_time())))
yield (
@@ -266,7 +266,7 @@ class Info(plugins.PluginInterface):
# yield (0, ("SafeBootMode", "True" if kuser.SafeBootMode else "False"))
nt_header = self.get_ntheader_structure(
self.context, self.config_path, layer_name
self.context, self.config_path, kernel.layer_name
)
yield (
volatility3/framework/plugins/windows/modscan.py
+2 -2
View File
@@ -32,7 +32,7 @@ class ModScan(modules.Modules):
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="modules", component=modules.Modules, version=(3, 0, 0)
@@ -81,7 +81,7 @@ class ModScan(modules.Modules):
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, kernel.layer_name, kernel.symbol_table_name, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
volatility3/framework/plugins/windows/mutantscan.py
+9 -12
View File
@@ -14,6 +14,7 @@ class MutantScan(interfaces.plugins.PluginInterface):
"""Scans for mutexes present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
@@ -24,7 +25,7 @@ class MutantScan(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
]
@@ -32,36 +33,32 @@ class MutantScan(interfaces.plugins.PluginInterface):
def scan_mutants(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Scans for mutants using the poolscanner module and constraints.
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
kernel_module_name: The name of the module for the kernel
Returns:
A list of Mutant objects found by scanning memory for the Mutant pool signatures
"""
kernel = context.modules[kernel_module_name]
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table, [b"Mut\xe1", b"Muta"]
kernel.symbol_table_name, [b"Mut\xe1", b"Muta"]
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
for mutant in self.scan_mutants(
self.context, kernel.layer_name, kernel.symbol_table_name
):
for mutant in self.scan_mutants(self.context, self.config["kernel"]):
try:
name = mutant.get_name()
except (ValueError, exceptions.InvalidAddressException):
volatility3/framework/plugins/windows/netscan.py
+23 -29
View File
@@ -23,7 +23,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for network objects present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
@@ -34,10 +34,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
name="info", component=info.Info, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="verinfo", component=verinfo.VerInfo, version=(1, 0, 0)
@@ -117,15 +117,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def determine_tcpip_version(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
kernel_module_name: str,
) -> Tuple[str, Type]:
"""Tries to determine which symbol filename to use for the image's tcpip driver. The logic is partially taken from the info plugin.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
kernel_module_name: Name of the module for the kernel
Returns:
The filename of the symbol table to use.
@@ -137,12 +135,14 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# therefore we determine the version based on the kernel version as testing
# with several windows versions has showed this to work out correctly.
kernel = context.modules[kernel_module_name]
is_64bit = symbols.symbol_table_is_64bit(
context=context, symbol_table_name=nt_symbol_table
context=context, symbol_table_name=kernel.symbol_table_name
)
is_18363_or_later = versions.is_win10_18363_or_later(
context=context, symbol_table=nt_symbol_table
context=context, symbol_table=kernel.symbol_table_name
)
if is_64bit:
@@ -150,9 +150,9 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
else:
arch = "x86"
vers = info.Info.get_version_structure(context, layer_name, nt_symbol_table)
vers = info.Info.get_version_structure(context, kernel_module_name)
kuser = info.Info.get_kuser_structure(context, layer_name, nt_symbol_table)
kuser = info.Info.get_kuser_structure(context, kernel_module_name)
try:
vers_minor_version = int(vers.MinorVersion)
@@ -259,7 +259,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"Requiring further version inspection due to OS version by checking tcpip.sys's FileVersion header"
)
# the following is IntelLayer specific and might need to be adapted to other architectures.
physical_layer_name = context.layers[layer_name].config.get(
physical_layer_name = context.layers[kernel.layer_name].config.get(
"memory_layer", None
)
if physical_layer_name:
@@ -322,27 +322,26 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def create_netscan_symbol_table(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
kernel_module_name: str,
config_path: str,
) -> str:
"""Creates a symbol table for TCP Listeners and TCP/UDP Endpoints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
kernel_module_name: Name of the module for the kernel
config_path: The config path where to find symbol files
Returns:
The name of the constructed symbol table
"""
table_mapping = {"nt_symbols": nt_symbol_table}
kernel = context.modules[kernel_module_name]
table_mapping = {"nt_symbols": kernel.symbol_table_name}
symbol_filename, class_types = cls.determine_tcpip_version(
context,
layer_name,
nt_symbol_table,
kernel_module_name,
)
return intermed.IntermediateSymbolTable.create(
@@ -358,16 +357,14 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def scan(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
kernel_module_name: str,
netscan_symbol_table: str,
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Scans for network objects using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
kernel_module_name: The name of the module for the kernel
netscan_symbol_table: The name of the table containing the network object symbols (_TCP_LISTENER etc.)
Returns:
@@ -377,7 +374,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
constraints = cls.create_netscan_constraints(context, netscan_symbol_table)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, nt_symbol_table, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
@@ -385,16 +382,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def _generator(self, show_corrupt_results: Optional[bool] = None):
"""Generates the network objects for use in rendering."""
kernel = self.context.modules[self.config["kernel"]]
netscan_symbol_table = self.create_netscan_symbol_table(
self.context, kernel.layer_name, kernel.symbol_table_name, self.config_path
self.context, self.config["kernel"], self.config_path
)
for netw_obj in self.scan(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
netscan_symbol_table,
):
vollog.debug(
volatility3/framework/plugins/windows/netstat.py
+3 -3
View File
@@ -34,7 +34,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="netscan", component=netscan.NetScan, version=(1, 0, 0)
name="netscan", component=netscan.NetScan, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="modules", component=modules.Modules, version=(3, 0, 0)
@@ -43,7 +43,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
name="pdbutil", component=pdbutil.PDBUtility, version=(1, 0, 0)
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
name="info", component=info.Info, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="verinfo", component=verinfo.VerInfo, version=(1, 0, 0)
@@ -629,7 +629,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
kernel = self.context.modules[self.config["kernel"]]
netscan_symbol_table = netscan.NetScan.create_netscan_symbol_table(
self.context, kernel.layer_name, kernel.symbol_table_name, self.config_path
self.context, self.config["kernel"], self.config_path
)
tcpip_module = self.get_tcpip_module(self.context, self.config["kernel"])
volatility3/framework/plugins/windows/poolscanner.py
+21 -22
View File
@@ -129,7 +129,7 @@ class PoolScanner(plugins.PluginInterface):
"""A generic pool scanner plugin."""
_required_framework_version = (2, 0, 0)
_version = (1, 1, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -151,7 +151,7 @@ class PoolScanner(plugins.PluginInterface):
constraints = self.builtin_constraints(symbol_table)
for constraint, mem_object, header in self.generate_pool_scan(
self.context, kernel.layer_name, symbol_table, constraints
self.context, self.config["kernel"], constraints
):
# generate some type-specific info for sanity checking
if constraint.object_type == "Process":
@@ -365,8 +365,7 @@ class PoolScanner(plugins.PluginInterface):
def generate_pool_scan_extended(
cls,
context: interfaces.context.ContextInterface,
kernel_layer_name: str,
kernel_symbol_table_name: str,
kernel_module_name: str,
object_symbol_table_name: str,
constraints: List[PoolConstraint],
) -> Generator[
@@ -384,41 +383,42 @@ class PoolScanner(plugins.PluginInterface):
Args:
context: The context to retrieve required elements (layers, symbol tables) from
kernel_layer_name: The name of the base kernel layer
kernel_symbol_table_name: The name of the table containing the kernel symbols
kernel_module_name: The name of the module for the kernel
object_symbol_table_name: The name of the symbol table for the object being scanned for
constraints: List of pool constraints used to limit the scan results
Returns:
Iterable of tuples, containing the constraint that matched, the object from memory, the object header used to determine the object
"""
kernel = context.modules[kernel_module_name]
# get the object type map
type_map = handles.Handles.get_type_map(
context=context,
layer_name=kernel_layer_name,
symbol_table=kernel_symbol_table_name,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
)
cookie = handles.Handles.find_cookie(
context=context,
layer_name=kernel_layer_name,
symbol_table=kernel_symbol_table_name,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
)
is_windows_10 = versions.is_windows_10(context, kernel_symbol_table_name)
is_windows_10 = versions.is_windows_10(context, kernel.symbol_table_name)
is_windows_8_or_later = versions.is_windows_8_or_later(
context, kernel_symbol_table_name
context, kernel.symbol_table_name
)
# start off with the primary virtual layer
scan_layer = kernel_layer_name
scan_layer = kernel.layer_name
# switch to a non-virtual layer if necessary
if not is_windows_10:
scan_layer = context.layers[scan_layer].config["memory_layer"]
if symbols.symbol_table_is_64bit(
context=context, symbol_table_name=kernel_symbol_table_name
context=context, symbol_table_name=kernel.symbol_table_name
):
alignment = 0x10
else:
@@ -433,12 +433,11 @@ class PoolScanner(plugins.PluginInterface):
alignment=alignment,
):
# construct the object in its own layer, using its own types
mem_objects = header.get_object(
constraint=constraint,
use_top_down=is_windows_8_or_later,
native_layer_name=kernel_layer_name,
kernel_symbol_table=kernel_symbol_table_name,
native_layer_name=kernel.layer_name,
kernel_symbol_table=kernel.symbol_table_name,
)
for mem_object in mem_objects:
@@ -472,8 +471,7 @@ class PoolScanner(plugins.PluginInterface):
def generate_pool_scan(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
constraints: List[PoolConstraint],
) -> Generator[
Tuple[
@@ -489,17 +487,18 @@ class PoolScanner(plugins.PluginInterface):
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
kernel_module_name: The name of the module for the kernel
constraints: List of pool constraints used to limit the scan results
Returns:
Iterable of tuples, containing the constraint that matched, the object from memory, the object header used to determine the object
"""
kernel = context.modules[kernel_module_name]
# repeat the symbol table to match the original `generate_pool_scan` behaviour
yield from cls.generate_pool_scan_extended(
context, layer_name, symbol_table, symbol_table, constraints
context, kernel_module_name, kernel.symbol_table_name, constraints
)
@classmethod
volatility3/framework/plugins/windows/psscan.py
+24 -34
View File
@@ -23,7 +23,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for processes present in a particular windows memory image."""
_required_framework_version = (2, 3, 1)
_version = (1, 1, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
@@ -37,7 +37,10 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
name="info", component=info.Info, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -141,8 +144,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def scan_processes(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
filter_func: Callable[
[interfaces.objects.ObjectInterface], bool
] = lambda _: False,
@@ -151,19 +153,20 @@ class PsScan(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
kernel_module_name: The name of the module for the kernel
Returns:
A list of processes found by scanning the `layer_name` layer for process pool signatures
"""
kernel = context.modules[kernel_module_name]
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table, [b"Pro\xe3", b"Proc"]
kernel.symbol_table_name, [b"Pro\xe3", b"Proc"]
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
if not filter_func(mem_object):
@@ -173,16 +176,14 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def virtual_process_from_physical(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
proc: interfaces.objects.ObjectInterface,
) -> Optional[interfaces.objects.ObjectInterface]:
"""Returns a virtual process from a physical addressed one
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
kernel_module_name: The name of the module inside the kernel
proc: the process object with physical address
Returns:
@@ -190,16 +191,10 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""
version = cls.get_osversion(context, layer_name, symbol_table)
ntkrnlmp = context.modules[kernel_module_name]
version = cls.get_osversion(context, kernel_module_name)
# If it's WinXP->8.1 we have now a physical process address.
# We'll use the first thread to bounce back to the virtual process
kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
if not kvo:
raise ValueError(
"Intel layer does not have an associated kernel virtual offset, failing"
)
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
tleoffset = ntkrnlmp.get_type("_ETHREAD").relative_child_offset(
"ThreadListEntry"
)
@@ -208,7 +203,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# If (and only if) we're dealing with 64-bit Windows 7 SP1
# then add the other commonly seen member offset to the list
bits = context.layers[layer_name].bits_per_register
bits = context.layers[ntkrnlmp.layer_name].bits_per_register
if version == (6, 1, 7601) and bits == 64:
offsets.append(tleoffset + 8)
@@ -225,7 +220,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# Sanity check the bounce.
# This compares the original offset with the new one (translated from virtual layer)
(_, _, ph_offset, _, _) = list(
context.layers[layer_name].mapping(
context.layers[ntkrnlmp.layer_name].mapping(
offset=virtual_process.vol.offset, length=0
)
)[0]
@@ -237,23 +232,20 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def get_osversion(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> Tuple[int, int, int]:
"""Returns the complete OS version (MAJ,MIN,BUILD)
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
kernel_module_name: The name of the module for the kernel
Returns:
A tuple with (MAJ,MIN,BUILD)
"""
kuser = info.Info.get_kuser_structure(context, layer_name, symbol_table)
kuser = info.Info.get_kuser_structure(context, kernel_module_name)
nt_major_version = int(kuser.NtMajorVersion)
nt_minor_version = int(kuser.NtMinorVersion)
vers = info.Info.get_version_structure(context, layer_name, symbol_table)
vers = info.Info.get_version_structure(context, kernel_module_name)
build = vers.MinorVersion
return (nt_major_version, nt_minor_version, build)
@@ -268,8 +260,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
for proc in self.scan_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
filter_func=pslist.PsList.create_pid_filter(self.config.get("pid", None)),
):
file_output = "Disabled"
@@ -281,8 +272,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
try:
vproc = self.virtual_process_from_physical(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config["kernel"],
proc,
)
except exceptions.PagedInvalidAddressException:
volatility3/framework/plugins/windows/psxview.py
+13 -19
View File
@@ -11,7 +11,6 @@ from volatility3.framework.renderers import TreeGrid, format_hints
from volatility3.framework.symbols.windows import extensions
from volatility3.plugins.windows import (
handles,
info,
pslist,
psscan,
thrdscan,
@@ -49,14 +48,11 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
),
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="psscan", component=psscan.PsScan, version=(1, 0, 0)
name="psscan", component=psscan.PsScan, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="thrdscan", component=thrdscan.ThrdScan, version=(1, 0, 0)
@@ -114,10 +110,10 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
return self._proc_list_to_dict(tasks)
def _check_psscan(
self, layer_name: str, symbol_table: str
self,
) -> Dict[int, extensions.EPROCESS]:
res = psscan.PsScan.scan_processes(
context=self.context, layer_name=layer_name, symbol_table=symbol_table
context=self.context, kernel_module_name=self.config["kernel"]
)
return self._proc_list_to_dict(res)
@@ -144,20 +140,24 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
return self._proc_list_to_dict(ret)
def _check_csrss_handles(
self, tasks: Iterable[extensions.EPROCESS], layer_name: str, symbol_table: str
self, tasks: Iterable[extensions.EPROCESS]
) -> Dict[int, extensions.EPROCESS]:
ret: List[extensions.EPROCESS] = []
kernel = self.context.modules[self.config["kernel"]]
handles_plugin = handles.Handles(
context=self.context, config_path=self.config_path
)
type_map = handles_plugin.get_type_map(self.context, layer_name, symbol_table)
type_map = handles_plugin.get_type_map(
self.context, kernel.layer_name, kernel.symbol_table_name
)
cookie = handles_plugin.find_cookie(
context=self.context,
layer_name=layer_name,
symbol_table=symbol_table,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
)
for p in tasks:
@@ -179,8 +179,6 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
return self._proc_list_to_dict(ret)
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
kdbg_list_processes = list(
pslist.PsList.list_processes(
context=self.context, kernel_module_name=self.config["kernel"]
@@ -191,13 +189,9 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
processes: Dict[str, Dict[int, extensions.EPROCESS]] = {}
processes["pslist"] = self._check_pslist(kdbg_list_processes)
processes["psscan"] = self._check_psscan(
kernel.layer_name, kernel.symbol_table_name
)
processes["psscan"] = self._check_psscan()
processes["thrdscan"] = self._check_thrdscan()
processes["csrss"] = self._check_csrss_handles(
kdbg_list_processes, kernel.layer_name, kernel.symbol_table_name
)
processes["csrss"] = self._check_csrss_handles(kdbg_list_processes)
# Unique set of all offsets from all sources
offsets = set(chain(*(mapping.keys() for mapping in processes.values())))
volatility3/framework/plugins/windows/registry/hivescan.py
+4 -5
View File
@@ -26,10 +26,10 @@ class HiveScan(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="bigpools", plugin=bigpools.BigPools, version=(1, 0, 0)
name="bigpools", plugin=bigpools.BigPools, version=(2, 0, 0)
),
]
@@ -62,8 +62,7 @@ class HiveScan(interfaces.plugins.PluginInterface):
for pool in bigpools.BigPools.list_big_pools(
context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
kernel_module_name=kernel_name,
tags=["CM10"],
):
cmhive = ntkrnlmp.object(
@@ -77,7 +76,7 @@ class HiveScan(interfaces.plugins.PluginInterface):
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, kernel.layer_name, kernel.symbol_table_name, constraints
context, kernel_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
volatility3/framework/plugins/windows/svcscan.py
+1 -4
View File
@@ -20,7 +20,7 @@ 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 services as services_types
from volatility3.plugins.windows import poolscanner, pslist
from volatility3.plugins.windows import pslist
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
@@ -53,9 +53,6 @@ class SvcScan(interfaces.plugins.PluginInterface):
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="hivelist", plugin=hivelist.HiveList, version=(2, 0, 0)
),
volatility3/framework/plugins/windows/symlinkscan.py
+11 -9
View File
@@ -17,6 +17,8 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
return [
@@ -25,14 +27,16 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0)
),
]
@classmethod
def scan_symlinks(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
kernel_module_name: str,
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Scans for links using the poolscanner module and constraints.
@@ -45,22 +49,20 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
A list of symlink objects found by scanning memory for the Symlink pool signatures
"""
kernel = context.modules[kernel_module_name]
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table, [b"Sym\xe2", b"Symb"]
kernel.symbol_table_name, [b"Sym\xe2", b"Symb"]
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table, constraints
context, kernel_module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
for link in self.scan_symlinks(
self.context, kernel.layer_name, kernel.symbol_table_name
):
for link in self.scan_symlinks(self.context, self.config["kernel"]):
try:
from_name = link.get_link_name()
except (ValueError, exceptions.InvalidAddressException):
volatility3/framework/plugins/windows/thrdscan.py
+4 -6
View File
@@ -34,7 +34,7 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
name="poolscanner", plugin=poolscanner.PoolScanner, version=(2, 0, 0)
),
]
@@ -54,16 +54,14 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
A list of _ETHREAD objects found by scanning memory for the "Thre" / "Thr\\xE5" pool signatures
"""
module = context.modules[module_name]
layer_name = module.layer_name
symbol_table = module.symbol_table_name
kernel = context.modules[module_name]
constraints = poolscanner.PoolScanner.builtin_constraints(
symbol_table, [b"Thr\xe5", b"Thre"]
kernel.symbol_table_name, [b"Thr\xe5", b"Thre"]
)
for result in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, symbol_table, constraints
context, module_name, constraints
):
_constraint, mem_object, _header = result
yield mem_object
volatility3/framework/plugins/windows/windowstations.py
+7 -2
View File
@@ -46,6 +46,12 @@ class WindowStations(interfaces.plugins.PluginInterface):
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="poolscanner", component=poolscanner.PoolScanner, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="modules", component=modules.Modules, version=(3, 0, 0)
),
]
@staticmethod
@@ -152,8 +158,7 @@ class WindowStations(interfaces.plugins.PluginInterface):
for result in poolscanner.PoolScanner.generate_pool_scan_extended(
context=context,
kernel_layer_name=kernel.layer_name,
kernel_symbol_table_name=kernel.symbol_table_name,
kernel_module_name=kernel_module_name,
object_symbol_table_name=gui_table_name,
constraints=constraints,
):