From 037ac2853407284179b7d367f878d30a68ec609f Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 14 May 2018 01:01:29 +0100 Subject: [PATCH] Move around the scanning functions and create an API for the iterators. --- volatility/framework/interfaces/layers.py | 84 ++++++++++++++++------- volatility/framework/layers/intel.py | 46 ------------- 2 files changed, 58 insertions(+), 72 deletions(-) diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 304f846ae..1f655b0d1 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -24,6 +24,7 @@ except ImportError: pass ProgressValue = typing.Union['DummyProgress', multiprocessing.Value] +IteratorValue = typing.Tuple[typing.List[typing.Tuple[str, int, int]], int] class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): @@ -171,7 +172,9 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR scanner: ScannerInterface, progress_callback: validity.ProgressCallback = None, min_address: typing.Optional[int] = None, - max_address: typing.Optional[int] = None) -> typing.Iterable[typing.Any]: + max_address: typing.Optional[int] = None, + scan_iterator: typing.Optional[typing.Callable[['ScannerInterface', int, int], IteratorValue]] = None) -> \ + typing.Iterable[typing.Any]: """Scans a Translation layer by chunk Note: this will skip missing/unmappable chunks of memory @@ -179,6 +182,9 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR if progress_callback is not None and not callable(progress_callback): raise TypeError("Progress_callback is not callable") + if scan_iterator is None: + scan_iterator = self._scan_iterator + scanner = self._check_type(scanner, ScannerInterface) scanner.context = context scanner.layer_name = self.name @@ -197,7 +203,7 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR try: progress = DummyProgress() # type: ProgressValue - scan_iterator = functools.partial(self._scan_iterator, scanner, min_address, max_address) + scan_iterator = functools.partial(scan_iterator, scanner, min_address, max_address) scan_metric = functools.partial(self._scan_metric, scanner, min_address, max_address) if scanner.thread_safe and not constants.DISABLE_MULTITHREADED_SCANNING: progress = multiprocessing.Manager().Value("Q", 0) @@ -231,27 +237,51 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR def _scan_iterator(self, scanner: 'ScannerInterface', min_address: int, - max_address: int) -> typing.Iterable[typing.Any]: - return range(min_address, max_address, scanner.chunk_size) + max_address: int) \ + -> typing.Iterable[IteratorValue]: + """Iterator that indicates which blocks in the layer are to be read by for the scanning + Returns a list of blocks (potentially in lower layers) that make up this chunk contiguously. + Chunks can be no bigger than scanner.chunk_size + scanner.overlap + DataLayers by default are assumed to have no holes + """ + offset, mapped_offset, length, layer_name = min_address, min_address, max_address - min_address, self.name + while length > 0: + chunk_size = min(length, scanner.chunk_size + scanner.overlap) + yield [(layer_name, mapped_offset, chunk_size)], offset + chunk_size + # It we've got more than the scanner's chunk_size, only move up by the chunk_size + if chunk_size > scanner.chunk_size: + chunk_size -= scanner.overlap + length -= chunk_size + mapped_offset += chunk_size + offset += chunk_size + + # We ignore the type due to the iterator_value, actually it only needs to match the output from _scan_iterator def _scan_chunk(self, scanner: 'ScannerInterface', min_address: int, max_address: int, - progress: ProgressValue, - iterator_value: typing.Any) -> typing.List[typing.Any]: - length = min(scanner.chunk_size + scanner.overlap, max_address - iterator_value) - chunk = self.read(iterator_value, length) - # Don't include the overlaps, or we'll go over 100% - progress.value += min(scanner.chunk_size, max_address - iterator_value) - return list(scanner(chunk, iterator_value)) + progress: 'ProgressValue', + iterator_value: IteratorValue) -> typing.List[typing.Any]: + data_to_scan, chunk_end = iterator_value + data = b'' + for layer_name, address, chunk_size in data_to_scan: + try: + data += self.context.memory[layer_name].read(address, chunk_size) + except exceptions.InvalidAddressException: + vollog.debug( + "Invalid address in layer {} found scanning {} at address {:x}".format(layer_name, self.name, + address)) + + progress.value = chunk_end + return list(scanner(data, chunk_end - len(data))) def _scan_metric(self, _scanner: 'ScannerInterface', min_address: int, max_address: int, value: int) -> float: - return max(0, (value * 100) / (max_address - min_address)) + return max(0, ((value - min_address) * 100) / (max_address - min_address)) def build_configuration(self) -> interfaces.configuration.HierarchicalDict: config = super().build_configuration() @@ -344,20 +374,22 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): # ## Scan implementation with knowledge of pages - def _scan_chunk(self, - scanner: 'interfaces.layers.ScannerInterface', - min_address: int, - max_address: int, - progress: ProgressValue, - iterator_value: typing.Any) -> typing.List[typing.Any]: - size_to_scan = min(max_address - min_address, scanner.chunk_size + scanner.overlap) - result = [] # type: typing.List[typing.Any] - for map in self.mapping(iterator_value, size_to_scan, ignore_errors = True): - offset, mapped_offset, length, layer = map - progress.value += length - chunk = self._context.memory.read(layer, mapped_offset, length) - result += [x for x in scanner(chunk, offset)] - return result + def _scan_iterator(self, + scanner: 'ScannerInterface', + min_address: int, + max_address: int) \ + -> typing.Iterable[IteratorValue]: + for mapped in self.mapping(min_address, max_address - min_address, ignore_errors = True): + offset, mapped_offset, length, layer_name = mapped + while length > 0: + chunk_size = min(length, scanner.chunk_size + scanner.overlap) + yield [(layer_name, mapped_offset, chunk_size)], offset + chunk_size + # It we've got more than the scanner's chunk_size, only move up by the chunk_size + if chunk_size > scanner.chunk_size: + chunk_size -= scanner.overlap + length -= chunk_size + mapped_offset += chunk_size + offset += chunk_size class Memory(validity.ValidityRoutines, collections.abc.Mapping): diff --git a/volatility/framework/layers/intel.py b/volatility/framework/layers/intel.py index a03c2c524..a61328d76 100644 --- a/volatility/framework/layers/intel.py +++ b/volatility/framework/layers/intel.py @@ -9,8 +9,6 @@ from volatility.framework.configuration import requirements vollog = logging.getLogger(__name__) -IteratorValue = typing.Tuple[typing.List[typing.Tuple[str, int, int]], int] - class classproperty(object): """Class property decorator @@ -196,50 +194,6 @@ class Intel(interfaces.layers.TranslationLayerInterface): requirements.StringRequirement(name = 'linux_banner', optional = True)] - def _scan_iterator(self, - scanner: interfaces.layers.ScannerInterface, - min_address: int, - max_address: int) \ - -> typing.Iterable[IteratorValue]: - for mapped in self.mapping(min_address, max_address - min_address, ignore_errors = True): - offset, mapped_offset, length, layer_name = mapped - while length > 0: - chunk_size = min(length, scanner.chunk_size + scanner.overlap) - yield [(layer_name, mapped_offset, chunk_size)], offset + chunk_size - # It we've got more than the scanner's chunk_size, only move up by the chunk_size - if chunk_size > scanner.chunk_size: - chunk_size -= scanner.overlap - length -= chunk_size - mapped_offset += chunk_size - offset += chunk_size - - # We ignore the type due to the iterator_value, actually it only needs to match the output from _scan_iterator - def _scan_chunk(self, - scanner: interfaces.layers.ScannerInterface, - min_address: int, - max_address: int, - progress: interfaces.layers.ProgressValue, - iterator_value: IteratorValue) -> typing.List[typing.Any]: - data_to_scan, chunk_end = iterator_value - data = b'' - for layer_name, address, chunk_size in data_to_scan: - try: - data += self.context.memory[layer_name].read(address, chunk_size) - except exceptions.InvalidAddressException: - vollog.debug( - "Invalid address in layer {} found scanning {} at address {:x}".format(layer_name, self.name, - address)) - - progress.value = chunk_end - return list(scanner(data, chunk_end - len(data))) - - def _scan_metric(self, - _scanner: interfaces.layers.ScannerInterface, - min_address: int, - max_address: int, - value: int) -> float: - return max(0, ((value - min_address) * 100) / (max_address - min_address)) - class IntelPAE(Intel): """Class for handling Physical Address Extensions for Intel architectures"""