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

Pool: Make object_header type checking the plugin's responsibility.

Browse Source
This commit is contained in:
Mike Auty
2019-12-04 22:11:42 +00:00
committed by ikelos
parent 8691c68604
commit 386f94d9ee
3 changed files with 40 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility/framework/plugins/windows/poolscanner.py
+16 -11
View File
@@ -13,7 +13,6 @@ from volatility.framework.layers import scanners
from volatility.framework.renderers import format_hints
from volatility.framework.symbols import intermed
from volatility.framework.symbols.windows import extensions
from volatility.framework.symbols.windows.extensions import pool
from volatility.plugins.windows import handles
vollog = logging.getLogger(__name__)
@@ -41,7 +40,8 @@ class PoolConstraint:
page_type: Optional[int] = None,
size: Optional[Tuple[Optional[int], Optional[int]]] = None,
index: Optional[Tuple[Optional[int], Optional[int]]] = None,
alignment: Optional[int] = 1) -> None:
alignment: Optional[int] = 1,
skip_type_test: bool = False) -> None:
self.tag = tag
self.type_name = type_name
self.object_type = object_type
@@ -49,6 +49,7 @@ class PoolConstraint:
self.size = size
self.index = index
self.alignment = alignment
self.skip_type_test = skip_type_test
class PoolHeaderScanner(interfaces.layers.ScannerInterface):
@@ -101,6 +102,7 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface):
if constraint.index[1]:
if header.PoolIndex > constraint.index[1]:
continue
except exceptions.InvalidAddressException:
# The tested object's header doesn't point to valid addresses, ignore it
continue
@@ -139,10 +141,10 @@ def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
# try the primary method based on the pe version in the ISF
def method(context: interfaces.context.ContextInterface, symbol_table: str) -> bool:
"""
Args:
context: The context that contains the symbol table named `symbol_table`
symbol_table: Name of the symbol table within the context to distinguish the version of
context: The context that contains the symbol table named `symbol_table`
symbol_table: Name of the symbol table within the context to distinguish the version of
Returns:
True if the symbol table is of the required version
@@ -249,6 +251,7 @@ class PoolScanner(plugins.PluginInterface):
type_name = symbol_table + constants.BANG + "_EPROCESS",
object_type = "Process",
size = (600, None),
skip_type_test = True,
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
# processes on windows starting with windows 8
PoolConstraint(b'Proc',
@@ -330,12 +333,12 @@ class PoolScanner(plugins.PluginInterface):
-> Generator[Tuple[
PoolConstraint, interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface], None, None]:
"""
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
constraints: List of pool constraints used to limit the scan results
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
@@ -359,16 +362,18 @@ class PoolScanner(plugins.PluginInterface):
for constraint, header in cls.pool_scan(context, scan_layer, symbol_table, constraints, alignment = 8):
mem_object = header.get_object(type_name = constraint.type_name,
type_map = type_map,
use_top_down = is_windows_8_or_later,
object_type = constraint.object_type,
native_layer_name = 'primary',
cookie = cookie)
executive = constraint.object_type is not None,
native_layer_name = 'primary')
if mem_object is None:
vollog.log(constants.LOGLEVEL_VVV, "Cannot create an instance of {}".format(constraint.type_name))
continue
if not constraint.skip_type_test:
if mem_object.get_object_header().get_object_type(type_map, cookie) != constraint.object_type:
continue
yield constraint, mem_object, header
@classmethod
volatility/framework/symbols/windows/extensions/__init__.py
+4 -4
View File
@@ -314,7 +314,7 @@ class DEVICE_OBJECT(objects.StructType, pool.ExecutiveObject):
"""A class for kernel device objects."""
def get_device_name(self) -> str:
header = self.object_header()
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
@@ -322,7 +322,7 @@ class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
"""A class for kernel driver objects."""
def get_driver_name(self) -> str:
header = self.object_header()
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def is_valid(self) -> bool:
@@ -334,7 +334,7 @@ class OBJECT_SYMBOLIC_LINK(objects.StructType, pool.ExecutiveObject):
"""A class for kernel link objects."""
def get_link_name(self) -> str:
header = self.object_header()
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def is_valid(self) -> bool:
@@ -375,7 +375,7 @@ class KMUTANT(objects.StructType, pool.ExecutiveObject):
def get_name(self) -> str:
"""Get the object's name from the object header."""
header = self.object_header()
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
volatility/framework/symbols/windows/extensions/pool.py
+20 -31
View File
@@ -15,17 +15,18 @@ class POOL_HEADER(objects.StructType):
def get_object(self,
type_name: str,
type_map: dict,
use_top_down: bool,
native_layer_name: Optional[str] = None,
object_type: Optional[str] = None,
cookie: Optional[int] = None) -> Optional[interfaces.objects.ObjectInterface]:
"""Carve an object or data structure from a kernel pool allocation.
executive: bool = False,
native_layer_name: Optional[str] = None) -> Optional[interfaces.objects.ObjectInterface]:
"""Carve an object or data structure from a kernel pool allocation
:param type_name: the data structure type name
:param native_layer_name: the name of the layer where the data originally lived
:param object_type: the object type (executive kernel objects only)
:return:
Args:
type_name: the data structure type name
native_layer_name: the name of the layer where the data originally lived
object_type: the object type (executive kernel objects only)
Returns:
An object as found from a POOL_HEADER
"""
symbol_table_name = self.vol.type_name.split(constants.BANG)[0]
@@ -36,7 +37,7 @@ class POOL_HEADER(objects.StructType):
pool_header_size = self.vol.size
# if there is no object type, then just instantiate a structure
if object_type is None:
if not executive:
mem_object = self._context.object(symbol_table_name + constants.BANG + type_name,
layer_name = self.vol.layer_name,
offset = self.vol.offset + pool_header_size,
@@ -52,6 +53,7 @@ class POOL_HEADER(objects.StructType):
# use the top down approach for windows 8 and later
if use_top_down:
body_offset = object_header_type.relative_child_offset('Body')
infomask_offset = object_header_type.relative_child_offset('InfoMask')
optional_headers, lengths_of_optional_headers = self._calculate_optional_header_lengths(
self._context, symbol_table_name)
@@ -112,21 +114,13 @@ class POOL_HEADER(objects.StructType):
continue
try:
mem_object = self._context.object(symbol_table_name + constants.BANG + type_name,
layer_name = self.vol.layer_name,
offset = addr + body_offset + start_offset,
native_layer_name = native_layer_name)
object_header = self._context.object(symbol_table_name + constants.BANG + "_OBJECT_HEADER",
layer_name = self.vol.layer_name,
offset = addr + start_offset,
native_layer_name = native_layer_name)
if not object_header.is_valid():
continue
object_type_string = object_header.get_object_type(type_map, cookie)
if object_type_string == object_type:
mem_object = object_header.Body.cast(symbol_table_name + constants.BANG + type_name)
if mem_object.is_valid():
return mem_object
if mem_object.is_valid():
return mem_object
except (TypeError, exceptions.InvalidAddressException):
pass
@@ -141,14 +135,9 @@ class POOL_HEADER(objects.StructType):
offset = self.vol.offset + self.BlockSize * alignment - rounded_size,
native_layer_name = native_layer_name)
object_header = mem_object.object_header()
try:
object_type_string = object_header.get_object_type(type_map, cookie)
if object_type_string == object_type:
if mem_object.is_valid():
return mem_object
else:
return None
except (TypeError, exceptions.InvalidAddressException):
return None
return None
@@ -181,7 +170,7 @@ class ExecutiveObject(interfaces.objects.ObjectInterface):
"""This is used as a "mixin" that provides all kernel executive objects
with a means of finding their own object header."""
def object_header(self) -> 'OBJECT_HEADER':
def get_object_header(self) -> 'OBJECT_HEADER':
if constants.BANG not in self.vol.type_name:
raise ValueError("Invalid symbol table name syntax (no {} found)".format(constants.BANG))
symbol_table_name = self.vol.type_name.split(constants.BANG)[0]