diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index 2f3b6ca0a..12e7b85d4 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -46,8 +46,10 @@ def convert_value_to_data(value: TUnion[int, float, bytes, str, bool], if struct_type == int and isinstance(value, int): # Doubling up on the isinstance is for mypy - return int.to_bytes( - value, length = data_format.length, byteorder = data_format.byteorder, signed = data_format.signed) + return int.to_bytes(value, + length = data_format.length, + byteorder = data_format.byteorder, + signed = data_format.signed) if struct_type == bool: struct_format = "?" elif struct_type == float: @@ -102,7 +104,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface): """Creates the appropriate class and returns it so that the native type is inherited. - The only reason the **kwargs is added, is so that the inherriting types can override __init__ + The only reason the kwargs is added, is so that the inherriting types can override __init__ without needing to override __new__ We also sneak in new_value, so that we don't have to do expensive (read: impossible) context reads @@ -176,11 +178,10 @@ class Bytes(PrimitiveObject, bytes): type_name: str, object_info: interfaces.objects.ObjectInformation, length: int = 1) -> None: - super().__init__( - context = context, - type_name = type_name, - object_info = object_info, - data_format = DataFormatInfo(length, "big", False)) + super().__init__(context = context, + type_name = type_name, + object_info = object_info, + data_format = DataFormatInfo(length, "big", False)) self._vol['length'] = length def __new__(cls: Type, @@ -192,13 +193,13 @@ class Bytes(PrimitiveObject, bytes): """Creates the appropriate class and returns it so that the native type is inherritted. - The only reason the **kwargs is added, is so that the + The only reason the kwargs is added, is so that the inherriting types can override __init__ without needing to override __new__ """ return cls._struct_type.__new__( - cls, - cls._unmarshall(context, data_format = DataFormatInfo(length, "big", False), object_info = object_info)) + cls, cls._unmarshall(context, data_format = DataFormatInfo(length, "big", False), + object_info = object_info)) class String(PrimitiveObject, str): @@ -217,11 +218,10 @@ class String(PrimitiveObject, str): max_length: int = 1, encoding: str = "utf-8", errors: str = "strict") -> None: - super().__init__( - context = context, - type_name = type_name, - object_info = object_info, - data_format = DataFormatInfo(max_length, "big", False)) + super().__init__(context = context, + type_name = type_name, + object_info = object_info, + data_format = DataFormatInfo(max_length, "big", False)) self._vol["max_length"] = max_length self._vol['encoding'] = encoding self._vol['errors'] = errors @@ -237,7 +237,7 @@ class String(PrimitiveObject, str): """Creates the appropriate class and returns it so that the native type is inherited. - The only reason the **kwargs is added, is so that the + The only reason the kwargs is added, is so that the inherriting types can override __init__ without needing to override __new__ """ @@ -297,9 +297,10 @@ class Pointer(Integer): layer_name = layer_name or self.vol.native_layer_name mask = self._context.layers[layer_name].address_mask offset = self & mask - return self.vol.subtype( - context = self._context, - object_info = interfaces.objects.ObjectInformation(layer_name = layer_name, offset = offset, parent = self)) + return self.vol.subtype(context = self._context, + object_info = interfaces.objects.ObjectInformation(layer_name = layer_name, + offset = offset, + parent = self)) def is_readable(self, layer_name: Optional[str] = None) -> bool: """Determines whether the address of this pointer can be read from @@ -559,11 +560,10 @@ class Array(interfaces.objects.ObjectInterface, abc.Sequence): return_list = False series = [series] for index in series: - object_info = ObjectInformation( - layer_name = self.vol.layer_name, - offset = mask & (self.vol.offset + (self.vol.subtype.size * index)), - parent = self, - native_layer_name = self.vol.native_layer_name) + object_info = ObjectInformation(layer_name = self.vol.layer_name, + offset = mask & (self.vol.offset + (self.vol.subtype.size * index)), + parent = self, + native_layer_name = self.vol.native_layer_name) result += [self.vol.subtype(context = self._context, object_info = object_info)] if not return_list: return result[0] @@ -587,8 +587,11 @@ class AggregateType(interfaces.objects.ObjectInterface): def __init__(self, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, size: int, members: Dict[str, Tuple[int, interfaces.objects.Template]]) -> None: - super().__init__( - context = context, type_name = type_name, object_info = object_info, size = size, members = members) + super().__init__(context = context, + type_name = type_name, + object_info = object_info, + size = size, + members = members) # self._check_members(members) self._concrete_members = {} # type: Dict[str, Dict] @@ -669,14 +672,13 @@ class AggregateType(interfaces.objects.ObjectInterface): elif attr in self.vol.members: mask = self._context.layers[self.vol.layer_name].address_mask relative_offset, member = self.vol.members[attr] - member = member( - context = self._context, - object_info = interfaces.objects.ObjectInformation( - layer_name = self.vol.layer_name, - offset = mask & (self.vol.offset + relative_offset), - member_name = attr, - parent = self, - native_layer_name = self.vol.native_layer_name)) + member = member(context = self._context, + object_info = interfaces.objects.ObjectInformation( + layer_name = self.vol.layer_name, + offset = mask & (self.vol.offset + relative_offset), + member_name = attr, + parent = self, + native_layer_name = self.vol.native_layer_name)) self._concrete_members[attr] = member return member # We duplicate this code to avoid polluting the methodspace diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 4e123ffaa..67d0ca700 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -65,8 +65,9 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface): def __call__(self, data: bytes, data_offset: int): for offset, pattern in self._subscanner(data, data_offset): - header = self._module.object( - object_type = "_POOL_HEADER", offset = offset - self._header_offset, absolute = True) + header = self._module.object(object_type = "_POOL_HEADER", + offset = offset - self._header_offset, + absolute = True) constraint = self._constraint_lookup[pattern] try: # Size check @@ -184,19 +185,20 @@ class PoolScanner(plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: return [ - requirements.TranslationLayerRequirement( - name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]), + requirements.TranslationLayerRequirement(name = 'primary', + description = 'Memory layer for the kernel', + architectures = ["Intel32", "Intel64"]), requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols") ] - 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)]) + 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)]) + 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): @@ -207,8 +209,9 @@ class PoolScanner(plugins.PluginInterface): symbol_table, constraints): # generate some type-specific info for sanity checking if constraint.object_type == "Process": - name = mem_object.ImageFileName.cast( - "string", max_length = mem_object.ImageFileName.vol.count, errors = "replace") + name = mem_object.ImageFileName.cast("string", + max_length = mem_object.ImageFileName.vol.count, + errors = "replace") elif constraint.object_type == "File": try: name = mem_object.FileName.String @@ -238,93 +241,80 @@ class PoolScanner(plugins.PluginInterface): builtins = [ # atom tables - PoolConstraint( - b'AtmT', - type_name = symbol_table + constants.BANG + "_RTL_ATOM_TABLE", - size = (200, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'AtmT', + type_name = symbol_table + constants.BANG + "_RTL_ATOM_TABLE", + size = (200, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # processes on windows before windows 8 - PoolConstraint( - b'Pro\xe3', - type_name = symbol_table + constants.BANG + "_EPROCESS", - object_type = "Process", - size = (600, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Pro\xe3', + type_name = symbol_table + constants.BANG + "_EPROCESS", + object_type = "Process", + size = (600, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # processes on windows starting with windows 8 - PoolConstraint( - b'Proc', - type_name = symbol_table + constants.BANG + "_EPROCESS", - object_type = "Process", - size = (600, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Proc', + type_name = symbol_table + constants.BANG + "_EPROCESS", + object_type = "Process", + size = (600, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # files on windows before windows 8 - PoolConstraint( - b'Fil\xe5', - type_name = symbol_table + constants.BANG + "_FILE_OBJECT", - object_type = "File", - size = (150, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Fil\xe5', + type_name = symbol_table + constants.BANG + "_FILE_OBJECT", + object_type = "File", + size = (150, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # files on windows starting with windows 8 - PoolConstraint( - b'File', - type_name = symbol_table + constants.BANG + "_FILE_OBJECT", - object_type = "File", - size = (150, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'File', + type_name = symbol_table + constants.BANG + "_FILE_OBJECT", + object_type = "File", + size = (150, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # mutants on windows before windows 8 - PoolConstraint( - b'Mut\xe1', - type_name = symbol_table + constants.BANG + "_KMUTANT", - object_type = "Mutant", - size = (64, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Mut\xe1', + type_name = symbol_table + constants.BANG + "_KMUTANT", + object_type = "Mutant", + size = (64, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # mutants on windows starting with windows 8 - PoolConstraint( - b'Muta', - type_name = symbol_table + constants.BANG + "_KMUTANT", - object_type = "Mutant", - size = (64, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Muta', + type_name = symbol_table + constants.BANG + "_KMUTANT", + object_type = "Mutant", + size = (64, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # drivers on windows before windows 8 - PoolConstraint( - b'Dri\xf6', - type_name = symbol_table + constants.BANG + "_DRIVER_OBJECT", - object_type = "Driver", - size = (248, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Dri\xf6', + type_name = symbol_table + constants.BANG + "_DRIVER_OBJECT", + object_type = "Driver", + size = (248, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # drivers on windows starting with windows 8 - PoolConstraint( - b'Driv', - type_name = symbol_table + constants.BANG + "_DRIVER_OBJECT", - object_type = "Driver", - size = (248, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Driv', + type_name = symbol_table + constants.BANG + "_DRIVER_OBJECT", + object_type = "Driver", + size = (248, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # kernel modules - PoolConstraint( - b'MmLd', - type_name = symbol_table + constants.BANG + "_LDR_DATA_TABLE_ENTRY", - size = (76, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'MmLd', + type_name = symbol_table + constants.BANG + "_LDR_DATA_TABLE_ENTRY", + size = (76, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # symlinks on windows before windows 8 - PoolConstraint( - b'Sym\xe2', - type_name = symbol_table + constants.BANG + "_OBJECT_SYMBOLIC_LINK", - object_type = "SymbolicLink", - size = (72, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Sym\xe2', + type_name = symbol_table + constants.BANG + "_OBJECT_SYMBOLIC_LINK", + object_type = "SymbolicLink", + size = (72, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # symlinks on windows starting with windows 8 - PoolConstraint( - b'Symb', - type_name = symbol_table + constants.BANG + "_OBJECT_SYMBOLIC_LINK", - object_type = "SymbolicLink", - size = (72, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'Symb', + type_name = symbol_table + constants.BANG + "_OBJECT_SYMBOLIC_LINK", + object_type = "SymbolicLink", + size = (72, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), # registry hives - PoolConstraint( - b'CM10', - type_name = symbol_table + constants.BANG + "_CMHIVE", - size = (800, None), - page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), + PoolConstraint(b'CM10', + type_name = symbol_table + constants.BANG + "_CMHIVE", + size = (800, None), + page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE), ] if not tags_filter: @@ -369,13 +359,12 @@ class PoolScanner(plugins.PluginInterface): for constraint, header in cls.pool_scan(context, scan_layer, symbol_table, constraints, alignment = 8): - mem_object = header.get_object( - type_name = constraint.type_name, - type_map = type_map, - use_top_down = is_windows_8_or_later, - object_type = constraint.object_type, - native_layer_name = 'primary', - cookie = cookie) + mem_object = header.get_object(type_name = constraint.type_name, + type_map = type_map, + use_top_down = is_windows_8_or_later, + object_type = constraint.object_type, + native_layer_name = 'primary', + cookie = cookie) if mem_object is None: vollog.log(constants.LOGLEVEL_VVV, "Cannot create an instance of {}".format(constraint.type_name))