mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-08 18:57:38 +02:00
Windows: Bug fixes and additions to modules and modscan
This pull request adds several changes to the modules and modscan plugins that fix bugs, unify options and operations, and allow for filtering by name and base address in both plugins. Previously, modscan and modules had a substantial amount of partially duplicated code, which led to bugs, such as: Only one allowed filtering by --name Only one cycled through the session layers to find the one hosting the kernel module With the new inheritance and combined implementation, the correct and complete implementation applies to both. This PR also addresses the feedback in https://github.com/volatilityfoundation/volatility3/pull/1099. In this PR, --name and –base apply to the processing of modules and determine both which plugins are displayed in output as well as which are processed for extraction. This fixes the issue of using --filters, which only filters the final command line output, but leads to every module being extracted.
This commit is contained in:
Brandon Barnacle
committed by
Brandon Barnacle
Brandon Barnacle
parent
887747e3ce
commit
b35169d0bc
@@ -9,16 +9,20 @@ from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.framework.symbols import intermed
|
||||
from volatility3.framework.symbols.windows.extensions import pe
|
||||
from volatility3.plugins.windows import poolscanner, dlllist, pslist
|
||||
from volatility3.plugins.windows import poolscanner, dlllist, pslist, modules
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ModScan(interfaces.plugins.PluginInterface):
|
||||
class ModScan(modules.Modules):
|
||||
"""Scans for modules present in a particular windows memory image."""
|
||||
|
||||
_required_framework_version = (2, 0, 0)
|
||||
_version = (1, 0, 0)
|
||||
_version = (2, 0, 0)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self._enumeration_method = self.scan_modules
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls):
|
||||
@@ -31,6 +35,9 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
requirements.VersionRequirement(
|
||||
name="poolscanner", component=poolscanner.PoolScanner, version=(1, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="modules", component=modules.Modules, version=(1, 1, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(2, 0, 0)
|
||||
),
|
||||
@@ -43,6 +50,17 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
default=False,
|
||||
optional=True,
|
||||
),
|
||||
requirements.IntRequirement(
|
||||
name="base",
|
||||
description="Extract a single module with BASE address",
|
||||
optional=True,
|
||||
),
|
||||
requirements.StringRequirement(
|
||||
name="name",
|
||||
description="module name/sub string",
|
||||
optional=True,
|
||||
default=None,
|
||||
),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@@ -72,157 +90,3 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
):
|
||||
_constraint, mem_object, _header = result
|
||||
yield mem_object
|
||||
|
||||
@classmethod
|
||||
def get_session_layers(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
layer_name: str,
|
||||
symbol_table: str,
|
||||
pids: List[int] = None,
|
||||
) -> Generator[str, None, None]:
|
||||
"""Build a cache of possible virtual layers, in priority starting with
|
||||
the primary/kernel layer. Then keep one layer per session by cycling
|
||||
through the process list.
|
||||
|
||||
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
|
||||
pids: A list of process identifiers to include exclusively or None for no filter
|
||||
|
||||
Returns:
|
||||
A list of session layer names
|
||||
"""
|
||||
seen_ids: List[interfaces.objects.ObjectInterface] = []
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
for proc in pslist.PsList.list_processes(
|
||||
context=context,
|
||||
layer_name=layer_name,
|
||||
symbol_table=symbol_table,
|
||||
filter_func=filter_func,
|
||||
):
|
||||
proc_id = "Unknown"
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
# create the session space object in the process' own layer.
|
||||
# not all processes have a valid session pointer.
|
||||
session_space = context.object(
|
||||
symbol_table + constants.BANG + "_MM_SESSION_SPACE",
|
||||
layer_name=layer_name,
|
||||
offset=proc.Session,
|
||||
)
|
||||
|
||||
if session_space.SessionId in seen_ids:
|
||||
continue
|
||||
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV,
|
||||
"Process {} does not have a valid Session or a layer could not be constructed for it".format(
|
||||
proc_id
|
||||
),
|
||||
)
|
||||
continue
|
||||
|
||||
# save the layer if we haven't seen the session yet
|
||||
seen_ids.append(session_space.SessionId)
|
||||
yield proc_layer_name
|
||||
|
||||
@classmethod
|
||||
def find_session_layer(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
session_layers: Iterable[str],
|
||||
base_address: int,
|
||||
):
|
||||
"""Given a base address and a list of layer names, find a layer that
|
||||
can access the specified address.
|
||||
|
||||
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
|
||||
session_layers: A list of session layer names
|
||||
base_address: The base address to identify the layers that can access it
|
||||
|
||||
Returns:
|
||||
Layer name or None if no layers that contain the base address can be found
|
||||
"""
|
||||
|
||||
for layer_name in session_layers:
|
||||
if context.layers[layer_name].is_valid(base_address):
|
||||
return layer_name
|
||||
|
||||
return None
|
||||
|
||||
def _generator(self):
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
|
||||
session_layers = list(
|
||||
self.get_session_layers(
|
||||
self.context, kernel.layer_name, kernel.symbol_table_name
|
||||
)
|
||||
)
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(
|
||||
self.context, self.config_path, "windows", "pe", class_types=pe.class_types
|
||||
)
|
||||
|
||||
for mod in self.scan_modules(
|
||||
self.context, kernel.layer_name, kernel.symbol_table_name
|
||||
):
|
||||
try:
|
||||
BaseDllName = mod.BaseDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
BaseDllName = ""
|
||||
|
||||
try:
|
||||
FullDllName = mod.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
FullDllName = ""
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config["dump"]:
|
||||
session_layer_name = self.find_session_layer(
|
||||
self.context, session_layers, mod.DllBase
|
||||
)
|
||||
file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}"
|
||||
if session_layer_name:
|
||||
file_handle = dlllist.DllList.dump_pe(
|
||||
self.context,
|
||||
pe_table_name,
|
||||
mod,
|
||||
self.open,
|
||||
layer_name=session_layer_name,
|
||||
)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
file_output,
|
||||
),
|
||||
)
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
("Offset", format_hints.Hex),
|
||||
("Base", format_hints.Hex),
|
||||
("Size", format_hints.Hex),
|
||||
("Name", str),
|
||||
("Path", str),
|
||||
("File output", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
|
||||
@@ -22,6 +22,10 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
_required_framework_version = (2, 0, 0)
|
||||
_version = (1, 1, 0)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self._enumeration_method = self.list_modules
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
return [
|
||||
@@ -42,6 +46,11 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
default=False,
|
||||
optional=True,
|
||||
),
|
||||
requirements.IntRequirement(
|
||||
name="base",
|
||||
description="Extract a single module with BASE address",
|
||||
optional=True,
|
||||
),
|
||||
requirements.StringRequirement(
|
||||
name="name",
|
||||
description="module name/sub string",
|
||||
@@ -50,49 +59,74 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
),
|
||||
]
|
||||
|
||||
def dump_module(self, session_layers, pe_table_name, mod):
|
||||
session_layer_name = self.find_session_layer(
|
||||
self.context, session_layers, mod.DllBase
|
||||
)
|
||||
file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}"
|
||||
if session_layer_name:
|
||||
file_handle = dlllist.DllList.dump_pe(
|
||||
self.context,
|
||||
pe_table_name,
|
||||
mod,
|
||||
self.open,
|
||||
layer_name=session_layer_name,
|
||||
)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
return file_output
|
||||
|
||||
def process_module(self, session_layers, pe_table_name, mod):
|
||||
if self.config["base"] and self.config["base"] != mod.DllBase:
|
||||
return None
|
||||
|
||||
try:
|
||||
BaseDllName = mod.BaseDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
BaseDllName = ""
|
||||
|
||||
if self.config["name"] and self.config["name"] not in BaseDllName:
|
||||
return None
|
||||
|
||||
try:
|
||||
FullDllName = mod.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
FullDllName = ""
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config["dump"]:
|
||||
file_output = self.dump_module(session_layers, pe_table_name, mod)
|
||||
|
||||
return (
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
file_output,
|
||||
)
|
||||
|
||||
def _generator(self):
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(
|
||||
self.context, self.config_path, "windows", "pe", class_types=pe.class_types
|
||||
)
|
||||
|
||||
for mod in self.list_modules(
|
||||
session_layers = list(
|
||||
self.get_session_layers(
|
||||
self.context, kernel.layer_name, kernel.symbol_table_name
|
||||
)
|
||||
)
|
||||
|
||||
for mod in self._enumeration_method(
|
||||
self.context, kernel.layer_name, kernel.symbol_table_name
|
||||
):
|
||||
try:
|
||||
BaseDllName = mod.BaseDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
BaseDllName = ""
|
||||
|
||||
try:
|
||||
FullDllName = mod.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
FullDllName = ""
|
||||
|
||||
if self.config["name"] and self.config["name"] not in BaseDllName:
|
||||
continue
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config["dump"]:
|
||||
file_handle = dlllist.DllList.dump_pe(
|
||||
self.context, pe_table_name, mod, self.open
|
||||
)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_handle.close()
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
file_output,
|
||||
),
|
||||
)
|
||||
record = self.process_module(session_layers, pe_table_name, mod)
|
||||
if record:
|
||||
yield (0, record)
|
||||
|
||||
@classmethod
|
||||
def get_session_layers(
|
||||
|
||||
Reference in New Issue
Block a user