Merge branch 'volatilityfoundation:develop' into feature/vadwalk

This commit is contained in:
Donghyun Kim
2022-08-27 01:21:17 +09:00
committed by GitHub
2 changed files with 39 additions and 19 deletions
@@ -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):
@@ -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),