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

Refactor poolscanner layer to move more into the parallel scanner.

Browse Source
This commit is contained in:
Mike Auty
2019-08-07 18:37:56 +01:00
parent 9ed1a5a5b6
commit 253ce69fc9
1 changed files with 71 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility/framework/plugins/windows/poolscanner.py
+71 -53
View File
@@ -72,6 +72,62 @@ class PoolConstraint:
self.alignment = alignment
class PoolHeaderScanner(interfaces.layers.ScannerInterface):
def __init__(self, module: interfaces.context.ModuleInterface, constraint_lookup: Dict[bytes, PoolConstraint],
alignment: int):
super().__init__()
self._module = module
self._constraint_lookup = constraint_lookup
self._alignment = alignment
header_type = self._module.get_type('_POOL_HEADER')
self._header_offset = header_type.relative_child_offset('PoolTag')
self._subscanner = scanners.MultiStringScanner([c for c in constraint_lookup.keys()])
def __call__(self, data: bytes, data_offset: int):
for offset, pattern in self._subscanner(data, data_offset):
header = self._module.object(type_name = "_POOL_HEADER", offset = offset - self._header_offset)
constraint = self._constraint_lookup[pattern]
try:
# Size check
if constraint.size is not None:
if constraint.size[0]:
if (self._alignment * header.BlockSize) < constraint.size[0]:
continue
if constraint.size[1]:
if (self._alignment * header.BlockSize) > constraint.size[1]:
continue
# Type check
if constraint.page_type is not None:
checks_pass = False
if (constraint.page_type & PoolType.FREE) and header.PoolType == 0:
checks_pass = True
elif (constraint.page_type & PoolType.PAGED) and header.PoolType % 2 == 0 and header.PoolType > 0:
checks_pass = True
elif (constraint.page_type & PoolType.NONPAGED) and header.PoolType % 2 == 1:
checks_pass = True
if not checks_pass:
continue
if constraint.index is not None:
if constraint.index[0]:
if header.index < constraint.index[0]:
continue
if constraint.index[1]:
if header.index > constraint.index[1]:
continue
except exceptions.InvalidAddressException:
# The tested object's header doesn't point to valid addresses, ignore it
continue
# We found one that passed!
yield (constraint, header)
def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
fallback_checks: List[Tuple[str, Optional[str], bool]]
) -> Callable[[interfaces.context.ContextInterface, str], bool]:
@@ -322,15 +378,24 @@ class PoolScanner(plugins.PluginInterface):
"""Returns the _POOL_HEADER object (based on the symbol_table template) after scanning through layer_name
returning all headers that match any of the constraints provided. Only one constraint can be provided per tag"""
# Setup the pattern
constraint_lookup = {} # type: Dict[bytes, List[PoolConstraint]]
constraint_lookup = {} # type: Dict[bytes, PoolConstraint]
for constraint in pool_constraints:
temp_list = constraint_lookup.get(constraint.tag, [])
temp_list.append(constraint)
constraint_lookup[constraint.tag] = temp_list
if constraint.tag in constraint_lookup:
raise ValueError("Constraint tag is used for more than one constraint: {}".format(constraint.tag))
constraint_lookup[constraint.tag] = constraint
module = cls._get_pool_header_module(context, layer_name, symbol_table)
# Run the scan locating the offsets of a particular tag
layer = context.layers[layer_name]
scanner = PoolHeaderScanner(module, constraint_lookup, alignment)
yield from layer.scan(context, scanner, progress_callback)
@classmethod
def _get_pool_header_module(cls, context, layer_name, symbol_table):
# Setup the pool header and offset differential
try:
module = context.module(symbol_table, layer_name, offset = 0)
header_type = module.get_type('_POOL_HEADER')
except exceptions.SymbolError:
# We have to manually load a symbol table
@@ -350,54 +415,7 @@ class PoolScanner(plugins.PluginInterface):
filename = pool_header_json_filename,
table_mapping = {'nt_symbols': symbol_table})
module = context.module(new_table_name, layer_name, offset = 0)
header_type = module.get_type('_POOL_HEADER')
header_offset = header_type.relative_child_offset('PoolTag')
# Run the scan locating the offsets of a particular tag
layer = context.layers[layer_name]
scanner = scanners.MultiStringScanner([c for c in constraint_lookup.keys()])
for offset, pattern in layer.scan(context, scanner, progress_callback = progress_callback):
header = module.object(type_name = "_POOL_HEADER", offset = offset - header_offset)
for constraint in constraint_lookup[pattern]:
# Size check
try:
if constraint.size is not None:
if constraint.size[0]:
if (alignment * header.BlockSize) < constraint.size[0]:
continue
if constraint.size[1]:
if (alignment * header.BlockSize) > constraint.size[1]:
continue
# Type check
if constraint.page_type is not None:
checks_pass = False
if (constraint.page_type & PoolType.FREE) and header.PoolType == 0:
checks_pass = True
elif (constraint.page_type &
PoolType.PAGED) and header.PoolType % 2 == 0 and header.PoolType > 0:
checks_pass = True
elif (constraint.page_type & PoolType.NONPAGED) and header.PoolType % 2 == 1:
checks_pass = True
if not checks_pass:
continue
if constraint.index is not None:
if constraint.index[0]:
if header.index < constraint.index[0]:
continue
if constraint.index[1]:
if header.index > constraint.index[1]:
continue
except exceptions.InvalidAddressException:
# The tested object's header doesn't point to valid addresses, ignore it
continue
# We found one that passed!
yield (constraint, header)
return module
def run(self) -> renderers.TreeGrid:
return renderers.TreeGrid([("Tag", str), ("Offset", format_hints.Hex), ("Layer", str), ("Name", str)],