diff --git a/volatility/framework/plugins/windows/bigpools.py b/volatility/framework/plugins/windows/bigpools.py index 70351e5ce..e115b45a0 100644 --- a/volatility/framework/plugins/windows/bigpools.py +++ b/volatility/framework/plugins/windows/bigpools.py @@ -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([ diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 7cc58dba4..708e45694 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -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: diff --git a/volatility/framework/plugins/windows/registry/hivelist.py b/volatility/framework/plugins/windows/registry/hivelist.py index e9de1da04..1f1080e15 100644 --- a/volatility/framework/plugins/windows/registry/hivelist.py +++ b/volatility/framework/plugins/windows/registry/hivelist.py @@ -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()) \ No newline at end of file + return renderers.TreeGrid([("Offset", format_hints.Hex), ("FileFullPath", str)], self._generator()) diff --git a/volatility/framework/plugins/windows/registry/hivescan.py b/volatility/framework/plugins/windows/registry/hivescan.py index bbcdad104..9e5d3074f 100644 --- a/volatility/framework/plugins/windows/registry/hivescan.py +++ b/volatility/framework/plugins/windows/registry/hivescan.py @@ -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: diff --git a/volatility/framework/symbols/windows/extensions/pool.py b/volatility/framework/symbols/windows/extensions/pool.py index 464e361bf..e4ae00e0e 100644 --- a/volatility/framework/symbols/windows/extensions/pool.py +++ b/volatility/framework/symbols/windows/extensions/pool.py @@ -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]: