diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 9a4c856e3..aba7f6fd2 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -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)],