Issue #186: Rebase and apply yapf

This commit is contained in:
Mike Auty
2020-05-27 23:16:18 +01:00
committed by ikelos
parent 36c19735ee
commit a0da5fef83
5 changed files with 49 additions and 51 deletions
@@ -24,9 +24,9 @@ class BigPools(interfaces.plugins.PluginInterface):
is_vista_or_later = poolscanner.os_distinguisher(version_check = lambda x: x >= (6, 0),
fallback_checks = [("KdCopyDataBlock", None, True)])
is_win10 = poolscanner.os_distinguisher(version_check=lambda x: (10, 0) <= x,
fallback_checks=[("ObHeaderCookie", None, True),
("_HANDLE_TABLE", "HandleCount", False)])
is_win10 = poolscanner.os_distinguisher(version_check = lambda x: (10, 0) <= x,
fallback_checks = [("ObHeaderCookie", None, True),
("_HANDLE_TABLE", "HandleCount", False)])
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -36,18 +36,18 @@ class BigPools(interfaces.plugins.PluginInterface):
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.StringRequirement(name='tags',
description="Comma separated list of pool tags to filter pools returned",
optional=True,
default=None)
requirements.StringRequirement(name = 'tags',
description = "Comma separated list of pool tags to filter pools returned",
optional = True,
default = None)
]
@classmethod
def list_big_pools(cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
tags: Optional[list] = None):
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
tags: Optional[list] = None):
"""Returns the big page pool objects from the kernel PoolBigPageTable array.
Args:
@@ -60,15 +60,13 @@ class BigPools(interfaces.plugins.PluginInterface):
A big page pool object
"""
kvo = context.layers[layer_name].config['kernel_virtual_offset']
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
big_page_table_offset = ntkrnlmp.get_symbol("PoolBigPageTable").address
big_page_table = ntkrnlmp.object(object_type="unsigned long long",
offset=big_page_table_offset)
big_page_table = ntkrnlmp.object(object_type = "unsigned long long", offset = big_page_table_offset)
big_page_table_size_offset = ntkrnlmp.get_symbol("PoolBigPageTableSize").address
big_page_table_size = ntkrnlmp.object(object_type="unsigned long",
offset=big_page_table_size_offset)
big_page_table_size = ntkrnlmp.object(object_type = "unsigned long", offset = big_page_table_size_offset)
try:
big_page_table_type = ntkrnlmp.get_type("_POOL_TRACKER_BIG_PAGED")
@@ -89,27 +87,27 @@ class BigPools(interfaces.plugins.PluginInterface):
big_pools_json_filename += "-x86"
new_table_name = intermed.IntermediateSymbolTable.create(
context=context,
config_path=configuration.path_join(context.symbol_space[symbol_table].config_path, "bigpools"),
sub_path="windows",
filename=big_pools_json_filename,
table_mapping={'nt_symbols': symbol_table},
class_types={'_POOL_TRACKER_BIG_PAGES': extensions.pool.POOL_TRACKER_BIG_PAGES})
module = context.module(new_table_name, layer_name, offset=0)
context = context,
config_path = configuration.path_join(context.symbol_space[symbol_table].config_path, "bigpools"),
sub_path = "windows",
filename = big_pools_json_filename,
table_mapping = {'nt_symbols': symbol_table},
class_types = {'_POOL_TRACKER_BIG_PAGES': extensions.pool.POOL_TRACKER_BIG_PAGES})
module = context.module(new_table_name, layer_name, offset = 0)
big_page_table_type = module.get_type("_POOL_TRACKER_BIG_PAGES")
big_pools = ntkrnlmp.object(object_type="array",
offset=big_page_table,
subtype=big_page_table_type,
count=big_page_table_size,
absolute=True)
big_pools = ntkrnlmp.object(object_type = "array",
offset = big_page_table,
subtype = big_page_table_type,
count = big_page_table_size,
absolute = True)
for big_pool in big_pools:
if big_pool.is_valid():
if tags is None or big_pool.get_key() in tags:
yield big_pool
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]: #, str, int]]]:
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]: #, str, int]]]:
if self.config.get("tags"):
tags = [tag for tag in self.config["tags"].split(',')]
else:
@@ -124,10 +122,7 @@ class BigPools(interfaces.plugins.PluginInterface):
if not isinstance(num_bytes, interfaces.renderers.BaseAbsentValue):
num_bytes = format_hints.Hex(num_bytes)
yield (0, (format_hints.Hex(big_pool.Va),
big_pool.get_key(),
big_pool.get_pool_type(),
num_bytes))
yield (0, (format_hints.Hex(big_pool.Va), big_pool.get_key(), big_pool.get_pool_type(), num_bytes))
def run(self):
return renderers.TreeGrid([
@@ -318,7 +318,7 @@ class PoolScanner(plugins.PluginInterface):
type_name = symbol_table + constants.BANG + "_CMHIVE",
size = (800, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE,
skip_type_test=True),
skip_type_test = True),
]
if not tags_filter:
@@ -16,6 +16,7 @@ vollog = logging.getLogger(__name__)
class HiveGenerator():
"""Walks the registry HiveList linked list in a given direction and stores an invalid offset
if it's unable to fully walk the list"""
def __init__(self, cmhive, forward = True):
self.cmhive = cmhive
self.forward = forward
@@ -45,7 +46,7 @@ class HiveList(interfaces.plugins.PluginInterface):
description = "String to filter hive names returned",
optional = True,
default = None),
requirements.PluginRequirement(name='hivescan', plugin=hivescan.HiveScan, version=(1, 0, 0)),
requirements.PluginRequirement(name = 'hivescan', plugin = hivescan.HiveScan, version = (1, 0, 0)),
]
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]:
@@ -133,7 +134,7 @@ class HiveList(interfaces.plugins.PluginInterface):
# Run through the list forwards
seen = set()
hg = HiveGenerator(cmhive, forward=True)
hg = HiveGenerator(cmhive, forward = True)
for hive in hg:
if hive.vol.offset in seen:
vollog.debug("Hivelist found an already seen offset {} while "\
@@ -146,8 +147,9 @@ class HiveList(interfaces.plugins.PluginInterface):
forward_invalid = hg.invalid
if forward_invalid:
vollog.debug("Hivelist failed traversing the list forwards at {}, traversing backwards".format(hex(forward_invalid)))
hg = HiveGenerator(cmhive, forward=False)
vollog.debug("Hivelist failed traversing the list forwards at {}, traversing backwards".format(
hex(forward_invalid)))
hg = HiveGenerator(cmhive, forward = False)
for hive in hg:
if hive.vol.offset in seen:
vollog.debug("Hivelist found an already seen offset {} while "\
@@ -175,18 +177,21 @@ class HiveList(interfaces.plugins.PluginInterface):
start_hive_offset = hive.HiveList.Flink - reloff
## Now instantiate the first hive in virtual address space as normal
start_hive = ntkrnlmp.object(object_type="_CMHIVE", offset=start_hive_offset,
absolute=True)
start_hive = ntkrnlmp.object(object_type = "_CMHIVE",
offset = start_hive_offset,
absolute = True)
for forward in (True, False):
for linked_hive in start_hive.HiveList.to_list(hive.vol.type_name, "HiveList", forward):
if not linked_hive.is_valid() or linked_hive.vol.offset in seen:
continue
seen.add(linked_hive.vol.offset)
if filter_string is None or filter_string.lower() in str(linked_hive.get_name() or "").lower():
if filter_string is None or filter_string.lower() in str(linked_hive.get_name()
or "").lower():
if context.layers[layer_name].is_valid(linked_hive.vol.offset):
yield linked_hive
except exceptions.InvalidAddressException:
vollog.debug("InvalidAddressException when traversing hive {} found from scan, skipping".format(hex(hive.vol.offset)))
vollog.debug("InvalidAddressException when traversing hive {} found from scan, skipping".format(
hex(hive.vol.offset)))
def run(self) -> renderers.TreeGrid:
return renderers.TreeGrid([("Offset", format_hints.Hex), ("FileFullPath", str)], self._generator())
return renderers.TreeGrid([("Offset", format_hints.Hex), ("FileFullPath", str)], self._generator())
@@ -52,13 +52,13 @@ class HiveScan(interfaces.plugins.PluginInterface):
if is_windows_8_1_or_later and is_64bit:
kvo = context.layers[layer_name].config['kernel_virtual_offset']
ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
for pool in bigpools.BigPools.list_big_pools(context,
layer_name=layer_name,
symbol_table=symbol_table,
tags=["CM10"]):
cmhive = ntkrnlmp.object(object_type="_CMHIVE", offset=pool.Va, absolute=True)
layer_name = layer_name,
symbol_table = symbol_table,
tags = ["CM10"]):
cmhive = ntkrnlmp.object(object_type = "_CMHIVE", offset = pool.Va, absolute = True)
yield cmhive
else:
@@ -186,9 +186,7 @@ class POOL_TRACKER_BIG_PAGES(objects.StructType):
def get_key(self) -> str:
"""Returns the Key value as a 4 character string"""
tag_bytes = objects.convert_value_to_data(self.Key,
int,
objects.DataFormatInfo(4, "little", False))
tag_bytes = objects.convert_value_to_data(self.Key, int, objects.DataFormatInfo(4, "little", False))
return "".join([chr(x) if 32 < x < 127 else '' for x in tag_bytes])
def get_pool_type(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: