diff --git a/volatility3/framework/plugins/windows/netstat.py b/volatility3/framework/plugins/windows/netstat.py index 486957565..93ac3af93 100644 --- a/volatility3/framework/plugins/windows/netstat.py +++ b/volatility3/framework/plugins/windows/netstat.py @@ -433,7 +433,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): self.context, interfaces.configuration.path_join(self.config_path, 'tcpip'), kernel.layer_name, "tcpip.pdb", tcpip_module.DllBase, tcpip_module.SizeOfImage) except exceptions.VolatilityException: - vollog.warning("Unable to locate symbols for the memory image's tcpip module") + vollog.error("Unable to locate symbols for the memory image's tcpip module") for netw_obj in self.list_sockets(self.context, kernel.layer_name, kernel.symbol_table_name, netscan_symbol_table, tcpip_module.DllBase, tcpip_symbol_table): diff --git a/volatility3/plugins/windows/registry/certificates.py b/volatility3/plugins/windows/registry/certificates.py index 91f17fb2d..6029c0a5c 100644 --- a/volatility3/plugins/windows/registry/certificates.py +++ b/volatility3/plugins/windows/registry/certificates.py @@ -1,11 +1,14 @@ +import contextlib +import logging import struct -from typing import List, Iterator, Tuple +from typing import List, Iterator, Optional, Tuple, Type -from volatility3.framework import interfaces, renderers +from volatility3.framework import constants, exceptions, interfaces, renderers from volatility3.framework.configuration import requirements from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes from volatility3.plugins.windows.registry import hivelist, printkey +vollog = logging.getLogger(__name__) class Certificates(interfaces.plugins.PluginInterface): """Lists the certificates in the registry's Certificate Store.""" @@ -15,12 +18,14 @@ class Certificates(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: return [ - requirements.TranslationLayerRequirement(name = 'primary', - description = 'Memory layer for the kernel', + requirements.ModuleRequirement(name = 'kernel', description = 'Windows 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 = 'printkey', plugin = printkey.PrintKey, version = (1, 0, 0)) + requirements.PluginRequirement(name = 'printkey', plugin = printkey.PrintKey, version = (1, 0, 0)), + requirements.BooleanRequirement(name = 'dump', + description = "Extract listed certificates", + default = False, + optional = True) ] def parse_data(self, data: bytes) -> Tuple[str, bytes]: @@ -34,36 +39,51 @@ class Certificates(interfaces.plugins.PluginInterface): elif ctype == 0x100000020: certificate_data = cvalue return (name, certificate_data) + + @classmethod + def dump_certificate(cls, certificate_data: bytes, hive_offset: int, + reg_section: str, key_hash: str, + open_method: Type[interfaces.plugins.FileHandlerInterface]) -> \ + Optional[interfaces.plugins.FileHandlerInterface]: + try: + if not isinstance(certificate_data, interfaces.renderers.BaseAbsentValue): + dump_name = "{}-{}-{}.crt".format(hive_offset, reg_section, key_hash) + file_handle = open_method(dump_name) + file_handle.write(certificate_data) + return file_handle + except exceptions.InvalidAddressException: + vollog.debug(f"Unable to certificate file at {hive_offset:#x}") + return None + def _generator(self) -> Iterator[Tuple[int, Tuple[str, str, str, str]]]: + kernel = self.context.modules[self.config['kernel']] + for hive in hivelist.HiveList.list_hives(self.context, base_config_path = self.config_path, - layer_name = self.config['primary'], - symbol_table = self.config['nt_symbols']): + layer_name = kernel.layer_name, + symbol_table = kernel.symbol_table_name): for top_key in [ "Microsoft\\SystemCertificates", "Software\\Microsoft\\SystemCertificates", ]: - try: + with contextlib.suppress(KeyError, exceptions.InvalidAddressException): # Walk it node_path = hive.get_key(top_key, return_list = True) - for (depth, is_key, last_write_time, key_path, volatility, - node) in printkey.PrintKey.key_iterator(hive, node_path, recurse = True): + for (_depth, is_key, _last_write_time, key_path, _volatility, node) in printkey.PrintKey.key_iterator(hive, node_path, recurse = True): if not is_key and RegValueTypes(node.Type).name == "REG_BINARY": name, certificate_data = self.parse_data(node.decode_data()) unique_key_offset = key_path.casefold().index(top_key.casefold()) + len(top_key) + 1 reg_section = key_path[unique_key_offset:key_path.index("\\", unique_key_offset)] key_hash = key_path[key_path.rindex("\\") + 1:] - if not isinstance(certificate_data, interfaces.renderers.BaseAbsentValue): - with self.open("{} - {} - {}.crt".format(hex(hive.hive_offset), reg_section, - key_hash)) as file_data: - file_data.write(certificate_data) + if self.config['dump']: + file_handle = self.dump_certificate(certificate_data, hive.hive_offset, reg_section, key_hash, self.open) + if file_handle: + file_handle.close() + yield (0, (top_key, reg_section, key_hash, name)) - except KeyError: - # Key wasn't found in this hive, carry on - pass def run(self) -> renderers.TreeGrid: return renderers.TreeGrid([("Certificate path", str), ("Certificate section", str), ("Certificate ID", str),