diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index e4df4b30a..50f931d6a 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -20,7 +20,7 @@ import enum import logging -from typing import Dict, Generator, List, Optional, Tuple +from typing import Dict, Generator, List, Optional, Tuple, Callable import volatility.plugins.windows.handles as handles @@ -72,6 +72,57 @@ class PoolConstraint: self.alignment = alignment +def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool], + fallback_checks: List[Tuple[str, Optional[str], bool]] + ) -> Callable[[interfaces.context.ContextInterface, str], bool]: + """Distinguishes a symbol table as being above a particular version or point + + This will primarily check the version metadata first and foremost. + If that metadata isn't available then each item in the fallback_checks is tested. + If invert is specified then the result will be true if the version is less than that specified, or in the case of + fallback, if any of the fallback checks is successful. + + A fallback check is made up of: + * a symbol or type name + * a member name (implying that the value before was a type name) + * whether that symbol, type or member must be present or absent for the symbol table to be more above the required point + + Note: Specifying that a member must not be present includes the whole type not being present too (ie, either will pass the test) + """ + + # try the primary method based on the pe version in the ISF + def method(context: interfaces.context.ContextInterface, symbol_table: str) -> bool: + + try: + pe_version = context.symbol_space[symbol_table].metadata.pe_version + major, minor, revision, build = pe_version + return version_check((major, minor, revision, build)) + except (AttributeError, ValueError, TypeError): + vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available") + + if not fallback_checks: + raise ValueError("No fallback methods for os_distinguishing provided") + + # fall back to the backup method, if necessary + for name, member, response in fallback_checks: + if member is None: + if (context.symbol_space.has_symbol(symbol_table + constants.BANG + name) + or context.symbol_space.has_type(symbol_table + constants.BANG + name)) != response: + return False + else: + try: + symbol_type = context.symbol_space.get_type(symbol_table + constants.BANG + name) + if symbol_type.has_member(member) != response: + return False + except exceptions.SymbolError: + if not response: + return False + + return True + + return method + + class PoolScanner(plugins.PluginInterface): """A generic pool scanner plugin""" @@ -83,61 +134,14 @@ class PoolScanner(plugins.PluginInterface): requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols") ] - @staticmethod - def is_windows_10(context: interfaces.context.ContextInterface, symbol_table: str) -> bool: - """Determine if the analyzed sample is Windows 10""" - - # try the primary method based on the pe version in the ISF - try: - pe_version = context.symbol_space[symbol_table].metadata.pe_version - major, minor, _revision, _build = pe_version - return (major, minor) >= (10, 0) - except (AttributeError, ValueError, TypeError): - vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available") - - # fall back to the backup method, if necessary - try: - _symbol = context.symbol_space.get_symbol(symbol_table + constants.BANG + "ObHeaderCookie") - return True - except exceptions.SymbolError: - return False - - @staticmethod - def is_windows_8_or_later(context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str) -> bool: - """Determine if the analyzed sample is Windows 8 or later""" - - # try the primary method based on the pe version in the ISF - try: - pe_version = context.symbol_space[symbol_table].metadata.pe_version - major, minor, _revision, _build = pe_version - return (major, minor) >= (6, 2) - except (AttributeError, ValueError, TypeError): - vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available") - - # fall back to the backup method, if necessary - kvo = context.memory[layer_name].config['kernel_virtual_offset'] - ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo) - handle_table_type = ntkrnlmp.get_type("_HANDLE_TABLE") - return not handle_table_type.has_member("HandleCount") - - @staticmethod - def is_windows_7(context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str) -> bool: - """Determine if the analyzed sample is Windows 7""" - - # try the primary method based on the pe version in the ISF - try: - pe_version = context.symbol_space[symbol_table].metadata.pe_version - major, minor, _revision, _build = pe_version - return (major, minor) == (6, 1) - except (AttributeError, ValueError): - vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available") - - # fall back to the backup method, if necessary - kvo = context.memory[layer_name].config['kernel_virtual_offset'] - ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo) - handle_table_type = ntkrnlmp.get_type("_OBJECT_HEADER") - return (handle_table_type.has_member("TypeIndex") - and not PoolScanner.is_windows_8_or_later(context, layer_name, symbol_table)) + is_windows_10 = os_distinguisher( + version_check = lambda x: x >= (10, 0), fallback_checks = [("ObHeaderCookie", None, True)]) + is_windows_8_or_later = os_distinguisher( + version_check = lambda x: x >= (6, 2), fallback_checks = [("_HANDLE_TABLE", "HandleCount", False)]) + # Technically, this is win7 or less + is_windows_7 = os_distinguisher( + version_check = lambda x: x == (6, 1), + fallback_checks = [("_OBJECT_HEADER", "TypeIndex", True), ("_HANDLE_TABLE", "HandleCount", True)]) def _generator(self): @@ -281,8 +285,7 @@ class PoolScanner(plugins.PluginInterface): cookie = handles.Handles.find_cookie(context = context, layer_name = layer_name, symbol_table = symbol_table) is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table) - is_windows_8_or_later = cls.is_windows_8_or_later( - context = context, layer_name = layer_name, symbol_table = symbol_table) + is_windows_8_or_later = cls.is_windows_8_or_later(context = context, symbol_table = symbol_table) # start off with the primary virtual layer scan_layer = layer_name @@ -332,7 +335,7 @@ class PoolScanner(plugins.PluginInterface): # We have to manually load a symbol table if symbols.symbol_table_is_64bit(context, symbol_table): - is_win_7 = PoolScanner.is_windows_7(context, 'primary', symbol_table) + is_win_7 = cls.is_windows_7(context = context, symbol_table = symbol_table) if is_win_7: pool_header_json_filename = "poolheader-x64-win7" else: