WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-07 18:27:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into modxview_plugin

Browse Source
This commit is contained in:
ikelos
2025-01-03 00:26:53 +00:00
committed by GitHub
parent 89b8da8c39 cedc0b074a
commit 32a9d16f58
30 changed files with 950 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/simple-plugin.rst
+62 -36
View File
@@ -41,24 +41,36 @@ to be able to run properly. Any that are defined as optional need not necessari
@classmethod
def get_requirements(cls):
return [requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
optional = True),
requirements.PluginRequirement(name = 'pslist',
plugin = pslist.PsList,
version = (2, 0, 0))]
return [
requirements.ModuleRequirement(
name = 'kernel',
description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]
),
requirements.ListRequirement(
name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
optional = True
),
requirements.PluginRequirement(
name = 'pslist',
plugin = pslist.PsList,
version = (2, 0, 0)
),
]
This is a classmethod, because it is called before the specific plugin object has been instantiated (in order to know how
This is a classmethod, so it can be called before the specific plugin object has been instantiated (in order to know how
to instantiate the plugin). At the moment these requirements are fairly straightforward:
::
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ModuleRequirement(
name = 'kernel',
description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]
),
This requirement specifies the need for a particular submodule. Each module requires a
:py:class:`TranslationLayer <volatility3.framework.interfaces.layers.TranslationLayerInterface>` and a
@@ -85,9 +97,11 @@ not be requested directly from the user.
::
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"]
),
This requirement indicates that the plugin will operate on a single
:py:class:`TranslationLayer <volatility3.framework.interfaces.layers.TranslationLayerInterface>`. The name of the
@@ -110,8 +124,10 @@ not be requested directly from the user.
::
requirements.SymbolTableRequirement(name = "nt_symbols",
description = "Windows kernel symbols"),
requirements.SymbolTableRequirement(
name = "nt_symbols",
description = "Windows kernel symbols"
),
This requirement specifies the need for a particular
:py:class:`SymbolTable <volatility3.framework.interfaces.symbols.SymbolTableInterface>`
@@ -127,10 +143,12 @@ not be requested directly from the user.
::
requirements.ListRequirement(name = 'pid',
description = 'Filter on specific process IDs',
element_type = int,
optional = True),
requirements.ListRequirement(
name = 'pid',
description = 'Filter on specific process IDs',
element_type = int,
optional = True
),
The next requirement is a List Requirement, populated by integers. The description will be presented to the user to
describe what the value represents. The optional flag indicates that the plugin can function without the ``pid`` value
@@ -138,9 +156,11 @@ being defined within the configuration tree at all.
::
requirements.PluginRequirement(name = 'pslist',
plugin = pslist.PsList,
version = (2, 0, 0))]
requirements.PluginRequirement(
name = 'pslist',
plugin = pslist.PsList,
version = (2, 0, 0)
)
This requirement indicates that the plugin will make use of another plugin's code, and specifies the version requirements
on that plugin. The version is specified in terms of Semantic Versioning meaning that, to be compatible, the major
@@ -180,16 +200,24 @@ that will be output as part of the :py:class:`~volatility3.framework.interfaces.
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int),
("Process", str),
("Base", format_hints.Hex),
("Size", format_hints.Hex),
("Name", str),
("Path", str)],
self._generator(pslist.PsList.list_processes(self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func = filter_func)))
return renderers.TreeGrid(
[
("PID", int),
("Process", str),
("Base", format_hints.Hex),
("Size", format_hints.Hex),
("Name", str),
("Path", str),
],
self._generator(
pslist.PsList.list_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func = filter_func
)
)
)
In this instance, the plugin constructs a filter (using the PsList plugin's *classmethod* for creating filters).
It checks the plugin's configuration for the ``pid`` value, and passes it in as a list if it finds it, or None if
@@ -281,5 +309,3 @@ such as ``<table>!_UNICODE``) and the parameters to that type.
Since the cast value must populate a string typed column, it had to be a Python string (such as being cast to the native
type string) and could not have been a special Structure such as ``_UNICODE``. For the format hint columns, the format
hint type must be used to ensure the error checking does not fail.
pyproject.toml
+4
View File
@@ -20,6 +20,10 @@ full = [
"capstone>=5.0.3,<6",
"pycryptodome>=3.21.0,<4",
"leechcorepyc>=2.19.2,<3; sys_platform != 'darwin'",
# https://github.com/python-pillow/Pillow/blob/main/CHANGES.rst
# 10.0.0 dropped support for Python3.7
# 11.0.0 dropped support for Python3.8, which is still supported by Volatility3
"pillow>=10.0.0,<11.0.0",
]
cloud = [
volatility3/cli/text_filter.py
+4 -3
View File
@@ -1,7 +1,8 @@
import logging
from typing import Any, List, Optional
from volatility3.framework import constants, interfaces
import re
from typing import Any, List, Optional
from volatility3.framework import constants, interfaces
vollog = logging.getLogger(__name__)
@@ -67,8 +68,8 @@ class ColumnFilter:
) -> None:
self.column_num = column_num
self.pattern = pattern
self.exclude = exclude
self.regex = regex
self.exclude = exclude
def find(self, item) -> bool:
"""Identifies whether an item is found in the appropriate column"""
volatility3/framework/automagic/symbol_cache.py
+7
View File
@@ -301,6 +301,13 @@ class SqliteCache(CacheManagerInterface):
This also updates remote locations based on a cache timeout.
"""
if progress_callback is None:
def dummy_progress(*args, **kargs) -> None:
return None
progress_callback = dummy_progress
on_disk_locations = set(
[
filename
volatility3/framework/constants/_version.py
+1 -1
View File
@@ -1,6 +1,6 @@
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 14 # Number of changes that only add to the interface
VERSION_MINOR = 16 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
volatility3/framework/interfaces/configuration.py
+4 -5
View File
@@ -53,7 +53,7 @@ ConfigSimpleType = Optional[Union[SimpleTypes, List[SimpleTypes]]]
def path_join(*args) -> str:
"""Joins configuration paths together."""
# If a path element (particularly the first) is empty, then remove it from the list
args = tuple([arg for arg in args if arg])
args = tuple(arg for arg in args if arg)
return CONFIG_SEPARATOR.join(args)
@@ -772,17 +772,16 @@ class ConfigurableInterface(metaclass=ABCMeta):
str: The newly generated full configuration path
"""
random_config_dict = "".join(
random.SystemRandom().choice(string.ascii_uppercase + string.digits)
for _ in range(8)
random.SystemRandom().choices(string.ascii_uppercase + string.digits, k=8)
)
new_config_path = path_join(base_config_path, random_config_dict)
# TODO: Check that the new_config_path is empty, although it's not critical if it's not since the values are merged in
# This should check that each k corresponds to a requirement and each v is of the appropriate type
# This would require knowledge of the new configurable itself to verify, and they should do validation in the
# constructor anyway, however, to prevent bad types getting into the config tree we just verify that v is a simple type
# constructor anyway, however, to prevent bad types getting into the config tree we just verify that v is a basic type
for k, v in kwargs.items():
if not isinstance(v, (int, str, bool, float, bytes)):
if not isinstance(v, BasicTypes):
raise TypeError(
"Config values passed to make_subconfig can only be simple types"
)
volatility3/framework/layers/msf.py
+1 -1
View File
@@ -194,7 +194,7 @@ class PdbMSFStream(linear.LinearlyMappedLayer):
) -> None:
super().__init__(context, config_path, name, metadata)
self._base_layer = self.config["base_layer"]
self._pages = self.config.get("pages", None)
self._pages = self.config.get("pages", [])
self._pages_len = len(self._pages)
if not self._pages:
raise PDBFormatException(name, "Invalid/no pages specified")
volatility3/framework/layers/registry.py
+7 -1
View File
@@ -140,7 +140,13 @@ class RegistryHive(linear.LinearlyMappedLayer):
"""Returns the appropriate Node, interpreted from the Cell based on its
Signature."""
cell = self.get_cell(cell_offset)
signature = cell.cast("string", max_length=2, encoding="latin-1")
try:
signature = cell.cast("string", max_length=2, encoding="latin-1")
except (RegistryInvalidIndex, exceptions.InvalidAddressException):
vollog.debug(
f"Failed to get cell signature for cell (0x{cell.vol.offset:x})"
)
return cell
if signature == "nk":
return cell.u.KeyNode
elif signature == "sk":
volatility3/framework/layers/vmware.py
+4
View File
@@ -57,6 +57,10 @@ class VmwareLayer(segmented.SegmentedLayer):
)
meta_layer = self.context.layers.get(self._meta_layer, None)
if meta_layer is None:
raise exceptions.LayerException(
self._meta_layer, "VMware: Meta layer not found"
)
header_size = struct.calcsize(self.header_structure)
data = meta_layer.read(0, header_size)
magic, unknown, groupCount = struct.unpack(self.header_structure, data)
volatility3/framework/plugins/isfinfo.py
+1
View File
@@ -132,6 +132,7 @@ class IsfInfo(plugins.PluginInterface):
valid = check_valid(data)
except (UnicodeDecodeError, json.decoder.JSONDecodeError):
vollog.warning(f"Invalid ISF: {entry}")
continue
yield (
0,
(
volatility3/framework/plugins/linux/graphics/fbdev.py
+334
View File
@@ -0,0 +1,334 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
import io
from dataclasses import dataclass
from typing import Type, List, Dict, Tuple
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import (
format_hints,
TreeGrid,
NotAvailableValue,
UnreadableValue,
)
from volatility3.framework.objects import utility
from volatility3.framework.constants import architectures
from volatility3.framework.symbols import linux
# Image manipulation functions are kept in the plugin,
# to prevent a general exit on missing PIL (pillow) dependency.
try:
from PIL import Image
has_pil = True
except ImportError:
has_pil = False
vollog = logging.getLogger(__name__)
@dataclass
class Framebuffer:
"""Framebuffer object internal representation. This is useful to unify a framebuffer with precalculated
properties and pass it through functions conveniently."""
id: str
xres_virtual: int
yres_virtual: int
line_length: int
bpp: int
"""Bits Per Pixel"""
size: int
color_fields: Dict[str, Tuple[int, int, int]]
fb_info: interfaces.objects.ObjectInterface
class Fbdev(interfaces.plugins.PluginInterface):
"""Extract framebuffers from the fbdev graphics subsystem"""
_version = (1, 0, 0)
_required_framework_version = (2, 11, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 2, 0)
),
requirements.BooleanRequirement(
name="dump",
description="Dump framebuffers",
default=False,
optional=True,
),
]
@classmethod
def parse_fb_pixel_bitfields(
cls, fb_var_screeninfo: interfaces.objects.ObjectInterface
) -> Dict[str, Tuple[int, int, int]]:
"""Organize a framebuffer pixel format into a dictionary.
This is needed to know the position and bitlength of a color inside
a pixel.
Args:
fb_var_screeninfo: a fb_var_screeninfo kernel object instance
Returns:
The color fields mappings
Documentation:
include/uapi/linux/fb.h:
struct fb_bitfield {
__u32 offset; /* beginning of bitfield */
__u32 length; /* length of bitfield */
__u32 msb_right; /* != 0 : Most significant bit is right */
};
"""
# Naturally order by RGBA
color_mappings = [
("R", fb_var_screeninfo.red),
("G", fb_var_screeninfo.green),
("B", fb_var_screeninfo.blue),
("A", fb_var_screeninfo.transp),
]
color_fields = {}
for color_code, fb_bitfield in color_mappings:
color_fields[color_code] = (
int(fb_bitfield.offset),
int(fb_bitfield.length),
int(fb_bitfield.msb_right),
)
return color_fields
@classmethod
def convert_fb_raw_buffer_to_image(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
fb: Framebuffer,
):
"""Convert raw framebuffer pixels to an image.
Args:
fb: the relevant Framebuffer object
Returns:
A PIL Image object
Documentation:
include/uapi/linux/fb.h:
/* Interpretation of offset for color fields: All offsets are from the right,
* inside a "pixel" value, which is exactly 'bits_per_pixel' wide (means: you
* can use the offset as right argument to <<). A pixel afterwards is a bit
* stream and is written to video memory as that unmodified.
"""
kernel = context.modules[kernel_name]
kernel_layer = context.layers[kernel.layer_name]
raw_pixels = io.BytesIO(kernel_layer.read(fb.fb_info.screen_base, fb.size))
bytes_per_pixel = fb.bpp // 8
image = Image.new("RGBA", (fb.xres_virtual, fb.yres_virtual))
# This is not designed to be extremely fast (numpy isn't available),
# but convenient and dynamic for any color field layout.
for y in range(fb.yres_virtual):
for x in range(fb.xres_virtual):
raw_pixel = int.from_bytes(raw_pixels.read(bytes_per_pixel), "little")
pixel = [0, 0, 0, 255]
# The framebuffer is expected to have been correctly constructed,
# especially by parse_fb_pixel_bitfields, to get the needed RGBA mappings.
for i, color_code in enumerate(["R", "G", "B", "A"]):
offset, length, msb_right = fb.color_fields[color_code]
if length == 0:
continue
color_value = (raw_pixel >> offset) & (2**length - 1)
if msb_right:
# Reverse bit order
color_value = int(
"{:0{length}b}".format(color_value, length=length)[::-1], 2
)
pixel[i] = color_value
image.putpixel((x, y), tuple(pixel))
return image
@classmethod
def dump_fb(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
open_method: Type[interfaces.plugins.FileHandlerInterface],
fb: Framebuffer,
convert_to_png_image: bool,
) -> str:
"""Dump a Framebuffer buffer to disk.
Args:
fb: the relevant Framebuffer object
convert_to_image: a boolean specifying if the buffer should be converted to an image
Returns:
The filename of the dumped buffer.
"""
kernel = context.modules[kernel_name]
kernel_layer = context.layers[kernel.layer_name]
id = "N-A" if isinstance(fb.id, NotAvailableValue) else fb.id
base_filename = f"{id}_{fb.xres_virtual}x{fb.yres_virtual}_{fb.bpp}bpp"
if convert_to_png_image:
image_object = cls.convert_fb_raw_buffer_to_image(context, kernel_name, fb)
raw_io_output = io.BytesIO()
image_object.save(raw_io_output, "PNG")
final_fb_buffer = raw_io_output.getvalue()
filename = f"{base_filename}.png"
else:
final_fb_buffer = kernel_layer.read(fb.fb_info.screen_base, fb.size)
filename = f"{base_filename}.raw"
with open_method(filename) as f:
f.write(final_fb_buffer)
return f.preferred_filename
@classmethod
def parse_fb_info(
cls,
fb_info: interfaces.objects.ObjectInterface,
) -> Framebuffer:
"""Parse an fb_info struct
Args:
fb_info: an fb_info kernel object live instance
Returns:
A Framebuffer object
Documentation:
https://docs.kernel.org/fb/api.html:
- struct fb_fix_screeninfo stores device independent unchangeable information about the frame buffer device and the current format.
Those information can't be directly modified by applications, but can be changed by the driver when an application modifies the format.
- struct fb_var_screeninfo stores device independent changeable information about a frame buffer device, its current format and video mode,
as well as other miscellaneous parameters.
"""
id = utility.array_to_string(fb_info.fix.id) or NotAvailableValue()
color_fields = None
# 0 = color, 1 = grayscale, >1 = FOURCC
if fb_info.var.grayscale in [0, 1]:
color_fields = cls.parse_fb_pixel_bitfields(fb_info.var)
# There a lot of tricky pixel formats used by drivers and vendors in include/uapi/linux/videodev2.h.
# As Volatility3 is not a video format converter, it is best to play it safe and let the user parse
# the raw data manually (with ffmpeg for example).
elif fb_info.var.grayscale > 1:
fourcc = linux.LinuxUtilities.convert_fourcc_code(fb_info.var.grayscale)
warn_msg = f"""Framebuffer "{id}" uses a FOURCC pixel format "{fourcc}" that isn't natively supported.
You can try using ffmpeg to decode the raw buffer. Example usage:
"ffmpeg -pix_fmts" to list supported formats, then
"ffmpeg -f rawvideo -video_size {fb_info.var.xres_virtual}x{fb_info.var.yres_virtual} -i <FILENAME>.raw -pix_fmt <FORMAT> output.png"."""
vollog.warning(warn_msg)
# Prefer using the virtual resolution, instead of the visible one.
# This prevents missing non-visible data stored in the framebuffer.
fb = Framebuffer(
id,
xres_virtual=fb_info.var.xres_virtual,
yres_virtual=fb_info.var.yres_virtual,
line_length=fb_info.fix.line_length,
bpp=fb_info.var.bits_per_pixel,
size=fb_info.var.yres_virtual * fb_info.fix.line_length,
color_fields=color_fields,
fb_info=fb_info,
)
return fb
def _generator(self):
if not has_pil:
vollog.error(
"PIL (pillow) module is required to use this plugin. Please install it manually or through pyproject.toml."
)
return None
kernel_name = self.config["kernel"]
kernel = self.context.modules[kernel_name]
if not kernel.has_symbol("num_registered_fb"):
raise exceptions.SymbolError(
"num_registered_fb",
kernel.symbol_table_name,
"The provided symbol does not exist in the symbol table. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt.",
)
num_registered_fb = kernel.object_from_symbol("num_registered_fb")
if num_registered_fb < 1:
vollog.info("No registered framebuffer in the fbdev API.")
return None
registered_fb = kernel.object_from_symbol("registered_fb")
fb_info_list = utility.array_of_pointers(
registered_fb,
num_registered_fb,
kernel.symbol_table_name + constants.BANG + "fb_info",
self.context,
)
for fb_info in fb_info_list:
fb = self.parse_fb_info(fb_info)
file_output = "Disabled"
if self.config["dump"]:
try:
file_output = self.dump_fb(
self.context, kernel_name, self.open, fb, bool(fb.color_fields)
)
file_output = str(file_output)
except exceptions.InvalidAddressException as excp:
vollog.error(
f'Layer {excp.layer_name} failed to read address {hex(excp.invalid_address)} when dumping framebuffer "{fb.id}".'
)
file_output = UnreadableValue()
try:
fb_device_name = utility.pointer_to_string(
fb.fb_info.dev.kobj.name, 256
)
except exceptions.InvalidAddressException:
fb_device_name = NotAvailableValue()
yield (
0,
(
format_hints.Hex(fb.fb_info.screen_base),
fb_device_name,
fb.id,
fb.size,
f"{fb.xres_virtual}x{fb.yres_virtual}",
fb.bpp,
"RUNNING" if fb.fb_info.state == 0 else "SUSPENDED",
file_output,
),
)
def run(self):
columns = [
("Address", format_hints.Hex),
("Device", str),
("ID", str),
("Size", int),
("Virtual resolution", str),
("BPP", int),
("State", str),
("Filename", str),
]
return TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/kmsg.py
+21 -17
View File
@@ -149,7 +149,7 @@ class ABCKmsg(ABC):
# This might seem insignificant but it could cause some issues
# when compared with userland tool results or when used in
# timelines.
return f"{nsec / 1000000000:lu}.{(nsec % 1000000000) / 1000:06lu}"
return f"{nsec // 1000000000}.{(nsec % 1000000000) // 1000:06}"
def get_timestamp_in_sec_str(self, obj) -> str:
# obj could be log, printk_log or printk_info
@@ -166,7 +166,7 @@ class ABCKmsg(ABC):
def get_caller_text(self, caller_id):
caller_name = "CPU" if caller_id & 0x80000000 else "Task"
caller = f"{caller_name}({caller_id & ~0x80000000:u})"
caller = f"{caller_name}({caller_id & ~0x80000000})"
return caller
def get_prefix(self, obj) -> Tuple[int, int, str, str]:
@@ -317,23 +317,27 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
while cur_idx < end_idx:
msg_offset = log_buf_ptr + cur_idx # type: ignore
msg = self.vmlinux.object(object_type=log_struct_name, offset=msg_offset)
if msg.len == 0:
# As per kernel/printk.c:
# A length == 0 for the next message indicates a wrap-around to
# the beginning of the buffer.
cur_idx = 0
end_idx = log_next_idx
else:
facility, level, timestamp, caller = self.get_prefix(msg)
level_txt = self.get_level_text(level)
facility_txt = self.get_facility_text(facility)
try:
if msg.len == 0:
# As per kernel/printk.c:
# A length == 0 for the next message indicates a wrap-around to
# the beginning of the buffer.
cur_idx = 0
end_idx = log_next_idx
else:
facility, level, timestamp, caller = self.get_prefix(msg)
level_txt = self.get_level_text(level)
facility_txt = self.get_facility_text(facility)
for line in self.get_log_lines(msg):
yield facility_txt, level_txt, timestamp, caller, line
for line in self.get_dict_lines(msg):
yield facility_txt, level_txt, timestamp, caller, line
for line in self.get_log_lines(msg):
yield facility_txt, level_txt, timestamp, caller, line
for line in self.get_dict_lines(msg):
yield facility_txt, level_txt, timestamp, caller, line
cur_idx += msg.len
cur_idx += msg.len
except exceptions.InvalidAddressException:
vollog.warning("Kmsg buffer msg length could not be read")
return
class Kmsg_3_11_to_5_10(Kmsg_3_5_to_3_11):
volatility3/framework/plugins/windows/amcache.py
+5 -5
View File
@@ -543,7 +543,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\InventoryDriverBinary") # type: ignore
)
)
except KeyError:
except (KeyError, registry.RegistryFormatException):
# Registry key not found
pass
@@ -554,7 +554,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\Programs")
) # type: ignore
}
except KeyError:
except (KeyError, registry.RegistryFormatException):
programs = {}
try:
@@ -564,7 +564,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
key=_entry_sort_key,
)
except KeyError:
except (KeyError, registry.RegistryFormatException):
files = []
for program_id, file_entries in itertools.groupby(
@@ -593,7 +593,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\InventoryApplication") # type: ignore
)
)
except KeyError:
except (KeyError, registry.RegistryFormatException):
programs = {}
try:
@@ -603,7 +603,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
key=_entry_sort_key,
)
except KeyError:
except (KeyError, registry.RegistryFormatException):
files = []
for program_id, file_entries in itertools.groupby(
volatility3/framework/plugins/windows/cachedump.py
+8 -3
View File
@@ -8,7 +8,7 @@ from typing import Tuple
from Crypto.Cipher import ARC4, AES
from Crypto.Hash import HMAC
from volatility3.framework import interfaces, renderers
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.framework.symbols.windows import versions
@@ -140,9 +140,14 @@ class Cachedump(interfaces.plugins.PluginInterface):
if cache_item.Name == "NL$Control":
continue
data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
if data is None:
try:
data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
except exceptions.InvalidAddressException:
continue
if not data:
continue
(
uname_len,
domain_len,
volatility3/framework/plugins/windows/direct_system_calls.py
+2
View File
@@ -433,6 +433,8 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
proc_layer = self.context.layers[proc_layer_name]
vads = self.get_vad_maps(proc)
if not vads:
continue
# for each valid process, look for malicious syscall invocations
for address, vad_path in self._get_rule_hits(
volatility3/framework/plugins/windows/handles.py
+33 -7
View File
@@ -68,7 +68,12 @@ class Handles(interfaces.plugins.PluginInterface):
if not self.context.layers[virtual].is_valid(handle_table_entry.Object):
return None
fast_ref = handle_table_entry.Object.cast("_EX_FAST_REF")
object_header = fast_ref.dereference().cast("_OBJECT_HEADER")
try:
object_header = fast_ref.dereference().cast("_OBJECT_HEADER")
except exceptions.InvalidAddressException:
return None
object_header.GrantedAccess = handle_table_entry.GrantedAccess
except AttributeError:
# starting with windows 8
@@ -77,16 +82,26 @@ class Handles(interfaces.plugins.PluginInterface):
)
if is_64bit:
if handle_table_entry.ObjectPointerBits == 0:
try:
pointer_bits = handle_table_entry.ObjectPointerBits
except exceptions.InvalidAddressException:
return None
offset = handle_table_entry.ObjectPointerBits << 4
if pointer_bits == 0:
return None
offset = pointer_bits << 4
else:
if handle_table_entry.InfoTable == 0:
try:
info_table = handle_table_entry.InfoTable
except exceptions.InvalidAddressException:
return None
offset = handle_table_entry.InfoTable & ~7
if info_table == 0:
return None
offset = info_table & ~7
# print("LowValue: {0:#x} Magic: {1:#x} Offset: {2:#x}".format(handle_table_entry.InfoTable, magic, offset))
object_header = self.context.object(
@@ -94,7 +109,10 @@ class Handles(interfaces.plugins.PluginInterface):
virtual,
offset=offset,
)
object_header.GrantedAccess = handle_table_entry.GrantedAccessBits
try:
object_header.GrantedAccess = handle_table_entry.GrantedAccessBits
except exceptions.InvalidAddressException:
return None
object_header.HandleValue = handle_value
return object_header
@@ -160,7 +178,7 @@ class Handles(interfaces.plugins.PluginInterface):
except exceptions.InvalidAddressException:
vollog.log(
constants.LOGLEVEL_VVV,
f"Cannot access _OBJECT_HEADER Name at {objt.vol.offset:#x}",
f"Cannot access _OBJECT_HEADER Name at {ptr.vol.offset:#x}",
)
continue
@@ -226,6 +244,14 @@ class Handles(interfaces.plugins.PluginInterface):
masked_offset = offset & layer_object.maximum_address
for entry in table:
# This triggered a backtrace in many testing samples
# in the level == 0 path
# The code above this calls `is_valid` on the `offset`
# It is sent but then does not validate `entry` before
# sending it to `_get_item`
if not self.context.layers[virtual].is_valid(entry.vol.offset):
continue
if level > 0:
yield from self._make_handle_array(entry, level - 1, depth)
depth += 1
volatility3/framework/plugins/windows/hashdump.py
+1 -1
View File
@@ -332,7 +332,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
try:
if hive:
result = hive.get_key(key)
except KeyError:
except (KeyError, registry.RegistryFormatException):
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
)
volatility3/framework/plugins/windows/lsadump.py
+5 -2
View File
@@ -8,7 +8,7 @@ from typing import Optional
from Crypto.Cipher import ARC4, DES, AES
from Crypto.Hash import MD5, SHA256
from volatility3.framework import interfaces, renderers
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.framework.symbols.windows import versions
@@ -81,7 +81,10 @@ class Lsadump(interfaces.plugins.PluginInterface):
if not enc_reg_value:
return None
obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength)
try:
obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength)
except exceptions.InvalidAddressException:
return None
if not obf_lsa_key:
return None
volatility3/framework/plugins/windows/netstat.py
+113 -44
View File
@@ -111,8 +111,21 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
The list of indices at which a 1 was found.
"""
ret = []
# This value is broken in many samples and was causing essentially infinite loops
# Testing showed that 8192 is the current size across all Windows versions
# We give some leeway in case it increases in later versions, while still keeping it sane
# The problematic samples had values that looked like addresses, so in the billions
if bitmap_size_in_byte > 8192 * 10:
return ret
for idx in range(bitmap_size_in_byte):
current_byte = context.layers[layer_name].read(bitmap_offset + idx, 1)[0]
try:
current_byte = context.layers[layer_name].read(bitmap_offset + idx, 1)[
0
]
except exceptions.InvalidAddressException:
continue
current_offs = idx * 8
for bit in range(8):
if current_byte & (1 << bit) != 0:
@@ -154,32 +167,37 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
)
else:
# invalid argument.
return None
return
vollog.debug(f"Current Port: {port}")
# the given port serves as a shifted index into the port pool lists
list_index = port >> 8
truncated_port = port & 0xFF
# constructing port_pool object here so callers don't have to
port_pool = context.object(
net_symbol_table + constants.BANG + "_INET_PORT_POOL",
layer_name=layer_name,
offset=port_pool_addr,
)
try:
# constructing port_pool object here so callers don't have to
port_pool = context.object(
net_symbol_table + constants.BANG + "_INET_PORT_POOL",
layer_name=layer_name,
offset=port_pool_addr,
)
# first, grab the given port's PortAssignment (`_PORT_ASSIGNMENT`)
inpa = port_pool.PortAssignments[list_index]
# first, grab the given port's PortAssignment (`_PORT_ASSIGNMENT`)
inpa = port_pool.PortAssignments[list_index]
# then parse the port assignment list (`_PORT_ASSIGNMENT_LIST`) and grab the correct entry
assignment = inpa.InPaBigPoolBase.Assignments[truncated_port]
# then parse the port assignment list (`_PORT_ASSIGNMENT_LIST`) and grab the correct entry
assignment = inpa.InPaBigPoolBase.Assignments[truncated_port]
except exceptions.InvalidAddressException:
return
if not assignment:
return None
return
# the value within assignment.Entry is a) masked and b) points inside of the network object
# first decode the pointer
netw_inside = cls._decode_pointer(assignment.Entry)
try:
netw_inside = cls._decode_pointer(assignment.Entry)
except exceptions.InvalidAddressException:
return
if netw_inside:
# if the value is valid, calculate the actual object address by subtracting the offset
@@ -188,16 +206,30 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
)
yield curr_obj
try:
next_obj_address = cls._decode_pointer(curr_obj.Next)
except exceptions.InvalidAddressException:
return
# if the same port is used on different interfaces multiple objects are created
# those can be found by following the pointer within the object's `Next` field until it is empty
while curr_obj.Next:
curr_obj = context.object(
obj_name,
layer_name=layer_name,
offset=cls._decode_pointer(curr_obj.Next) - ptr_offset,
)
while next_obj_address:
try:
curr_obj = context.object(
obj_name,
layer_name=layer_name,
offset=next_obj_address - ptr_offset,
)
except exceptions.InvalidAddressException:
return
yield curr_obj
try:
next_obj_address = cls._decode_pointer(curr_obj.Next)
except exceptions.InvalidAddressException:
return
@classmethod
def get_tcpip_module(
cls,
@@ -243,16 +275,25 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
The hash table entries which are _not_ empty
"""
# we are looking for entries whose values are not their own address
# smear sanity check from mass testing
if ht_length > 4096:
return
for index in range(ht_length):
current_addr = ht_offset + index * alignment
current_pointer = context.object(
net_symbol_table + constants.BANG + "pointer",
layer_name=layer_name,
offset=current_addr,
)
try:
current_pointer = context.object(
net_symbol_table + constants.BANG + "pointer",
layer_name=layer_name,
offset=current_addr,
)
except exceptions.InvalidAddressException:
continue
# check if addr of pointer is equal to the value pointed to
if current_pointer.vol.offset == current_pointer:
continue
yield current_pointer
@classmethod
@@ -292,11 +333,15 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
tcpip_symbol_table + constants.BANG + "PartitionCount"
).address
part_table_addr = context.object(
net_symbol_table + constants.BANG + "pointer",
layer_name=layer_name,
offset=tcpip_module_offset + part_table_symbol,
)
try:
part_table_addr = context.object(
net_symbol_table + constants.BANG + "pointer",
layer_name=layer_name,
offset=tcpip_module_offset + part_table_symbol,
)
except exceptions.InvalidAddressException:
vollog.debug("`PartitionTable` not present in memory.")
return
# part_table is the actual partition table offset and consists out of a dynamic amount of _PARTITION objects
part_table = context.object(
@@ -304,10 +349,18 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
layer_name=layer_name,
offset=part_table_addr,
)
part_count = int.from_bytes(
context.layers[layer_name].read(tcpip_module_offset + part_count_symbol, 1),
"little",
)
try:
part_count = int.from_bytes(
context.layers[layer_name].read(
tcpip_module_offset + part_count_symbol, 1
),
"little",
)
except exceptions.InvalidAddressException:
vollog.debug("`PartitionCount` not present in memory.")
return
part_table.Partitions.count = part_count
vollog.debug(
@@ -316,9 +369,21 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
entry_offset = context.symbol_space.get_type(obj_name).relative_child_offset(
"ListEntry"
)
for ctr, partition in enumerate(part_table.Partitions):
try:
partitions = part_table.Partitions
except exceptions.InvalidAddressException:
vollog.debug("Partitions member not present in memory")
return
for ctr, partition in enumerate(partitions):
vollog.debug(f"Parsing partition {ctr}")
if partition.Endpoints.NumEntries > 0:
try:
num_entries = partition.Endpoints.NumEntries
except exceptions.InvalidAddressException:
continue
if num_entries > 0:
for endpoint_entry in cls.parse_hashtable(
context,
layer_name,
@@ -402,6 +467,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
upp_symbol = context.symbol_space.get_symbol(
tcpip_symbol_table + constants.BANG + "UdpPortPool"
).address
upp_addr = context.object(
net_symbol_table + constants.BANG + "pointer",
layer_name=layer_name,
@@ -498,13 +564,16 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# then, towards the UDP and TCP port pools
# first, find their addresses
upp_addr, tpp_addr = cls.find_port_pools(
context,
layer_name,
net_symbol_table,
tcpip_symbol_table,
tcpip_module_offset,
)
try:
upp_addr, tpp_addr = cls.find_port_pools(
context,
layer_name,
net_symbol_table,
tcpip_symbol_table,
tcpip_module_offset,
)
except (exceptions.SymbolError, exceptions.InvalidAddressException):
vollog.debug("Unable to reconstruct port pools")
# create port pool objects at the detected address and parse the port bitmap
upp_obj = context.object(
volatility3/framework/plugins/windows/skeleton_key_check.py
+14 -10
View File
@@ -282,14 +282,13 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
for proc in proc_list:
try:
proc_id = proc.UniqueProcessId
proc_layer_name = proc.add_process_layer()
return proc, proc_layer_name
except exceptions.InvalidAddressException as excp:
vollog.debug(
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
f"Invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None, None
@@ -431,15 +430,20 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
# we do not want to fail just because the count is not in memory
# 16 was the size on samples I tested, so I chose it as the default
count = 16
if target_address:
count = int.from_bytes(
self.context.layers[proc_layer_name].read(
target_address, 4
),
"little",
)
else:
count = 16
try:
count = int.from_bytes(
self.context.layers[proc_layer_name].read(
target_address, 4
),
"little",
)
except exceptions.InvalidAddressException:
vollog.debug(
"Unable to read `cCsystems`. Defaulting to 16."
)
found_count = True
volatility3/framework/plugins/windows/suspended_threads.py
+148
View File
@@ -0,0 +1,148 @@
import logging
from typing import Dict
import functools
from volatility3.framework import renderers, interfaces, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
import volatility3.plugins.windows.pslist as pslist
import volatility3.plugins.windows.threads as threads
import volatility3.plugins.windows.pe_symbols as pe_symbols
from volatility3.framework.objects import utility
vollog = logging.getLogger(__name__)
class SuspendedThreads(interfaces.plugins.PluginInterface):
"""Enumerates suspended threads."""
_required_framework_version = (2, 13, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0)
),
requirements.VersionRequirement(
name="threads", component=threads.Threads, version=(1, 0, 0)
),
]
def _generator(self):
"""
The goal of this plugin is to report on threads that are suspended
Legitimate programs can start threads suspended but then will later resume them
Subsets of malware techniques, such as EDR evasion and process hollowing,
create suspended threads and do not resume them. These are the threads that this
plugin is designed to catch.
See the whitepaper from our DEF CON 2024 presentation for more details:
https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf
"""
kernel = self.context.modules[self.config["kernel"]]
vads_cache: Dict[int, pe_symbols.PESymbols.ranges_type] = {}
proc_modules = None
# walk the threads of each process checking for suspended threads
for proc in pslist.PsList.list_processes(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
):
for thread in threads.Threads.list_threads(kernel, proc):
try:
# we only care if the thread is suspended
if thread.Tcb.SuspendCount == 0:
continue
# 4 == terminated
if thread.Tcb.State == 4:
continue
owner_proc = thread.owning_process()
owner_proc_pid = thread.Cid.UniqueProcess
owner_proc_name = utility.array_to_string(owner_proc.ImageFileName)
thread_tid = thread.Cid.UniqueThread
thread_start_addr = thread.StartAddress
thread_win32_addr = thread.Win32StartAddress
except exceptions.InvalidAddressException:
continue
# Nothing useful to report if a process doesn't have VADs.. Also a sign of smear/terminated
vads = pe_symbols.PESymbols.get_vads_for_process_cache(
vads_cache, owner_proc
)
if not vads:
continue
# Only compute this if needed as its expensive and 99.9% of samples
# will not have suspended threads
if not proc_modules:
proc_modules = pe_symbols.PESymbols.get_process_modules(
self.context, kernel.layer_name, kernel.symbol_table_name, None
)
path_and_symbol = functools.partial(
pe_symbols.PESymbols.path_and_symbol_for_address,
self.context,
self.config_path,
proc_modules,
)
start_file, start_sym = path_and_symbol(vads, thread_start_addr)
win32_file, win32_sym = path_and_symbol(vads, thread_win32_addr)
# the only false positive found in mass scanning of samples
if start_file and start_file.endswith("\\WorkFoldersShell.dll"):
continue
if win32_file and win32_file.endswith("\\WorkFoldersShell.dll"):
continue
yield (
0,
(
owner_proc_name,
owner_proc_pid,
thread_tid,
start_file or renderers.NotAvailableValue(),
start_sym or renderers.NotAvailableValue(),
format_hints.Hex(thread_start_addr),
win32_file or renderers.NotAvailableValue(),
win32_sym or renderers.NotAvailableValue(),
format_hints.Hex(thread_win32_addr),
),
)
def run(self):
return renderers.TreeGrid(
[
("Process", str),
("PID", int),
("TID", int),
("StartFile", str),
("StartSymbol", str),
("StartAddress", format_hints.Hex),
("Win32StartFile", str),
("Win32StartSymbol", str),
("Win32StartAddress", format_hints.Hex),
],
self._generator(),
)
volatility3/framework/renderers/__init__.py
+2 -4
View File
@@ -83,8 +83,7 @@ class TreeNode(interfaces.renderers.TreeNode):
raise TypeError(
"Values must be a list of objects made up of simple types and number the same as the columns"
)
for index in range(len(self._treegrid.columns)):
column = self._treegrid.columns[index]
for index, column in enumerate(self._treegrid.columns):
val = values[index]
if not isinstance(val, (column.type, interfaces.renderers.BaseAbsentValue)):
raise TypeError(
@@ -413,8 +412,7 @@ class ColumnSortKey(interfaces.renderers.ColumnSortKey):
_index = None
self._type = None
self.ascending = ascending
for i in range(len(treegrid.columns)):
column = treegrid.columns[i]
for i, column in enumerate(treegrid.columns):
if column.name.lower() == column_name.lower():
_index = i
self._type = column.type
volatility3/framework/symbols/intermed.py
+4 -4
View File
@@ -101,7 +101,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
Args:
context: The volatility context for the symbol table
config_path: The configuration path for the symbol table
name: The name for the symbol table (this is used in symbols e.g. table!symbol )
name: The name for the symbol table (this is used in symbols e.g. table!symbol)
isf_url: The URL pointing to the ISF file location
native_types: The NativeSymbolTable that contains the native types for this symbol table
table_mapping: A dictionary linking names referenced in the file with symbol tables in the context
@@ -111,7 +111,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
"""
# Check there are no obvious errors
# Open the file and test the version
self._versions = dict([(x.version, x) for x in class_subclasses(ISFormatTable)])
self._versions = dict((x.version, x) for x in class_subclasses(ISFormatTable))
with resources.ResourceAccessor().open(isf_url) as fp:
reader = codecs.getreader("utf-8")
json_object = json.load(reader(fp)) # type: ignore
@@ -166,9 +166,9 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
format.
An interface version such as Major.Minor.Patch means that Major
of the provider must be equal to that of the consumer, and the
of the provider must be equal to that of the consumer, and the
provider (the JSON in this instance) must have a greater minor
(indicating that only additive changes have been made) than
(indicating that only additive changes have been made) than
the consumer (in this case, the file reader).
"""
major, minor, patch = (int(x) for x in version.split("."))
volatility3/framework/symbols/linux/__init__.py
+17 -1
View File
@@ -76,7 +76,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
class LinuxUtilities(interfaces.configuration.VersionableInterface):
"""Class with multiple useful linux functions."""
_version = (2, 1, 1)
_version = (2, 2, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@@ -483,6 +483,22 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
return kernel
@classmethod
def convert_fourcc_code(cls, code: int) -> str:
"""Convert a fourcc integer back to its fourcc string representation.
Args:
code: the numerical representation of the fourcc
Returns:
The fourcc code string.
"""
code_bytes_length = (code.bit_length() + 7) // 8
return "".join(
[chr((code >> (i * 8)) & 0xFF) for i in range(code_bytes_length)]
)
class IDStorage(ABC):
"""Abstraction to support both XArray and RadixTree"""
volatility3/framework/symbols/linux/extensions/__init__.py
+51 -24
View File
@@ -15,7 +15,7 @@ from typing import Generator, Iterable, Iterator, Optional, Tuple, List, Union,
from volatility3.framework import constants, exceptions, objects, interfaces, symbols
from volatility3.framework.renderers import conversion
from volatility3.framework.constants import linux as linux_constants
from volatility3.framework.layers import linear
from volatility3.framework.layers import linear, intel
from volatility3.framework.objects import utility
from volatility3.framework.symbols import generic, linux, intermed
from volatility3.framework.symbols.linux.extensions import elf
@@ -2549,16 +2549,13 @@ class address_space(objects.StructType):
class page(objects.StructType):
@property
@functools.lru_cache
@functools.cached_property
def pageflags_enum(self) -> Dict:
"""Returns 'pageflags' enumeration key/values
Returns:
A dictionary with the pageflags enumeration key/values
"""
# FIXME: It would be even better to use @functools.cached_property instead,
# however, this requires Python +3.8
try:
pageflags_enum = self._context.symbol_space.get_enumeration(
self.get_symbol_table_name() + constants.BANG + "pageflags"
@@ -2572,24 +2569,12 @@ class page(objects.StructType):
return pageflags_enum
def get_flags_list(self) -> List[str]:
"""Returns a list of page flags
@functools.cached_property
def _intel_vmemmap_start(self) -> int:
"""Determine the start of the struct page array, for Intel systems.
Returns:
List of page flags
"""
flags = []
for name, value in self.pageflags_enum.items():
if self.flags & (1 << value) != 0:
flags.append(name)
return flags
def to_paddr(self) -> int:
"""Converts a page's virtual address to its physical address using the current physical memory model.
Returns:
int: page physical address
int: vmemmap_start address
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
@@ -2629,13 +2614,39 @@ class page(objects.StructType):
"Something went wrong, we shouldn't be here"
)
page_type_size = vmlinux.get_type("page").size
return vmemmap_start
def _intel_to_paddr(self) -> int:
"""Converts a page's virtual address to its physical address using the current Intel memory model.
Returns:
int: page physical address
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
pagec = vmlinux_layer.canonicalize(self.vol.offset)
pfn = (pagec - vmemmap_start) // page_type_size
pfn = (pagec - self._intel_vmemmap_start) // vmlinux.get_type("page").size
page_paddr = pfn * vmlinux_layer.page_size
return page_paddr
def to_paddr(self) -> int:
"""Converts a page's virtual address to its physical address using the current CPU memory model.
Returns:
int: page physical address
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
if isinstance(vmlinux_layer, intel.Intel):
page_paddr = self._intel_to_paddr()
else:
raise exceptions.LayerException(
f"Architecture {type(vmlinux_layer)} vmemmap_start calculation isn't currently supported."
)
return page_paddr
def get_content(self) -> Union[str, None]:
"""Returns the page content
@@ -2644,7 +2655,10 @@ class page(objects.StructType):
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
physical_layer = vmlinux.context.layers["memory_layer"]
physical_layer_name = self._context.layers[self.vol.layer_name].config.get(
"memory_layer", self.vol.layer_name
)
physical_layer = self._context.layers[physical_layer_name]
page_paddr = self.to_paddr()
if not page_paddr:
return None
@@ -2652,6 +2666,19 @@ class page(objects.StructType):
page_data = physical_layer.read(page_paddr, vmlinux_layer.page_size)
return page_data
def get_flags_list(self) -> List[str]:
"""Returns a list of page flags
Returns:
List of page flags
"""
flags = []
for name, value in self.pageflags_enum.items():
if self.flags & (1 << value) != 0:
flags.append(name)
return flags
class IDR(objects.StructType):
IDR_BITS = 8
volatility3/framework/symbols/windows/extensions/__init__.py
+34 -5
View File
@@ -405,10 +405,24 @@ class DEVICE_OBJECT(objects.StructType, pool.ExecutiveObject):
def get_attached_devices(self) -> Generator[ObjectInterface, None, None]:
"""Enumerate the attached device's objects"""
device = self.AttachedDevice.dereference()
seen = set()
try:
device = self.AttachedDevice.dereference()
except exceptions.InvalidAddressException:
return
while device:
if device.vol.offset in seen:
break
seen.add(device.vol.offset)
yield device
device = device.AttachedDevice.dereference()
try:
device = device.AttachedDevice.dereference()
except exceptions.InvalidAddressException:
return
class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
@@ -421,10 +435,24 @@ class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
def get_devices(self) -> Generator[ObjectInterface, None, None]:
"""Enumerate the driver's device objects"""
device = self.DeviceObject.dereference()
seen = set()
try:
device = self.DeviceObject.dereference()
except exceptions.InvalidAddressException:
return
while device:
if device.vol.offset in seen:
return
seen.add(device.vol.offset)
yield device
device = device.NextDevice.dereference()
try:
device = device.NextDevice.dereference()
except exceptions.InvalidAddressException:
return
def is_valid(self) -> bool:
"""Determine if the object is valid."""
@@ -519,7 +547,8 @@ class ETHREAD(objects.StructType, pool.ExecutiveObject):
if not isinstance(ctime, datetime.datetime):
return False
if not (1998 < ctime.year < 2030):
current_year = datetime.datetime.now().year
if not (1998 < ctime.year < current_year + 10):
return False
except exceptions.InvalidAddressException:
volatility3/framework/symbols/windows/extensions/network.py
+7 -1
View File
@@ -219,7 +219,13 @@ class _TCP_ENDPOINT(_TCP_LISTENER):
return None
def is_valid(self):
if self.State not in self.State.choices.values():
# netstat calls this before validating the object itself
try:
state = self.State
except exceptions.InvalidAddressException:
return False
if state not in state.choices.values():
vollog.debug(
f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid tcp state {self.State}"
)
volatility3/framework/symbols/windows/extensions/pool.py
+10 -1
View File
@@ -376,7 +376,16 @@ class OBJECT_HEADER(objects.StructType):
try:
# vista and earlier have a Type member
self._vol["object_header_object_type"] = self.Type.Name.String
length = self.Type.member("Name").Length
if length == 0 or length > 128:
string = None
else:
string = self.Type.Name.String
if len(string) == 0 or len(string) > 128:
string = None
self._vol["object_header_object_type"] = string
except AttributeError:
# windows 7 and later have a TypeIndex, but windows 10
# further encodes the index value with nt1!ObHeaderCookie
volatility3/framework/symbols/windows/extensions/registry.py
+45 -10
View File
@@ -159,14 +159,20 @@ class CM_KEY_NODE(objects.StructType):
"""Extension to allow traversal of registry keys."""
def get_volatile(self) -> bool:
"""
Returns a bool indicating whether or not the key is volatile.
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
if not isinstance(self._context.layers[self.vol.layer_name], RegistryHive):
raise ValueError(
"Cannot determine volatility of registry key without an offset in a RegistryHive layer"
)
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
return bool(self.vol.offset & 0x80000000)
def get_subkeys(self) -> Iterator["CM_KEY_NODE"]:
"""Returns a list of the key nodes."""
"""Returns a list of the key nodes.
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
hive = self._context.layers[self.vol.layer_name]
if not isinstance(hive, RegistryHive):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
@@ -222,7 +228,10 @@ class CM_KEY_NODE(objects.StructType):
yield from self._get_subkeys_recursive(hive, subnode)
def get_values(self) -> Iterator["CM_KEY_VALUE"]:
"""Returns a list of the Value nodes for a key."""
"""Returns a list of the Value nodes for a key.
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
hive = self._context.layers[self.vol.layer_name]
if not isinstance(hive, RegistryHive):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
@@ -251,6 +260,11 @@ class CM_KEY_NODE(objects.StructType):
return self.Name.cast("string", max_length=namelength, encoding="latin-1")
def get_key_path(self) -> str:
"""
Returns the full path to this registry key.
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
reg = self._context.layers[self.vol.layer_name]
if not isinstance(reg, RegistryHive):
raise TypeError("Key was not instantiated on a RegistryHive layer")
@@ -276,7 +290,16 @@ class CM_KEY_VALUE(objects.StructType):
return RegValueTypes(self.Type)
def decode_data(self) -> Union[int, bytes]:
"""Properly decodes the data associated with the value node"""
"""
Properly decodes the data associated with the value node.
If an InvalidAddressException occurs when reading data from the
underlying RegistryHive layer, the data will be padded with null bytes
of the same length.
Raises ValueError if the data cannot be read
Raises TypeError if the class was not instantiated on a RegistryHive layer
"""
# Determine if the data is stored inline
datalen = self.DataLength
data = b""
@@ -310,14 +333,26 @@ class CM_KEY_VALUE(objects.StructType):
and block_offset < layer.maximum_address
):
amount = min(BIG_DATA_MAXLEN, datalen)
data += layer.read(
offset=layer.get_cell(block_offset).vol.offset, length=amount
)
try:
data += layer.read(
offset=layer.get_cell(block_offset).vol.offset,
length=amount,
)
except exceptions.InvalidAddressException:
vollog.debug(
f"Failed to read {amount:x} bytes of data, padding with {amount:x}"
)
datalen -= amount
else:
# Suspect Data actually points to a Cell,
# but the length at the start could be negative so just adding 4 to jump past it
data = layer.read(self.Data + 4, datalen)
try:
data = layer.read(self.Data + 4, datalen)
except exceptions.InvalidAddressException:
vollog.debug(
f"Failed to read {datalen:x} bytes of data, returning {datalen:x} null bytes"
)
data = b"\x00" * datalen
if self.get_type() == RegValueTypes.REG_DWORD:
if len(data) != struct.calcsize("<L"):
volatility3/plugins/windows/registry/__init__.py
+1 -1
View File
@@ -15,5 +15,5 @@ import os
import sys
# This is necessary to ensure the core plugins are available, whilst still be overridable
parent_module, module_name = ".".join(__name__.split(".")[:-1]), __name__.split(".")[-1]
parent_module, module_name = __name__.rsplit(".", maxsplit=1)
__path__ = [os.path.join(x, module_name) for x in sys.modules[parent_module].__path__]