diff --git a/volatility/framework/automagic/linux.py b/volatility/framework/automagic/linux.py index 2df88daef..f4c042591 100644 --- a/volatility/framework/automagic/linux.py +++ b/volatility/framework/automagic/linux.py @@ -24,7 +24,7 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface): @property def _linux_banners(self) -> linux_symbol_cache.LinuxBanners: """Creates a cached copy of the results, but only it's been requested""" - if self._linux_banners_ is None: + if not self._linux_banners_: self._linux_banners_ = linux_symbol_cache.LinuxSymbolCache.load_linux_banners() return self._linux_banners_ @@ -107,7 +107,7 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): context: interfaces.context.ContextInterface, layer_name: str, progress_callback: validity.ProgressCallback = None) \ - -> typing.Union[None, typing.Type[interfaces.layers.DataLayerInterface]]: + -> typing.Optional[interfaces.layers.DataLayerInterface]: """Attempts to identify linux within this layer""" layer = context.memory[layer_name] join = interfaces.configuration.path_join @@ -166,7 +166,7 @@ class LinuxUtilities(object): symbol_table: str, layer_name: str, progress_callback: validity.ProgressCallback = None) \ - -> typing.Tuple[typing.Union[None, int], typing.Union[None, int]]: + -> typing.Tuple[typing.Optional[int], typing.Optional[int]]: """Determines the offset of the actual DTB in physical space and its symbol offset""" init_task_symbol = symbol_table + constants.BANG + 'init_task' table_dtb = context.symbol_space.get_symbol(init_task_symbol).address diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index e6947cb8a..4a99f1b59 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -67,8 +67,8 @@ def scan(ctx: interfaces.context.ContextInterface, layer_name: str, page_size: int, progress_callback: validity.ProgressCallback = None, - start: typing.Union[int, None] = None, - end: typing.Union[int, None] = None) \ + start: typing.Optional[int] = None, + end: typing.Optional[int] = None) \ -> typing.Generator[typing.Dict[str, typing.Union[bytes, str, int]], None, None]: """Scans through `layer_name` at `ctx` looking for RSDS headers that indicate one of four common pdb kernel names (as listed in `self.pdb_names`) and returns the tuple (GUID, age, pdb_name, signature_offset, mz_offset) @@ -143,7 +143,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): config_path: str, requirement: interfaces.configuration.RequirementInterface, progress_callback: validity.ProgressCallback = None) \ - -> typing.Dict[bytes, typing.Dict]: + -> typing.Dict[str, typing.Iterable]: """Traverses the requirement tree, rooted at `requirement` looking for virtual layers that might contain a windows PDB. Returns a list of possible kernel locations in the physical memory @@ -157,7 +157,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): :return: A list of (layer_name, scan_results) """ sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) - results: typing.Dict[bytes, typing.Dict] = {} + results: typing.Dict[str, typing.Iterable] = {} if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement): # Check for symbols in this layer # FIXME: optionally allow a full (slow) scan @@ -317,7 +317,11 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): vollog.info("No suitable kernels found during pdbscan") return valid_kernels - def __call__(self, context, config_path, requirement, progress_callback = None): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.ConstructableRequirementInterface, + progress_callback: validity.ProgressCallback = None) -> None: # TODO: Check if we really need to search for pdbs if requirement.unsatisfied(context, config_path): if "pdbscan" not in context.symbol_space: diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index 978b17362..b5bf53434 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -113,7 +113,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): stacked_layers: typing.List, requirement: interfaces.configuration.RequirementInterface, context: interfaces.context.ContextInterface, - config_path: str) -> typing.Union[None, typing.Tuple[str, str]]: + config_path: str) -> typing.Optional[typing.Tuple[str, str]]: """Looks for translation layer requirements and attempts to apply the stacked layers to it. If it succeeds it returns the configuration path and layer name where the stacked nodes were spliced into the tree. diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index 54a751474..4601a988f 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -24,6 +24,7 @@ The self-referential indices for older versions of windows are listed below: """ import logging import struct +import typing from volatility.framework import interfaces, layers, validity from volatility.framework.configuration import requirements @@ -39,7 +40,11 @@ class DtbTest(validity.ValidityRoutines): and determine whether it points back to that page's offset. """ - def __init__(self, layer_type = None, ptr_struct = None, ptr_reference = None, mask = None): + def __init__(self, + layer_type: typing.Type[layers.intel.Intel] = None, + ptr_struct: str = None, + ptr_reference: int = None, + mask: int = None) -> None: self.layer_type = self._check_class(layer_type, layers.intel.Intel) self.ptr_struct = self._check_type(ptr_struct, str) self.ptr_size = struct.calcsize(ptr_struct) @@ -47,10 +52,13 @@ class DtbTest(validity.ValidityRoutines): self.mask = self._check_type(mask, int) self.page_size = layer_type.page_size - def _unpack(self, value): + def _unpack(self, value: bytes) -> int: return struct.unpack("<" + self.ptr_struct, value)[0] - def __call__(self, data, data_offset, page_offset): + def __call__(self, + data: bytes, + data_offset: int, + page_offset: int) -> typing.Optional[typing.Tuple[int, typing.Any]]: """Tests a specific page in a chunk of data to see if it contains a self-referential pointer. :param data: The chunk of data that contains the page to be scanned @@ -59,10 +67,10 @@ class DtbTest(validity.ValidityRoutines): :type data_offset: int :param page_offset: Where, within the data, the page to be scanned starts :type page_offset: int - :return: A valid DTB within this page + :return: A valid DTB within this page (and an additional parameter for data) """ value = data[page_offset + (self.ptr_reference * self.ptr_size):page_offset + ( - (self.ptr_reference + 1) * self.ptr_size)] + (self.ptr_reference + 1) * self.ptr_size)] ptr = self._unpack(value) # The value *must* be present (bit 0) since it's a mapped page # It's almost always writable (bit 1) @@ -72,8 +80,9 @@ class DtbTest(validity.ValidityRoutines): if ptr != 0 and (ptr & self.mask == data_offset + page_offset) & (ptr & 0xFF1 == 0x61): dtb = (ptr & self.mask) return self.second_pass(dtb, data, data_offset) + return None - def second_pass(self, dtb, data, data_offset): + def second_pass(self, dtb: int, data: bytes, data_offset: int) -> typing.Optional[typing.Tuple[int, typing.Any]]: """Re-reads over the whole page to validate other records based on the number of pages marked user vs super :param dtb: The identified dtb that needs validating @@ -95,7 +104,8 @@ class DtbTest(validity.ValidityRoutines): # We sometimes find bogus DTBs at 0x16000 with a very low sup_count and 0 usr_count # I have a winxpsp2-x64 image with identical usr/sup counts at 0x16000 and 0x24c00 as well as the actual 0x3c3000 if usr_count or sup_count > 5: - return dtb + return dtb, None + return None class DtbTest32bit(DtbTest): @@ -121,7 +131,7 @@ class DtbTestPae(DtbTest): ptr_reference = 0x3, mask = 0x3FFFFFFFFFF000) - def second_pass(self, dtb, data, data_offset): + def second_pass(self, dtb: int, data: bytes, data_offset: int) -> typing.Optional[typing.Tuple[int, typing.Any]]: """PAE top level directory tables contains four entries and the self-referential pointer occurs in the second level of tables (so as not to use up a full quarter of the space). This is very high in the space, and occurs in the fourht (last quarter) second-level table. The second-level tables appear always to come sequentially @@ -143,22 +153,28 @@ class DtbTestPae(DtbTest): pointers = data[dtb - data_offset + (3 * self.ptr_size): dtb - data_offset + (4 * self.ptr_size)] val = self._unpack(pointers) if (val & self.mask == dtb + 0x4000) and (val & 0xFFF == 0x001): - return dtb + return dtb, None + return None class DtbSelfReferential(DtbTest): """A generic DTB test which looks for a self-referential pointer at *any* index within the page.""" - def __init__(self, layer_type, ptr_struct, ptr_reference, mask): + def __init__(self, + layer_type: typing.Type[layers.intel.Intel], + ptr_struct: str, + ptr_reference: int, + mask: int) -> None: super().__init__(layer_type = layer_type, ptr_struct = ptr_struct, ptr_reference = ptr_reference, mask = mask) - def __call__(self, data, data_offset, page_offset): + def __call__(self, data: bytes, data_offset: int, page_offset: int) \ + -> typing.Optional[typing.Tuple[int, int]]: page = data[page_offset:page_offset + self.page_size] if not page: - return + return None ref_pages = set() for ref in range(0, self.page_size, self.ptr_size): ptr_data = page[ref:ref + self.ptr_size] @@ -168,7 +184,8 @@ class DtbSelfReferential(DtbTest): ref_pages.add(ref) # The DTB is extremely unlikely to refer back to itself. so the number of reference should always be exactly 1 if len(ref_pages) == 1: - return (data_offset + page_offset), ref_pages + return (data_offset + page_offset), ref_pages.pop() + return None class DtbSelfRef32bit(DtbSelfReferential): @@ -187,17 +204,18 @@ class PageMapScanner(interfaces.layers.ScannerInterface): """Scans through all pages using DTB tests to determine a dtb offset and architecture""" overlap = 0x4000 thread_safe = True - tests = [DtbTest32bit, DtbTest64bit, DtbTestPae] + tests = [DtbTest32bit(), DtbTest64bit(), DtbTestPae()] """The default tests to run when searching for DTBs""" - def __init__(self, tests): + def __init__(self, tests: typing.List[DtbTest]) -> None: super().__init__() for value in tests: self._check_type(value, DtbTest) self.tests = tests - def __call__(self, data, data_offset): - results = {} + def __call__(self, data: bytes, data_offset: int) \ + -> typing.Generator[typing.Tuple[DtbTest, typing.Set[int]], None, None]: + results: typing.Dict[DtbTest, typing.Set[int]] = {} for test in self.tests: results[test] = set() @@ -205,7 +223,7 @@ class PageMapScanner(interfaces.layers.ScannerInterface): for page_offset in range(0, len(data), 0x1000): result = test(data, data_offset, page_offset) if result is not None: - yield (test, result) + yield (test, result[0]) class WintelHelper(interfaces.automagic.AutomagicInterface): @@ -219,7 +237,12 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): priority = 20 tests = [DtbTest32bit(), DtbTest64bit(), DtbTestPae()] - def __call__(self, context, config_path, requirement, progress_callback = None): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.RequirementInterface, + progress_callback: validity.ProgressCallback = None) \ + -> None: useful = [] sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) if (isinstance(requirement, requirements.TranslationLayerRequirement) and @@ -243,8 +266,9 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): context.config[interfaces.configuration.path_join(sub_config_path, "page_map_offset")] = dtb break else: - return - requirement.construct(context, config_path) + return None + if isinstance(requirement, interfaces.configuration.ConstructableRequirementInterface): + requirement.construct(context, config_path) else: for subreq in requirement.requirements.values(): self(context, sub_config_path, subreq) @@ -252,7 +276,10 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): class WintelStacker(interfaces.automagic.StackerLayerInterface): @classmethod - def stack(cls, context, layer_name, progress_callback = None): + def stack(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + progress_callback: validity.ProgressCallback = None) -> typing.Optional[str]: """Attempts to determine and stack an intel layer on a physical layer where possible Where the DTB scan fails, it attempts a heuristic of checking for the DTB within a specific range. diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index 39a4a2fa0..bd36b0ff1 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -24,7 +24,9 @@ class MultiRequirement(interfaces_configuration.RequirementInterface): Technically the Interface could handle this, but it's an interface, so this is a concrete implementation. """ - def unsatisfied(self, context, config_path): + def unsatisfied(self, + context: interfaces_configuration.ContextInterface, + config_path: str) -> typing.List[str]: return self.unsatisfied_children(context, config_path) diff --git a/volatility/framework/contexts/__init__.py b/volatility/framework/contexts/__init__.py index 618e849ec..05feb7c82 100644 --- a/volatility/framework/contexts/__init__.py +++ b/volatility/framework/contexts/__init__.py @@ -3,6 +3,7 @@ This has been made an object to allow quick swapping and changing of contexts, to allow a plugin to act on multiple different contexts without them interfering eith each other. """ +import typing from volatility.framework import constants, interfaces, symbols @@ -36,30 +37,30 @@ class Context(interfaces.context.ContextInterface): # ## Symbol Space Functions @property - def config(self): + def config(self) -> interfaces.configuration.HierarchicalDict: """Returns a mutable copy of the configuration, but does not allow the whole configuration to be altered""" return self._config @config.setter - def config(self, value): + def config(self, value: interfaces.configuration.HierarchicalDict) -> None: if not isinstance(value, interfaces.configuration.HierarchicalDict): raise TypeError("Config must be of type HierarchicalDict") self._config = value @property - def symbol_space(self): + def symbol_space(self) -> interfaces.symbols.SymbolSpaceInterface: """The space of all symbols that can be accessed within this context. """ return self._symbol_space @property - def memory(self): + def memory(self) -> interfaces.layers.Memory: """A Memory object, allowing access to all data and translation layers currently available within the context""" return self._memory # ## Address Space Functions - def add_layer(self, layer): + def add_layer(self, layer: interfaces.layers.DataLayerInterface) -> None: """Adds a named translation layer to the context :param layer: The layer to be added to the memory @@ -71,7 +72,10 @@ class Context(interfaces.context.ContextInterface): # ## Object Factory Functions - def object(self, symbol, layer_name, offset, **arguments): + def object(self, + symbol: interfaces.objects.Template, + layer_name: str, + offset: int, **arguments) -> interfaces.objects.ObjectInterface: """Object factory, takes a context, symbol, offset and optional layername Looks up the layername in the context, finds the object template based on the symbol, @@ -95,16 +99,16 @@ class Context(interfaces.context.ContextInterface): object_info = interfaces.objects.ObjectInformation(layer_name = layer_name, offset = offset)) - def module(self, module_name, layer_name, offset): + def module(self, module_name: str, layer_name: str, offset: int) -> interfaces.context.Module: """Create a module object """ return Module(self, module_name, layer_name, offset) -def get_module_wrapper(method): +def get_module_wrapper(method: str) -> typing.Callable: """Returns a symbol using the symbol_table of the Module""" - def wrapper(self, name): + def wrapper(self, name: str) -> typing.Callable: self._check_type(name, str) if constants.BANG in name: raise ValueError("Name cannot reference another module") @@ -114,7 +118,11 @@ def get_module_wrapper(method): class Module(interfaces.context.Module): - def object(self, symbol_name = None, type_name = None, offset = None, **kwargs): + def object(self, + symbol_name: str = None, + type_name: str = None, + offset: int = None, + **kwargs) -> interfaces.objects.ObjectInterface: """Returns an object created using the symbol_table and layer_name of the Module @param symbol_name: Name of the symbol (within the module) to construct, type_name and offset must not be specified diff --git a/volatility/framework/validity.py b/volatility/framework/validity.py index 150e5e581..f7869ffa6 100644 --- a/volatility/framework/validity.py +++ b/volatility/framework/validity.py @@ -2,7 +2,7 @@ """ import typing -ProgressCallback = typing.Union[typing.Callable[[int, str], None], None] +ProgressCallback = typing.Optional[typing.Callable[[int, str], None]] class ValidityRoutines(object):