diff --git a/volatility/cli/text_renderer.py b/volatility/cli/text_renderer.py index 91124208e..098e21d59 100644 --- a/volatility/cli/text_renderer.py +++ b/volatility/cli/text_renderer.py @@ -69,6 +69,7 @@ def multitypedata_as_text(value: format_hints.MultiTypeData) -> str: def optional(func): + @wraps(func) def wrapped(x: Any) -> str: if isinstance(x, interfaces.renderers.BaseAbsentValue): @@ -82,6 +83,7 @@ def optional(func): def quoted_optional(func): + @wraps(func) def wrapped(x: Any) -> str: result = optional(func)(x) @@ -264,8 +266,7 @@ class PrettyTextRenderer(CLIRenderer): max_column_widths = dict([(column.name, len(column.name)) for column in grid.columns]) def visitor( - node: interfaces.renderers.TreeNode, - accumulator: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] + node: interfaces.renderers.TreeNode, accumulator: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] ) -> List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]]: # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case max_column_widths[tree_indent_column] = max(max_column_widths.get(tree_indent_column, 0), node.path_depth) @@ -327,8 +328,7 @@ class JsonRenderer(CLIRenderer): {}, []) # type: Tuple[Dict[str, List[interfaces.renderers.TreeNode]], List[interfaces.renderers.TreeNode]] def visitor( - node: interfaces.renderers.TreeNode, - accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]] + node: interfaces.renderers.TreeNode, accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]] ) -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]: # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case acc_map, final_tree = accumulator diff --git a/volatility/framework/automagic/symbol_finder.py b/volatility/framework/automagic/symbol_finder.py index aea189e16..50844d0fa 100644 --- a/volatility/framework/automagic/symbol_finder.py +++ b/volatility/framework/automagic/symbol_finder.py @@ -89,7 +89,8 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): # Check if the Stacker has already found what we're looking for if layer.config.get(self.banner_config_key, None): - banner_list = [(0, bytes(layer.config[self.banner_config_key], 'raw_unicode_escape'))] # type: Iterable[Any] + banner_list = [(0, bytes(layer.config[self.banner_config_key], + 'raw_unicode_escape'))] # type: Iterable[Any] else: # Swap to the physical layer for scanning # TODO: Fix this so it works for layers other than just Intel diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index c104b83ca..1556f3432 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -300,8 +300,7 @@ class TranslationLayerRequirement(interfaces.configuration.ConstructableRequirem args = {"context": context, "config_path": config_path, "name": name} if any( - [subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if - not subreq.optional]): + [subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if not subreq.optional]): return None obj = self._construct_class(context, config_path, args) @@ -356,8 +355,7 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn args = {"context": context, "config_path": config_path, "name": name} if any( - [subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if - not subreq.optional]): + [subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if not subreq.optional]): return None # Fill out the parameter for class creation diff --git a/volatility/framework/layers/codecs/__init__.py b/volatility/framework/layers/codecs/__init__.py index 76611fb95..550161e6d 100644 --- a/volatility/framework/layers/codecs/__init__.py +++ b/volatility/framework/layers/codecs/__init__.py @@ -2,5 +2,3 @@ """ - - diff --git a/volatility/framework/layers/resources.py b/volatility/framework/layers/resources.py index 33fc28485..bb5d9d494 100644 --- a/volatility/framework/layers/resources.py +++ b/volatility/framework/layers/resources.py @@ -33,7 +33,6 @@ except ImportError: vollog = logging.getLogger(__name__) - # TODO: Type-annotating the ResourceAccessor.open method is difficult because HTTPResponse is not actually an IO[Any] type # fix this @@ -117,9 +116,9 @@ class ResourceAccessor(object): else: # TODO: find a way to check if we already have this file (look at http headers?) block_size = 1028 * 8 - temp_filename = os.path.join(constants.CACHE_PATH, - "data_" + hashlib.sha512( - bytes(url, 'raw_unicode_escape')).hexdigest() + ".cache") + temp_filename = os.path.join( + constants.CACHE_PATH, + "data_" + hashlib.sha512(bytes(url, 'raw_unicode_escape')).hexdigest() + ".cache") if not os.path.exists(temp_filename): vollog.debug("Caching file at: {}".format(temp_filename)) diff --git a/volatility/framework/plugins/isfinfo.py b/volatility/framework/plugins/isfinfo.py index d933989ea..8852745a4 100644 --- a/volatility/framework/plugins/isfinfo.py +++ b/volatility/framework/plugins/isfinfo.py @@ -93,6 +93,7 @@ class IsfInfo(plugins.PluginInterface): def check_valid(data): return "True" if schemas.validate(data, True) else "False" except ImportError: + def check_valid(data): return "Unknown" @@ -116,14 +117,13 @@ class IsfInfo(plugins.PluginInterface): valid = check_valid(data) except (UnicodeDecodeError, json.decoder.JSONDecodeError): vollog.warning("Invalid ISF: {}".format(entry)) - yield (0, (entry, valid, num_bases, num_types, num_symbols, num_enums, - windows_info, linux_banner, mac_banner)) + yield (0, (entry, valid, num_bases, num_types, num_symbols, num_enums, windows_info, linux_banner, + mac_banner)) # Try to open the file, load it as JSON, read the data from it def run(self): - return renderers.TreeGrid( - [("URI", str), ("Valid", str), - ("Number of base_types", int), ("Number of types", int), ("Number of symbols", int), - ("Number of enums", int), ("Windows info", str), ("Linux banner", str), ("Mac banner", str)], - self._generator()) + return renderers.TreeGrid([("URI", str), ("Valid", str), + ("Number of base_types", int), ("Number of types", int), ("Number of symbols", int), + ("Number of enums", int), ("Windows info", str), ("Linux banner", str), + ("Mac banner", str)], self._generator()) diff --git a/volatility/framework/plugins/linux/check_idt.py b/volatility/framework/plugins/linux/check_idt.py index 722b8c029..c7430e256 100644 --- a/volatility/framework/plugins/linux/check_idt.py +++ b/volatility/framework/plugins/linux/check_idt.py @@ -23,7 +23,6 @@ class Check_idt(interfaces.plugins.PluginInterface): requirements.TranslationLayerRequirement(name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]), - requirements.SymbolTableRequirement(name = "vmlinux", description = "Linux kernel symbols"), requirements.VersionRequirement(name = 'linuxutils', component = linux.LinuxUtilities, version = (1, 0, 0)), requirements.PluginRequirement(name = 'lsmod', plugin = lsmod.Lsmod, version = (1, 0, 0)) @@ -61,7 +60,9 @@ class Check_idt(interfaces.plugins.PluginInterface): addrs = vmlinux.object_from_symbol("idt_table") - table = vmlinux.object(object_type = 'array', offset = addrs.vol.offset, subtype = vmlinux.get_type(idt_type), + table = vmlinux.object(object_type = 'array', + offset = addrs.vol.offset, + subtype = vmlinux.get_type(idt_type), count = idt_table_size) for i in check_idxs: @@ -90,6 +91,5 @@ class Check_idt(interfaces.plugins.PluginInterface): yield (0, [format_hints.Hex(i), format_hints.Hex(idt_addr), module_name, symbol_name]) def run(self): - return renderers.TreeGrid( - [("Index", format_hints.Hex), ("Address", format_hints.Hex), ("Module", str), ("Symbol", str)], - self._generator()) + return renderers.TreeGrid([("Index", format_hints.Hex), ("Address", format_hints.Hex), ("Module", str), + ("Symbol", str)], self._generator()) diff --git a/volatility/framework/plugins/timeliner.py b/volatility/framework/plugins/timeliner.py index 40c4ecbdd..495ee543b 100644 --- a/volatility/framework/plugins/timeliner.py +++ b/volatility/framework/plugins/timeliner.py @@ -143,13 +143,12 @@ class Timeliner(interfaces.plugins.PluginInterface): # Body format is: MD5|name|inode|mode_as_string|UID|GID|size|atime|mtime|ctime|crtime if self._any_time_present(times): - fp.write( - "|{} - {}||||||{}|{}|{}|{}\n".format( - plugin_name, self._sanitize_body_format(item), - self._text_format(times.get(TimeLinerType.ACCESSED, "")), - self._text_format(times.get(TimeLinerType.MODIFIED, "")), - self._text_format(times.get(TimeLinerType.CHANGED, "")), - self._text_format(times.get(TimeLinerType.CREATED, "")))) + fp.write("|{} - {}||||||{}|{}|{}|{}\n".format( + plugin_name, self._sanitize_body_format(item), + self._text_format(times.get(TimeLinerType.ACCESSED, "")), + self._text_format(times.get(TimeLinerType.MODIFIED, "")), + self._text_format(times.get(TimeLinerType.CHANGED, "")), + self._text_format(times.get(TimeLinerType.CREATED, "")))) self.produce_file(filedata) def _sanitize_body_format(self, value): @@ -188,7 +187,7 @@ class Timeliner(interfaces.plugins.PluginInterface): if isinstance(plugin, TimeLinerInterface): if not len(filter_list) or any( - [filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]): + [filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]): plugins_to_run.append(plugin) except exceptions.UnsatisfiedException as excp: # Remove the failed plugin from the list and continue diff --git a/volatility/framework/plugins/windows/cachedump.py b/volatility/framework/plugins/windows/cachedump.py index 74b57feb3..eeb38b7dc 100644 --- a/volatility/framework/plugins/windows/cachedump.py +++ b/volatility/framework/plugins/windows/cachedump.py @@ -21,14 +21,14 @@ class Cachedump(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls): - return [requirements.TranslationLayerRequirement(name = 'primary', - description = 'Memory layer for the kernel', - architectures = ["Intel32", "Intel64"]), - requirements.SymbolTableRequirement(name = "nt_symbols", - description = "Windows kernel symbols"), - requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)), - requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)) - ] + return [ + requirements.TranslationLayerRequirement(name = 'primary', + description = 'Memory layer for the kernel', + architectures = ["Intel32", "Intel64"]), + requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"), + requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)), + requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)) + ] def get_nlkm(self, sechive, lsakey, is_vista_or_later): return lsadump.Lsadump.get_secret_by_name(sechive, 'NL$KM', lsakey, is_vista_or_later) @@ -44,7 +44,7 @@ class Cachedump(interfaces.plugins.PluginInterface): aes = AES.new(nlkm[16:32], AES.MODE_CBC, ch) data = "" for i in range(0, len(edata), 16): - buf = edata[i: i + 16] + buf = edata[i:i + 16] if len(buf) < 16: buf += (16 - len(buf)) * "\00" data += aes.decrypt(buf) @@ -54,13 +54,12 @@ class Cachedump(interfaces.plugins.PluginInterface): (uname_len, domain_len) = unpack(" 6) or (nt_major_version == 6 and nt_minor_version >= 1) for proc in procs: @@ -110,13 +110,15 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): pass if dll_load_time_field: + # Versions prior to 6.1 won't have the LoadTime attribute + # and 32bit version shouldn't have the Quadpart according to MSDN try: DllLoadTime = conversion.wintime_to_datetime(entry.LoadTime.QuadPart) - except: + except AttributeError: pass dumped = False - if self.config.get('dump'): + if self.config['dump']: filedata = self.dump_pe(self.context, pe_table_name, entry, proc_layer_name) if filedata: filedata.preferred_filename = "pid.{0}.".format(proc_id) + filedata.preferred_filename @@ -130,16 +132,15 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): format_hints.Hex(entry.SizeOfImage), BaseDllName, FullDllName, DllLoadTime, dumped)) def generate_timeline(self): - for row in self._generator(pslist.PsList.list_processes(context = self.context, - layer_name = self.config['primary'], - symbol_table = self.config['nt_symbols'], - filter_func = pslist.PsList.create_pid_filter(None))): + for row in self._generator( + pslist.PsList.list_processes(context = self.context, + layer_name = self.config['primary'], + symbol_table = self.config['nt_symbols'])): _depth, row_data = row if not isinstance(row_data[6], datetime.datetime): continue - description = "DLL Load: Process {} {} Loaded {} ({}) Size {} Offset {}".format(row_data[0], row_data[1], - row_data[4], row_data[5], - row_data[3], row_data[2]) + description = "DLL Load: Process {} {} Loaded {} ({}) Size {} Offset {}".format( + row_data[0], row_data[1], row_data[4], row_data[5], row_data[3], row_data[2]) yield (description, timeliner.TimeLinerType.CREATED, row_data[6]) def run(self): diff --git a/volatility/framework/plugins/windows/filescan.py b/volatility/framework/plugins/windows/filescan.py index 5eb3b02ce..292c7f9e5 100644 --- a/volatility/framework/plugins/windows/filescan.py +++ b/volatility/framework/plugins/windows/filescan.py @@ -58,8 +58,4 @@ class FileScan(interfaces.plugins.PluginInterface): yield (0, (format_hints.Hex(fileobj.vol.offset), file_name, fileobj.Size)) def run(self): - return renderers.TreeGrid([ - ("Offset", format_hints.Hex), - ("Name", str), - ("Size", int) - ], self._generator()) + return renderers.TreeGrid([("Offset", format_hints.Hex), ("Name", str), ("Size", int)], self._generator()) diff --git a/volatility/framework/plugins/windows/lsadump.py b/volatility/framework/plugins/windows/lsadump.py index 5ebc42a7e..75c1d2609 100644 --- a/volatility/framework/plugins/windows/lsadump.py +++ b/volatility/framework/plugins/windows/lsadump.py @@ -23,13 +23,13 @@ class Lsadump(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls): - return [requirements.TranslationLayerRequirement(name = 'primary', - description = 'Memory layer for the kernel', - architectures = ["Intel32", "Intel64"]), - requirements.SymbolTableRequirement(name = "nt_symbols", - description = "Windows kernel symbols"), - requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)) - ] + return [ + requirements.TranslationLayerRequirement(name = 'primary', + description = 'Memory layer for the kernel', + architectures = ["Intel32", "Intel64"]), + requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"), + requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)) + ] @classmethod def decrypt_aes(cls, secret, key): @@ -45,7 +45,7 @@ class Lsadump(interfaces.plugins.PluginInterface): data = b"" for i in range(60, len(secret), 16): aes = AES.new(aeskey, AES.MODE_CBC, b'\x00' * 16) - buf = secret[i: i + 16] + buf = secret[i:i + 16] if len(buf) < 16: buf += (16 - len(buf)) * "\00" data += aes.decrypt(buf) @@ -100,8 +100,7 @@ class Lsadump(interfaces.plugins.PluginInterface): if not enc_secret_value: return None - enc_secret = sechive.read(enc_secret_value.Data + 4, - enc_secret_value.DataLength) + enc_secret = sechive.read(enc_secret_value.Data + 4, enc_secret_value.DataLength) if not enc_secret: return None @@ -131,7 +130,7 @@ class Lsadump(interfaces.plugins.PluginInterface): if len(key[j:j + 7]) < 7: j = len(key[j:j + 7]) - (dec_data_len,) = unpack(" List[poolscanner.PoolConstraint]: + def create_netscan_constraints(context: interfaces.context.ContextInterface, + symbol_table: str) -> List[poolscanner.PoolConstraint]: """Creates a list of Pool Tag Constraints for network objects. Args: @@ -74,10 +79,8 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ] @classmethod - def determine_tcpip_version(cls, - context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str) -> str: + def determine_tcpip_version(cls, context: interfaces.context.ContextInterface, layer_name: str, + nt_symbol_table: str) -> str: """Tries to determine which symbol filename to use for the image's tcpip driver. The logic is partially taken from the info plugin. Args: @@ -116,10 +119,11 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): except: # unsure what to raise here. Also, it might be useful to add some kind of fallback, # either to a user-provided version or to another method to determine tcpip.sys's version - raise exceptions.VolatilityException("Kernel Debug Structure missing VERSION/KUSER structure, unable to determine Windows version!") + raise exceptions.VolatilityException( + "Kernel Debug Structure missing VERSION/KUSER structure, unable to determine Windows version!") - vollog.debug("Determined OS Version: {}.{} {}.{}".format(kuser.NtMajorVersion, kuser.NtMinorVersion, - vers.MajorVersion, vers.MinorVersion)) + vollog.debug("Determined OS Version: {}.{} {}.{}".format(kuser.NtMajorVersion, kuser.NtMinorVersion, + vers.MajorVersion, vers.MinorVersion)) if nt_major_version == 10 and arch == "x64": # win10 x64 has an additional class type we have to include. @@ -127,9 +131,9 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): else: # default to general class types class_types = network.class_types - - # these versions are listed explicitly because symbol files differ based on - # version *and* architecture. this is currently the clearest way to show + + # these versions are listed explicitly because symbol files differ based on + # version *and* architecture. this is currently the clearest way to show # the differences, even if it introduces a fair bit of redundancy. # furthermore, it is easy to append new versions. if arch == "x86": @@ -192,21 +196,16 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): filename = version_dict.get(latest_version) vollog.debug("Unable to find exact matching symbol file, going with latest: {}".format(filename)) else: - raise NotImplementedError("This version of Windows is not supported: {}.{} {}.{}!".format(nt_major_version, - nt_minor_version, - vers.MajorVersion, - vers_minor_version)) + raise NotImplementedError("This version of Windows is not supported: {}.{} {}.{}!".format( + nt_major_version, nt_minor_version, vers.MajorVersion, vers_minor_version)) vollog.debug("Determined symbol filename: {}".format(filename)) return filename, class_types @classmethod - def create_netscan_symbol_table(cls, - context: interfaces.context.ContextInterface, - layer_name: str, - nt_symbol_table: str, - config_path: str) -> str: + def create_netscan_symbol_table(cls, context: interfaces.context.ContextInterface, layer_name: str, + nt_symbol_table: str, config_path: str) -> str: """Creates a symbol table for TCP Listeners and TCP/UDP Endpoints. Args: @@ -262,10 +261,8 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): def _generator(self, show_corrupt_results: Optional[bool] = None): """ Generates the network objects for use in rendering. """ - netscan_symbol_table = self.create_netscan_symbol_table(self.context, - self.config["primary"], - self.config["nt_symbols"], - self.config_path) + netscan_symbol_table = self.create_netscan_symbol_table(self.context, self.config["primary"], + self.config["nt_symbols"], self.config_path) for netw_obj in self.scan(self.context, self.config['primary'], self.config['nt_symbols'], netscan_symbol_table): @@ -280,14 +277,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # For UdpA, the state is always blank and the remote end is asterisks for ver, laddr, _ in netw_obj.dual_stack_sockets(): - yield (0, (format_hints.Hex(netw_obj.vol.offset), - "UDP" + ver, - laddr, - netw_obj.Port, - "*", 0, "", - netw_obj.get_owner_pid() or renderers.UnreadableValue(), - netw_obj.get_owner_procname() or renderers.UnreadableValue(), - netw_obj.get_create_time() or renderers.UnreadableValue())) + yield (0, (format_hints.Hex(netw_obj.vol.offset), "UDP" + ver, laddr, netw_obj.Port, "*", 0, "", + netw_obj.get_owner_pid() or renderers.UnreadableValue(), netw_obj.get_owner_procname() + or renderers.UnreadableValue(), netw_obj.get_create_time() + or renderers.UnreadableValue())) elif isinstance(netw_obj, network._TCP_ENDPOINT): vollog.debug("Found _TCP_ENDPOINT @ 0x{:2x}".format(netw_obj.vol.offset)) @@ -303,14 +296,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): except ValueError: state = renderers.UnreadableValue() - yield (0, (format_hints.Hex(netw_obj.vol.offset), proto, - netw_obj.get_local_address() or renderers.UnreadableValue(), - netw_obj.LocalPort, - netw_obj.get_remote_address() or renderers.UnreadableValue(), - netw_obj.RemotePort, - state, - netw_obj.get_owner_pid() or renderers.UnreadableValue(), - netw_obj.get_owner_procname() or renderers.UnreadableValue(), + yield (0, (format_hints.Hex(netw_obj.vol.offset), proto, netw_obj.get_local_address() + or renderers.UnreadableValue(), netw_obj.LocalPort, netw_obj.get_remote_address() + or renderers.UnreadableValue(), netw_obj.RemotePort, state, netw_obj.get_owner_pid() + or renderers.UnreadableValue(), netw_obj.get_owner_procname() or renderers.UnreadableValue(), netw_obj.get_create_time() or renderers.UnreadableValue())) # check for isinstance of tcp listener last, because all other objects are inherited from here @@ -319,15 +308,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # For TcpL, the state is always listening and the remote port is zero for ver, laddr, raddr in netw_obj.dual_stack_sockets(): - yield (0, (format_hints.Hex(netw_obj.vol.offset), "TCP" + ver, - laddr, - netw_obj.Port, - raddr, - 0, - "LISTENING", - netw_obj.get_owner_pid() or renderers.UnreadableValue(), - netw_obj.get_owner_procname() or renderers.UnreadableValue(), - netw_obj.get_create_time() or renderers.UnreadableValue())) + yield (0, (format_hints.Hex(netw_obj.vol.offset), "TCP" + ver, laddr, netw_obj.Port, raddr, 0, + "LISTENING", netw_obj.get_owner_pid() or renderers.UnreadableValue(), + netw_obj.get_owner_procname() or renderers.UnreadableValue(), netw_obj.get_create_time() + or renderers.UnreadableValue())) else: # this should not happen therefore we log it. vollog.debug("Found network object unsure of its type: {} of type {}".format(netw_obj, type(netw_obj))) @@ -338,8 +322,10 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # Skip network connections without creation time if not isinstance(row_data[9], datetime.datetime): continue - row_data = ["N/A" if isinstance(i, renderers.UnreadableValue) or isinstance(i, renderers.UnparsableValue) - else i for i in row_data] + row_data = [ + "N/A" if isinstance(i, renderers.UnreadableValue) or isinstance(i, renderers.UnparsableValue) else i + for i in row_data + ] description = "Network connection: Process {} {} Local Address {}:{} " \ "Remote Address {}:{} State {} Protocol {} ".format(row_data[7], row_data[8], row_data[2], row_data[3], @@ -361,4 +347,4 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ("PID", int), ("Owner", str), ("Created", datetime.datetime), - ], self._generator(show_corrupt_results=show_corrupt_results)) + ], self._generator(show_corrupt_results = show_corrupt_results)) diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 51a83307a..555a22ab8 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -386,13 +386,14 @@ class PoolScanner(plugins.PluginInterface): else: class_type = extensions.pool.POOL_HEADER - table_name = intermed.IntermediateSymbolTable.create( - context = context, - config_path = configuration.path_join(context.symbol_space[symbol_table].config_path, "poolheader"), - sub_path = "windows", - filename = pool_header_json_filename, - table_mapping = {'nt_symbols': symbol_table}, - class_types = {'_POOL_HEADER': class_type}) + table_name = intermed.IntermediateSymbolTable.create(context = context, + config_path = configuration.path_join( + context.symbol_space[symbol_table].config_path, + "poolheader"), + sub_path = "windows", + filename = pool_header_json_filename, + table_mapping = {'nt_symbols': symbol_table}, + class_types = {'_POOL_HEADER': class_type}) return table_name def run(self) -> renderers.TreeGrid: diff --git a/volatility/framework/plugins/windows/pslist.py b/volatility/framework/plugins/windows/pslist.py index 069eab8f7..2b2929c0c 100644 --- a/volatility/framework/plugins/windows/pslist.py +++ b/volatility/framework/plugins/windows/pslist.py @@ -209,5 +209,4 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ("Offset{0}".format(offsettype), format_hints.Hex), ("Threads", int), ("Handles", int), ("SessionId", int), ("Wow64", bool), ("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime), - ("Dumped", bool)], - self._generator()) + ("Dumped", bool)], self._generator()) diff --git a/volatility/framework/plugins/windows/registry/hivelist.py b/volatility/framework/plugins/windows/registry/hivelist.py index 9ec5d43ce..42e87e6f7 100644 --- a/volatility/framework/plugins/windows/registry/hivelist.py +++ b/volatility/framework/plugins/windows/registry/hivelist.py @@ -56,7 +56,6 @@ class HiveList(interfaces.plugins.PluginInterface): description = "Extract listed registry hives", default = False, optional = True) - ] def _sanitize_hive_name(self, name: str) -> str: @@ -73,11 +72,12 @@ class HiveList(interfaces.plugins.PluginInterface): dumped = False if self.config['dump']: # Construct the hive - hive = next(self.list_hives(self.context, - self.config_path, - layer_name = self.config["primary"], - symbol_table = self.config["nt_symbols"], - hive_offsets = [hive_object.vol.offset])) + hive = next( + self.list_hives(self.context, + self.config_path, + layer_name = self.config["primary"], + symbol_table = self.config["nt_symbols"], + hive_offsets = [hive_object.vol.offset])) maxaddr = hive.hive.Storage[0].Length hive_name = self._sanitize_hive_name(hive.get_name()) diff --git a/volatility/framework/plugins/windows/registry/hivescan.py b/volatility/framework/plugins/windows/registry/hivescan.py index a6aa927fa..16d555da6 100644 --- a/volatility/framework/plugins/windows/registry/hivescan.py +++ b/volatility/framework/plugins/windows/registry/hivescan.py @@ -69,7 +69,7 @@ class HiveScan(interfaces.plugins.PluginInterface): def _generator(self): for hive in self.scan_hives(self.context, self.config['primary'], self.config['nt_symbols']): - yield (0, (format_hints.Hex(hive.vol.offset),)) + yield (0, (format_hints.Hex(hive.vol.offset), )) def run(self): return renderers.TreeGrid([("Offset", format_hints.Hex)], self._generator()) diff --git a/volatility/framework/plugins/windows/registry/printkey.py b/volatility/framework/plugins/windows/registry/printkey.py index 63316e17d..85f0a5956 100644 --- a/volatility/framework/plugins/windows/registry/printkey.py +++ b/volatility/framework/plugins/windows/registry/printkey.py @@ -43,10 +43,10 @@ class PrintKey(interfaces.plugins.PluginInterface): @classmethod def key_iterator( - cls, - hive: RegistryHive, - node_path: Sequence[objects.StructType] = None, - recurse: bool = False + cls, + hive: RegistryHive, + node_path: Sequence[objects.StructType] = None, + recurse: bool = False ) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, interfaces.objects.ObjectInterface]]: """Walks through a set of nodes from a given node (last one in node_path). Avoids loops by not traversing into nodes already present diff --git a/volatility/framework/plugins/windows/svcscan.py b/volatility/framework/plugins/windows/svcscan.py index 49354829c..e47c51ce9 100644 --- a/volatility/framework/plugins/windows/svcscan.py +++ b/volatility/framework/plugins/windows/svcscan.py @@ -75,8 +75,7 @@ class SvcScan(interfaces.plugins.PluginInterface): symbol_filename = "services-win10-15063-x86" elif versions.is_windows_8_or_later(context = context, symbol_table = symbol_table) and is_64bit: symbol_filename = "services-win8-x64" - elif versions.is_windows_8_or_later(context = context, - symbol_table = symbol_table) and not is_64bit: + elif versions.is_windows_8_or_later(context = context, symbol_table = symbol_table) and not is_64bit: symbol_filename = "services-win8-x86" elif versions.is_vista_or_later(context = context, symbol_table = symbol_table) and is_64bit: symbol_filename = "services-vista-x64" diff --git a/volatility/framework/plugins/windows/vadinfo.py b/volatility/framework/plugins/windows/vadinfo.py index 132401988..c2191f46e 100644 --- a/volatility/framework/plugins/windows/vadinfo.py +++ b/volatility/framework/plugins/windows/vadinfo.py @@ -132,7 +132,7 @@ class VadInfo(interfaces.plugins.PluginInterface): offset = vad_start while offset < vad_end: to_read = min(chunk_size, vad_end - offset) - data = proc_layer.read(offset, to_read, pad=True) + data = proc_layer.read(offset, to_read, pad = True) if not data: break filedata.data.write(data) diff --git a/volatility/framework/plugins/windows/verinfo.py b/volatility/framework/plugins/windows/verinfo.py index d7d5a986b..e0a4aa462 100644 --- a/volatility/framework/plugins/windows/verinfo.py +++ b/volatility/framework/plugins/windows/verinfo.py @@ -109,8 +109,8 @@ class VerInfo(interfaces.plugins.PluginInterface): session_layer_name = modules.Modules.find_session_layer(self.context, session_layers, mod.DllBase) (major, minor, product, build) = [ - renderers.NotAvailableValue() - ] * 4 # type: Tuple[Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue]] + renderers.NotAvailableValue() + ] * 4 # type: Tuple[Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue]] try: (major, minor, product, build) = self.get_version_information(self._context, pe_table_name, session_layer_name, mod.DllBase) diff --git a/volatility/framework/symbols/linux/extensions/elf.py b/volatility/framework/symbols/linux/extensions/elf.py index 31c60fdc8..5113538bf 100644 --- a/volatility/framework/symbols/linux/extensions/elf.py +++ b/volatility/framework/symbols/linux/extensions/elf.py @@ -6,6 +6,7 @@ from typing import Dict, Tuple from volatility.framework import constants from volatility.framework import objects, interfaces + class elf(objects.StructType): ''' Class used to create elf objects. It overrides the typename to `Elf32_` or `Elf64_`, diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index e3d98bc8b..54bdfc436 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -16,7 +16,6 @@ from volatility.framework.symbols.windows.extensions import pool vollog = logging.getLogger(__name__) - # Keep these in a basic module, to prevent import cycles when symbol providers require them diff --git a/volatility/framework/symbols/windows/extensions/network.py b/volatility/framework/symbols/windows/extensions/network.py index 5540b2797..4919f3412 100644 --- a/volatility/framework/symbols/windows/extensions/network.py +++ b/volatility/framework/symbols/windows/extensions/network.py @@ -17,6 +17,7 @@ from typing import Dict, Tuple vollog = logging.getLogger(__name__) + def inet_ntop(address_family: int, packed_ip: Array) -> str: def inet_ntop4(packed_ip: Array) -> str: @@ -68,6 +69,7 @@ def inet_ntop(address_family: int, packed_ip: Array) -> str: return inet_ntop6(packed_ip) raise socket.error("[Errno 97] Address family not supported by protocol") + # Python's socket.AF_INET6 is 0x1e but Microsoft defines it # as a constant value of 0x17 in their source code. Thus we # need Microsoft's since that's what is found in memory. @@ -78,6 +80,7 @@ AF_INET6 = 0x17 inaddr_any = inet_ntop(socket.AF_INET, [0] * 4) inaddr6_any = inet_ntop(socket.AF_INET6, [0] * 16) + class _TCP_LISTENER(objects.StructType): """Class for objects found in TcpL pools. @@ -132,8 +135,9 @@ class _TCP_LISTENER(objects.StructType): def get_owner_procname(self): if self.get_owner().is_valid(): if self.get_owner().has_valid_member("ImageFileName"): - return self.get_owner().ImageFileName.cast( - "string", max_length = self.get_owner().ImageFileName.vol.count, errors = "replace") + return self.get_owner().ImageFileName.cast("string", + max_length = self.get_owner().ImageFileName.vol.count, + errors = "replace") return None @@ -196,6 +200,7 @@ class _TCP_LISTENER(objects.StructType): return False return True + class _TCP_ENDPOINT(_TCP_LISTENER): """Class for objects found in TcpE pools""" @@ -237,7 +242,8 @@ class _TCP_ENDPOINT(_TCP_LISTENER): vollog.debug("invalid due to invalid address_family {}".format(self.get_address_family())) return False - if not self.get_local_address() and (not self.get_owner() or self.get_owner().UniqueProcessId == 0 or self.get_owner().UniqueProcessId > 65535): + if not self.get_local_address() and (not self.get_owner() or self.get_owner().UniqueProcessId == 0 + or self.get_owner().UniqueProcessId > 65535): vollog.debug("invalid due to invalid owner data") return False @@ -247,21 +253,25 @@ class _TCP_ENDPOINT(_TCP_LISTENER): return True + class _UDP_ENDPOINT(_TCP_LISTENER): """Class for objects found in UdpA pools""" + class _LOCAL_ADDRESS(objects.StructType): @property def inaddr(self): return self.pData.dereference().dereference() + class _LOCAL_ADDRESS_WIN10_UDP(objects.StructType): @property def inaddr(self): return self.pData.dereference() + win10_x64_class_types = { '_TCP_ENDPOINT': _TCP_ENDPOINT, '_TCP_LISTENER': _TCP_LISTENER, diff --git a/volatility/framework/symbols/windows/extensions/pool.py b/volatility/framework/symbols/windows/extensions/pool.py index 6990506de..d79e8b596 100644 --- a/volatility/framework/symbols/windows/extensions/pool.py +++ b/volatility/framework/symbols/windows/extensions/pool.py @@ -39,11 +39,11 @@ class POOL_HEADER(objects.StructType): # because symbol_table_name will be different from kernel_symbol_table. if kernel_symbol_table: object_header_type = self._context.symbol_space.get_type(kernel_symbol_table + constants.BANG + - "_OBJECT_HEADER") + "_OBJECT_HEADER") else: # otherwise symbol_table_name *is* the kernel symbol table, so just use that. object_header_type = self._context.symbol_space.get_type(symbol_table_name + constants.BANG + - "_OBJECT_HEADER") + "_OBJECT_HEADER") pool_header_size = self.vol.size @@ -160,8 +160,8 @@ class POOL_HEADER(objects.StructType): headers = [] sizes = [] for header in [ - 'CREATOR_INFO', 'NAME_INFO', 'HANDLE_INFO', 'QUOTA_INFO', 'PROCESS_INFO', 'AUDIT_INFO', 'EXTENDED_INFO', - 'HANDLE_REVOCATION_INFO', 'PADDING_INFO' + 'CREATOR_INFO', 'NAME_INFO', 'HANDLE_INFO', 'QUOTA_INFO', 'PROCESS_INFO', 'AUDIT_INFO', 'EXTENDED_INFO', + 'HANDLE_REVOCATION_INFO', 'PADDING_INFO' ]: try: type_name = "{}{}_OBJECT_HEADER_{}".format(symbol_table_name, constants.BANG, header) diff --git a/volatility/framework/symbols/windows/pdbconv.py b/volatility/framework/symbols/windows/pdbconv.py index 1532650f6..4909f3726 100644 --- a/volatility/framework/symbols/windows/pdbconv.py +++ b/volatility/framework/symbols/windows/pdbconv.py @@ -265,7 +265,8 @@ class PdbReader: if not progress_callback: progress_callback = lambda x, y: None self._progress_callback = progress_callback - self.types = [] # type: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]] + self.types = [ + ] # type: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]] self.bases = {} # type: Dict[str, Any] self.user_types = {} # type: Dict[str, Any] self.enumerations = {} # type: Dict[str, Any] @@ -619,8 +620,8 @@ class PdbReader: else: leaf_type, name, value = self.types[index - 0x1000] if leaf_type in [ - leaf_type.LF_UNION, leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, - leaf_type.LF_STRUCTURE_ST, leaf_type.LF_INTERFACE + leaf_type.LF_UNION, leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, + leaf_type.LF_STRUCTURE_ST, leaf_type.LF_INTERFACE ]: if not value.properties.forward_reference: result = value.size @@ -664,8 +665,8 @@ class PdbReader: self._progress_callback(index * 100 / max_len, "Processing types") leaf_type, name, value = self.types[index] if leaf_type in [ - leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, leaf_type.LF_STRUCTURE_ST, - leaf_type.LF_INTERFACE + leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, leaf_type.LF_STRUCTURE_ST, + leaf_type.LF_INTERFACE ]: if not value.properties.forward_reference and name: self.user_types[name] = { @@ -692,17 +693,16 @@ class PdbReader: self.enumerations[name] = { 'base': base['name'], 'size': self.get_size_from_index(value.subtype_index), - 'constants': - dict([(name, enum.value) for _, name, enum in constants]) + 'constants': dict([(name, enum.value) for _, name, enum in constants]) } # Re-run through for ForwardSizeReferences self.user_types = self.replace_forward_references(self.user_types, type_references) def consume_type( - self, module: interfaces.context.ModuleInterface, offset: int, length: int + self, module: interfaces.context.ModuleInterface, offset: int, length: int ) -> Tuple[Tuple[Optional[interfaces.objects.ObjectInterface], Optional[str], Union[ - None, List, interfaces.objects.ObjectInterface]], int]: + None, List, interfaces.objects.ObjectInterface]], int]: """Returns a (leaf_type, name, object) Tuple for a type, and the number of bytes consumed.""" result = None, None, None # type: Tuple[Optional[interfaces.objects.ObjectInterface], Optional[str], Optional[Union[List, interfaces.objects.ObjectInterface]]] @@ -713,8 +713,8 @@ class PdbReader: remaining = length - consumed if leaf_type in [ - leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, leaf_type.LF_STRUCTURE_ST, - leaf_type.LF_INTERFACE + leaf_type.LF_CLASS, leaf_type.LF_CLASS_ST, leaf_type.LF_STRUCTURE, leaf_type.LF_STRUCTURE_ST, + leaf_type.LF_INTERFACE ]: structure = module.object(object_type = "LF_STRUCTURE", offset = offset + consumed) name_offset = structure.name.vol.offset - structure.vol.offset @@ -914,7 +914,6 @@ class PdbRetreiver: if __name__ == '__main__': import argparse - class PrintedProgress(object): """A progress handler that prints the progress value and the description onto the command line.""" @@ -935,7 +934,6 @@ if __name__ == '__main__': self._max_message_len = max([self._max_message_len, message_len]) print(message, end = (' ' * (self._max_message_len - message_len)) + '\r') - parser = argparse.ArgumentParser( description = "Read PDB files and convert to Volatility 3 Intermediate Symbol Format") parser.add_argument("-o", "--output", metavar = "OUTPUT", help = "Filename for data output", required = True) diff --git a/volatility/framework/symbols/windows/versions.py b/volatility/framework/symbols/windows/versions.py index e45aed3ad..e13042626 100644 --- a/volatility/framework/symbols/windows/versions.py +++ b/volatility/framework/symbols/windows/versions.py @@ -31,8 +31,8 @@ class OsDistinguisher: A function that takes a context and a symbol table name and determines whether that symbol table passes the distinguishing checks """ - def __init__(self, version_check: Callable[[Tuple[int, ...]], bool], - fallback_checks: List[Tuple[str, Optional[str], bool]]): + def __init__(self, version_check: Callable[[Tuple[int, ...]], bool], fallback_checks: List[Tuple[str, Optional[str], + bool]]): self._version_check = version_check self._fallback_checks = fallback_checks @@ -59,7 +59,7 @@ class OsDistinguisher: for name, member, response in self._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: + or context.symbol_space.has_type(symbol_table + constants.BANG + name)) != response: return False else: try: @@ -80,8 +80,7 @@ is_vista_or_later = OsDistinguisher(version_check = lambda x: x >= (6, 0), fallback_checks = [("KdCopyDataBlock", None, True)]) is_win10 = OsDistinguisher(version_check = lambda x: (10, 0) <= x, - fallback_checks = [("ObHeaderCookie", None, True), - ("_HANDLE_TABLE", "HandleCount", False)]) + fallback_checks = [("ObHeaderCookie", None, True), ("_HANDLE_TABLE", "HandleCount", False)]) is_windows_xp = OsDistinguisher(version_check = lambda x: (5, 1) <= x < (5, 2), fallback_checks = [("KdCopyDataBlock", None, False), @@ -106,8 +105,7 @@ is_win10_16299_or_later = OsDistinguisher(version_check = lambda x: x >= (10, 0, fallback_checks = [("ObHeaderCookie", None, True), ("_HANDLE_TABLE", "HandleCount", False), ("_EPROCESS", "KeepAliveCounter", False), - ("_EPROCESS", "ControlFlowGuardEnabled", - False)]) + ("_EPROCESS", "ControlFlowGuardEnabled", False)]) is_windows_10 = OsDistinguisher(version_check = lambda x: x >= (10, 0), fallback_checks = [("ObHeaderCookie", None, True)]) diff --git a/volatility/plugins/windows/registry/certificates.py b/volatility/plugins/windows/registry/certificates.py index b364f9613..987358e39 100644 --- a/volatility/plugins/windows/registry/certificates.py +++ b/volatility/plugins/windows/registry/certificates.py @@ -40,8 +40,8 @@ class Certificates(interfaces.plugins.PluginInterface): symbol_table = self.config['nt_symbols']): for top_key in [ - "Microsoft\\SystemCertificates", - "Software\\Microsoft\\SystemCertificates", + "Microsoft\\SystemCertificates", + "Software\\Microsoft\\SystemCertificates", ]: try: # Walk it