Merge branch 'develop' into modxview_plugin

This commit is contained in:
Abyss-W4tcher
2025-01-18 02:16:52 +01:00
committed by GitHub
21 changed files with 257 additions and 104 deletions
+50
View File
@@ -0,0 +1,50 @@
name: build-pyinstaller
on:
push:
branches:
- stable
- develop
- 'release/**'
pull_request:
branches:
- stable
- 'release/**'
jobs:
exe:
runs-on: windows-latest
strategy:
matrix:
python-version: ["3.11"]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pyinstaller
- name: Pyinstall executable
run: |
pyinstaller --clean -y vol.spec
pyinstaller --clean -y volshell.spec
- name: Move files
run: |
mv dist/vol.exe vol.exe
mv dist/volshell.exe volshell.exe
- name: Archive
uses: actions/upload-artifact@v4
with:
name: volatility3-pyinstaller
path: |
vol.exe
volshell.exe
README.md
LICENSE.txt
+1 -1
View File
@@ -88,7 +88,7 @@ The latest generated copy of the documentation can be found at: <https://volatil
## Licensing and Copyright
Copyright (C) 2007-2024 Volatility Foundation
Copyright (C) 2007-2025 Volatility Foundation
All Rights Reserved
+1 -1
View File
@@ -167,7 +167,7 @@ master_doc = "index"
# General information about the project.
project = "Volatility 3"
copyright = "2012-2024, Volatility Foundation"
copyright = "2012-2025, Volatility Foundation"
# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
+1
View File
@@ -367,6 +367,7 @@ class CommandLine:
plugin,
help=plugin_list[plugin].__doc__,
description=plugin_list[plugin].__doc__,
epilog=plugin_list[plugin].additional_description,
)
self.populate_requirements_argparse(plugin_parser, plugin_list[plugin])
+5
View File
@@ -71,6 +71,11 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
elif "init_level4_pgt" in table.symbols:
layer_class = intel.LinuxIntel32e
dtb_symbol_name = "init_level4_pgt"
elif "pkmap_count" in table.symbols and table.get_symbol(
"pkmap_count"
).type.count in (512, 2048):
layer_class = intel.LinuxIntelPAE
dtb_symbol_name = "swapper_pg_dir"
else:
layer_class = intel.LinuxIntel
dtb_symbol_name = "swapper_pg_dir"
@@ -112,6 +112,8 @@ class PluginInterface(
# Be careful with inheritance around this (We default to requiring a version which doesn't exist, so it must be set)
_required_framework_version: Tuple[int, int, int] = (0, 0, 0)
"""The _version variable is a quick way for plugins to define their current interface, it should follow SemVer rules"""
additional_description: str = None
"""Display additional description of the plugin after the description of the arguments. See: https://docs.python.org/3/library/argparse.html#epilog"""
def __init__(
self,
+3 -3
View File
@@ -192,9 +192,9 @@ class RegistryHive(linear.LinearlyMappedLayer):
while key_array and node_key:
subkeys = node_key[-1].get_subkeys()
for subkey in subkeys:
# registry keys are not case sensitive so compare lowercase
# https://msdn.microsoft.com/en-us/library/windows/desktop/ms724946(v=vs.85).aspx
if subkey.get_name().lower() == key_array[0].lower():
# registry keys are not case sensitive so compare likewise
# https://learn.microsoft.com/en-us/windows/win32/sysinfo/structure-of-the-registry
if subkey.get_name().casefold() == key_array[0].casefold():
node_key = node_key + [subkey]
found_key, key_array = found_key + [key_array[0]], key_array[1:]
break
+2 -2
View File
@@ -1,8 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
"""A module containing a plugin that recovers bash command history
from bash process memory."""
import datetime
import struct
@@ -1,8 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
"""A module containing a plugin that verifies the operation function
pointers of network protocols."""
import logging
from typing import List
@@ -1,8 +1,7 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
"""A module containing a plugin that checks the system call table for hooks."""
import contextlib
import logging
from typing import List
+2 -2
View File
@@ -1,8 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
"""A module containing a plugin for enumerating memory-mapped
ELF files across all processes."""
import logging
from typing import List, Optional, Type
+10 -4
View File
@@ -5,7 +5,7 @@
import logging
from typing import Iterable, Tuple
from volatility3.framework import renderers, interfaces
from volatility3.framework import renderers, interfaces, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
@@ -58,10 +58,16 @@ class Envars(plugins.PluginInterface):
Tuples of (key, value) representing each environment variable.
"""
task_name = utility.array_to_string(task.comm)
# This ensures the `task` is valid as well as its
# memory mapping structures
try:
task_name = utility.array_to_string(task.comm)
env_start = task.mm.env_start
env_end = task.mm.env_end
except exceptions.InvalidAddressException:
return None
task_pid = task.pid
env_start = task.mm.env_start
env_end = task.mm.env_end
env_area_size = env_end - env_start
if not (0 < env_area_size <= env_area_max_size):
vollog.debug(
+1 -2
View File
@@ -1,8 +1,7 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
"""A module containing a plugin that lists loaded kernel modules."""
import logging
from typing import List, Iterable
+17 -5
View File
@@ -34,7 +34,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Lists the processes present in a particular linux memory image."""
_required_framework_version = (2, 13, 0)
_version = (4, 0, 0)
_version = (4, 1, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -179,6 +179,10 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_output = "VMA start matching task start_code not found"
return file_output
@staticmethod
def _format_cred(cred):
return renderers.NotAvailableValue() if cred is None else cred
def _generator(
self,
pid_filter: Callable[[Any], bool],
@@ -212,16 +216,21 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
task_fields = self.get_task_fields(task, decorate_comm)
task_uid = self._format_cred(task_fields.uid)
task_gid = self._format_cred(task_fields.gid)
task_euid = self._format_cred(task_fields.euid)
task_egid = self._format_cred(task_fields.egid)
yield 0, (
format_hints.Hex(task_fields.offset),
task_fields.user_pid,
task_fields.user_tid,
task_fields.user_ppid,
task_fields.name,
task_fields.uid or renderers.NotAvailableValue(),
task_fields.gid or renderers.NotAvailableValue(),
task_fields.euid or renderers.NotAvailableValue(),
task_fields.egid or renderers.NotAvailableValue(),
task_uid,
task_gid,
task_euid,
task_egid,
task_fields.creation_time or renderers.NotAvailableValue(),
file_output,
)
@@ -250,6 +259,9 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# Note that the init_task itself is not yielded, since "ps" also never shows it.
for task in init_task.tasks:
if not task.is_valid():
continue
if filter_func(task):
continue
@@ -67,6 +67,7 @@ class CmdScan(interfaces.plugins.PluginInterface):
Args:
conhost_proc: the process object for conhost.exe
size_filter: size above which vads will not be returned
Returns:
A list of tuples of:
@@ -99,8 +100,8 @@ class CmdScan(interfaces.plugins.PluginInterface):
kernel_layer_name: The name of the layer on which to operate
kernel_symbol_table_name: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
procs: list of process objects
max_history: an initial set of CommandHistorySize values
procs: List of process objects
max_history: An initial set of CommandHistorySize values
Returns:
The conhost process object, the command history structure, a dictionary of properties for
@@ -227,7 +228,6 @@ class CmdScan(interfaces.plugins.PluginInterface):
"data": command_history.CommandCountMax,
}
)
command_history_properties.append(
{
"level": 1,
@@ -236,6 +236,7 @@ class CmdScan(interfaces.plugins.PluginInterface):
"data": "",
}
)
for (
cmd_index,
bucket_cmd,
@@ -352,7 +353,7 @@ class CmdScan(interfaces.plugins.PluginInterface):
def _conhost_proc_filter(self, proc: interfaces.objects.ObjectInterface):
"""
Used to filter to only conhost.exe processes
Used to filter only conhost.exe processes
"""
process_name = utility.array_to_string(proc.ImageFileName)
@@ -64,7 +64,7 @@ class DriverScan(interfaces.plugins.PluginInterface):
names associated with a driver
Args:
driver: A Eriver object
driver: A Driver object
Returns:
A tuple of strings of (driver name, service key, driver alt. name)
@@ -305,14 +305,14 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
If a number of validity checks are passed, this method will return the `SHIM_CACHE_HEAD`
object. Otherwise, `None` is returned.
"""
# print("checking RTL_AVL_TABLE at offset %s" % hex(offset))
# Check RTL_AVL_TABLE at offset
rtl_avl_table = context.object(
symbol_table + constants.BANG + "_RTL_AVL_TABLE", layer_name, offset
)
if not rtl_avl_table.is_valid(mod_page_start, mod_page_end):
return None
vollog.debug(f"Candidate RTL_AVL_TABLE found at offset {hex(offset)}")
vollog.debug(f"Candidate RTL_AVL_TABLE found at offset {offset:#x}")
ersrc_size = context.symbol_space.get_type(
kernel_symbol_table + constants.BANG + "_ERESOURCE"
@@ -324,13 +324,13 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
# 0x20 if context.symbol_space.get_type("pointer").size == 8 else 0x10
)
vollog.debug(
f"ERESOURCE size: {hex(ersrc_size)}, ERESOURCE alignment: {hex(ersrc_alignment)}"
f"ERESOURCE size: {ersrc_size:#x}, ERESOURCE alignment: {ersrc_alignment:#x}"
)
eresource_rel_off = ersrc_size + ((offset - ersrc_size) % ersrc_alignment)
eresource_offset = offset - eresource_rel_off
vollog.debug(f"Constructing ERESOURCE at {hex(eresource_offset)}")
vollog.debug(f"Constructing ERESOURCE at {eresource_offset:#x}")
eresource = context.object(
kernel_symbol_table + constants.BANG + "_ERESOURCE",
layer_name,
@@ -408,8 +408,8 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
# iterate over ahcache kernel module's .data section in search of *two* SHIM handles
shim_heads = []
vollog.debug(f"PAGE offset: {hex(mod_page_offset)}")
vollog.debug(f".data offset: {hex(data_sec_offset)}")
vollog.debug(f"PAGE offset: {mod_page_offset:#x}")
vollog.debug(f".data offset: {data_sec_offset:#x}")
handle_type = context.symbol_space.get_type(
shimcache_symbol_table + constants.BANG + "SHIM_CACHE_HANDLE"
@@ -419,7 +419,7 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
data_sec_offset + data_sec_size,
8 if symbols.symbol_table_is_64bit(context, nt_symbol_table) else 4,
):
vollog.debug(f"Building shim handle pointer at {hex(offset)}")
vollog.debug(f"Building shim handle pointer at {offset:#x}")
shim_handle = context.object(
object_type=shimcache_symbol_table + constants.BANG + "pointer",
layer_name=kernel_layer_name,
@@ -430,7 +430,7 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
if shim_handle.is_valid(mod_page_offset, mod_page_offset + mod_page_size):
if shim_handle.head is not None:
vollog.debug(
f"Found valid shim handle @ {hex(shim_handle.vol.offset)}"
f"Found valid shim handle @ {shim_handle.vol.offset:#x}"
)
shim_heads.append(shim_handle.head)
if len(shim_heads) == 2:
@@ -440,7 +440,7 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
vollog.debug("Failed to identify two valid SHIM_CACHE_HANDLE structures")
return
# On Windows 8 x64, the frist cache contains the shim cache
# On Windows 8 x64, the first cache contains the shim cache.
# On Windows 8 x86, 8.1 x86/x64, and 10, the second cache contains the shim cache.
if (
not symbols.symbol_table_is_64bit(context, nt_symbol_table)
@@ -18,7 +18,7 @@ def wintime_to_datetime(
unix_time = wintime // 10000000
if unix_time == 0:
return renderers.NotApplicableValue()
unix_time = unix_time - 11644473600
unix_time -= 11644473600
try:
return datetime.datetime.fromtimestamp(unix_time, datetime.timezone.utc)
# Windows sometimes throws OSErrors rather than ValueError/OverflowError when it can't convert a value
@@ -71,7 +71,7 @@ def round(addr: int, align: int, up: bool = False) -> int:
Args:
addr: the address
align: the alignment value
up: Whether to round up or not
up: whether to round up or not
Returns:
The aligned address
@@ -122,11 +122,12 @@ def convert_port(port_as_integer):
def convert_network_four_tuple(family, four_tuple):
"""Converts the connection four_tuple: (source ip, source port, dest ip,
dest port)
"""Converts the connection four_tuple:
(source ip, source port, dest ip, dest port)
into their string equivalents. IP addresses are expected as a tuple
of unsigned shorts Ports are converted to proper endianness as well
of unsigned shorts. Ports are converted to proper endianness as well.
"""
if family == socket.AF_INET:
@@ -306,6 +306,46 @@ class module(generic.GenericIntelProcess):
class task_struct(generic.GenericIntelProcess):
def is_valid(self) -> bool:
layer = self._context.layers[self.vol.layer_name]
# Make sure the entire task content is readable
if not layer.is_valid(self.vol.offset, self.vol.size):
return False
if self.pid < 0 or self.tgid < 0:
return False
if self.has_member("signal") and not (
self.signal and self.signal.is_readable()
):
return False
if self.has_member("nsproxy") and not (
self.nsproxy and self.nsproxy.is_readable()
):
return False
if self.has_member("real_parent") and not (
self.real_parent and self.real_parent.is_readable()
):
return False
if (
self.has_member("active_mm")
and self.active_mm
and not self.active_mm.is_readable()
):
return False
if self.mm:
if not self.mm.is_readable():
return False
if self.mm != self.active_mm:
return False
return True
def add_process_layer(
self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
) -> Optional[str]:
@@ -323,9 +363,11 @@ class task_struct(generic.GenericIntelProcess):
raise TypeError(
"Parent layer is not a translation layer, unable to construct process layer"
)
dtb, layer_name = parent_layer.translate(pgd)
if not dtb:
try:
dtb, layer_name = parent_layer.translate(pgd)
except exceptions.InvalidAddressException:
return None
if preferred_name is None:
preferred_name = self.vol.layer_name + f"_Process{self.pid}"
# Add the constructed layer and return the name
@@ -398,6 +440,8 @@ class task_struct(generic.GenericIntelProcess):
tasks_iterable = self._get_tasks_iterable()
threads_seen = set([self.vol.offset])
for task in tasks_iterable:
if not task.is_valid():
continue
if task.vol.offset not in threads_seen:
threads_seen.add(task.vol.offset)
yield task
@@ -808,23 +852,30 @@ class mm_struct(objects.StructType):
def _get_mmap_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns an iterator for the mmap list member of an mm_struct. Use this only if
required, get_vma_iter() will choose the correct _get_maple_tree_iter() or
_get_mmap_iter() automatically as required."""
_get_mmap_iter() automatically as required.
Yields:
vm_area_struct objects
"""
if not self.has_member("mmap"):
raise AttributeError(
"_get_mmap_iter called on mm_struct where no mmap member exists."
)
if not self.mmap:
vma_pointer = self.mmap
if not (vma_pointer and vma_pointer.is_readable()):
return None
yield self.mmap
vma_object = vma_pointer.dereference()
yield vma_object
seen = {self.mmap.vol.offset}
link = self.mmap.vm_next
seen = {vma_pointer}
vma_pointer = vma_pointer.vm_next
while link != 0 and link.vol.offset not in seen:
yield link
seen.add(link.vol.offset)
link = link.vm_next
while vma_pointer and vma_pointer.is_readable() and vma_pointer not in seen:
vma_object = vma_pointer.dereference()
yield vma_object
seen.add(vma_pointer)
vma_pointer = vma_pointer.vm_next
# TODO: As of version 3.0.0 this method should be removed
def get_maple_tree_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
@@ -839,7 +890,11 @@ class mm_struct(objects.StructType):
def _get_maple_tree_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns an iterator for the mm_mt member of an mm_struct. Use this only if
required, get_vma_iter() will choose the correct _get_maple_tree_iter() or
get_mmap_iter() automatically as required."""
get_mmap_iter() automatically as required.
Yields:
vm_area_struct objects
"""
if not self.has_member("mm_mt"):
raise AttributeError(
@@ -847,20 +902,27 @@ class mm_struct(objects.StructType):
)
symbol_table_name = self.get_symbol_table_name()
for vma_pointer in self.mm_mt.get_slot_iter():
# convert pointer to vm_area_struct and yield
vma = self._context.object(
# Convert pointer to vm_area_struct and yield
vma_object = self._context.object(
symbol_table_name + constants.BANG + "vm_area_struct",
layer_name=self.vol.native_layer_name,
offset=vma_pointer,
)
yield vma
yield vma_object
def get_vma_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns an iterator for the VMAs in an mm_struct. Automatically choosing the mmap or mm_mt as required."""
"""Returns an iterator for the VMAs in an mm_struct.
Automatically choosing the mmap or mm_mt as required.
Yields:
vm_area_struct objects
"""
if self.has_member("mmap"):
# kernels < 6.1
yield from self._get_mmap_iter()
elif self.has_member("mm_mt"):
# kernels >= 6.1 d4af56c5c7c6781ca6ca8075e2cf5bc119ed33d1
yield from self._get_maple_tree_iter()
else:
raise AttributeError("Unable to find mmap or mm_mt in mm_struct")
@@ -1206,35 +1268,43 @@ class list_head(objects.StructType, collections.abc.Iterable):
Objects of the type specified via the "symbol_type" argument.
"""
layer = layer or self.vol.layer_name
layer_name = layer or self.vol.layer_name
trans_layer = self._context.layers[layer_name]
if not trans_layer.is_valid(self.vol.offset):
return None
relative_offset = self._context.symbol_space.get_type(
symbol_type
).relative_child_offset(member)
direction = "prev"
if forward:
direction = "next"
try:
link = getattr(self, direction).dereference()
except exceptions.InvalidAddressException:
direction = "next" if forward else "prev"
link_ptr = getattr(self, direction)
if not (link_ptr and link_ptr.is_readable()):
return None
link = link_ptr.dereference()
if not sentinel:
yield self._context.object(
symbol_type, layer, offset=self.vol.offset - relative_offset
)
obj_offset = self.vol.offset - relative_offset
if not trans_layer.is_valid(obj_offset):
return None
yield self._context.object(symbol_type, layer_name, offset=obj_offset)
seen = {self.vol.offset}
while link.vol.offset not in seen:
obj = self._context.object(
symbol_type, layer, offset=link.vol.offset - relative_offset
)
yield obj
obj_offset = link.vol.offset - relative_offset
if not trans_layer.is_valid(obj_offset):
return None
yield self._context.object(symbol_type, layer_name, offset=obj_offset)
seen.add(link.vol.offset)
try:
link = getattr(link, direction).dereference()
except exceptions.InvalidAddressException:
link_ptr = getattr(link, direction)
if not (link_ptr and link_ptr.is_readable()):
break
link = link_ptr.dereference()
def __iter__(self) -> Iterator[interfaces.objects.ObjectInterface]:
return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)
@@ -962,56 +962,55 @@ class LIST_ENTRY(objects.StructType, collections.abc.Iterable):
) -> Iterator[interfaces.objects.ObjectInterface]:
"""Returns an iterator of the entries in the list."""
layer = layer or self.vol.layer_name
layer_name = layer or self.vol.layer_name
native_layer_name = layer_name or self.vol.native_layer_name
trans_layer = self._context.layers[layer_name]
if not trans_layer.is_valid(self.vol.offset):
return None
relative_offset = self._context.symbol_space.get_type(
symbol_type
).relative_child_offset(member)
direction = "Blink"
if forward:
direction = "Flink"
direction = "Flink" if forward else "Blink"
trans_layer = self._context.layers[layer]
try:
is_valid = trans_layer.is_valid(self.vol.offset)
if not is_valid:
return None
link = getattr(self, direction).dereference()
except exceptions.InvalidAddressException:
link_ptr = getattr(self, direction)
if not (link_ptr and link_ptr.is_readable()):
return None
link = link_ptr.dereference()
if not sentinel:
obj_offset = self.vol.offset - relative_offset
if not trans_layer.is_valid(obj_offset):
return None
yield self._context.object(
symbol_type,
layer,
offset=self.vol.offset - relative_offset,
native_layer_name=layer or self.vol.native_layer_name,
layer_name,
offset=obj_offset,
native_layer_name=native_layer_name,
)
seen = {self.vol.offset}
while link.vol.offset not in seen:
obj_offset = link.vol.offset - relative_offset
if not trans_layer.is_valid(obj_offset):
return None
obj = self._context.object(
yield self._context.object(
symbol_type,
layer,
layer_name,
offset=obj_offset,
native_layer_name=layer or self.vol.native_layer_name,
native_layer_name=native_layer_name,
)
yield obj
seen.add(link.vol.offset)
try:
link = getattr(link, direction).dereference()
except exceptions.InvalidAddressException:
link_ptr = getattr(link, direction)
if not (link_ptr and link_ptr.is_readable()):
return None
link = link_ptr.dereference()
def __iter__(self) -> Iterator[interfaces.objects.ObjectInterface]:
return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)
@@ -1,11 +1,11 @@
import contextlib
import logging
import struct
from typing import List, Iterator, Optional, Tuple, Type
from typing import Iterator, List, Optional, Tuple, Type
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist, printkey
vollog = logging.getLogger(__name__)
@@ -81,7 +81,11 @@ class Certificates(interfaces.plugins.PluginInterface):
"Microsoft\\SystemCertificates",
"Software\\Microsoft\\SystemCertificates",
]:
with contextlib.suppress(KeyError, exceptions.InvalidAddressException):
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
exceptions.InvalidAddressException,
):
# Walk it
node_path = hive.get_key(top_key, return_list=True)
for (
@@ -92,7 +96,11 @@ class Certificates(interfaces.plugins.PluginInterface):
_volatility,
node,
) in printkey.PrintKey.key_iterator(hive, node_path, recurse=True):
if not is_key and RegValueTypes(node.Type).name == "REG_BINARY":
if (
not is_key
and registry.RegValueTypes(node.Type)
== registry.RegValueTypes.REG_BINARY
):
name, certificate_data = self.parse_data(node.decode_data())
unique_key_offset = (
key_path.casefold().index(top_key.casefold())