Merge branch 'develop' into linux_module_symbols_improvements

This commit is contained in:
Gustavo Moreira
2025-01-30 19:53:50 +11:00
87 changed files with 1982 additions and 558 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
+21 -18
View File
@@ -1,7 +1,15 @@
[project]
name = "volatility3"
description = "Memory forensics framework"
keywords = ["volatility", "memory", "forensics", "framework", "windows", "linux", "volshell"]
keywords = [
"volatility",
"memory",
"forensics",
"framework",
"windows",
"linux",
"volshell",
]
readme = "README.md"
authors = [
{ name = "Volatility Foundation", email = "volatility@volatilityfoundation.org" },
@@ -10,9 +18,7 @@ requires-python = ">=3.8.0"
license = { text = "VSL" }
dynamic = ["version"]
dependencies = [
"pefile>=2024.8.26",
]
dependencies = ["pefile>=2024.8.26"]
[project.optional-dependencies]
full = [
@@ -26,15 +32,12 @@ full = [
"pillow>=10.0.0,<11.0.0",
]
cloud = [
"gcsfs>=2024.10.0",
"s3fs>=2024.10.0",
]
cloud = ["gcsfs>=2024.10.0", "s3fs>=2024.10.0"]
dev = [
"volatility3[full,cloud]",
"jsonschema>=4.23.0,<5",
"pyinstaller>=6.11.0,<7",
"pyinstaller>=6.5.0,<7",
"pyinstaller-hooks-contrib>=2024.9",
"types-jsonschema>=4.23.0,<5",
]
@@ -48,8 +51,8 @@ test = [
docs = [
"volatility3[dev]",
"sphinx>=8.0.0,<7",
"sphinx-autodoc-typehints>=2.5.0,<3",
"sphinx>=4.0.0,<9",
"sphinx-autodoc-typehints>=2.0.0,<3",
"sphinx-rtd-theme>=3.0.1,<4",
]
@@ -79,16 +82,16 @@ target-version = "py38"
[tool.ruff.lint]
select = [
"F", # pyflakes
"E", # pycodestyle errors
"W", # pycodestyle warnings
"G", # flake8-logging-format
"PIE", # flake8-pie
"UP", # pyupgrade
"F", # pyflakes
"E", # pycodestyle errors
"W", # pycodestyle warnings
"G", # flake8-logging-format
"PIE", # flake8-pie
"UP", # pyupgrade
]
ignore = [
"E501", # ignore due to conflict with formatter
"E501", # ignore due to conflict with formatter
]
[build-system]
+19 -6
View File
@@ -708,6 +708,24 @@ def test_linux_page_cache_inodepages(image, volatility, python):
inode_address = hex(0x88001AB5C270)
inode_dump_filename = f"inode_{inode_address}.dmp"
rc, out, _err = runvol_plugin(
"linux.pagecache.InodePages",
image,
volatility,
python,
pluginargs=["--inode", inode_address],
)
assert rc == 0
assert out.count(b"\n") > 4
# PageVAddr PagePAddr MappingAddr .. DumpSafe
assert re.search(
rb"0xea000054c5f8\s0x18389000\s0x88001ab5c3b0.*?True",
out,
)
try:
rc, out, _err = runvol_plugin(
"linux.pagecache.InodePages",
@@ -718,13 +736,8 @@ def test_linux_page_cache_inodepages(image, volatility, python):
)
assert rc == 0
assert out.count(b"\n") > 4
assert out.count(b"\n") >= 4
# PageVAddr PagePAddr MappingAddr .. DumpSafe
assert re.search(
rb"0xea000054c5f8\s0x18389000\s0x88001ab5c3b0.*?True",
out,
)
assert os.path.exists(inode_dump_filename)
with open(inode_dump_filename, "rb") as fp:
inode_contents = fp.read()
+24 -9
View File
@@ -363,10 +363,21 @@ class CommandLine:
metavar="PLUGIN",
)
for plugin in sorted(plugin_list):
# First line of a plugin docstring will be the short description for -h.
# Text after the first two consecutive new lines will be
# the additional description (argparse epilog).
short_help = additional_help = None
if plugin_list[plugin].__doc__ is not None:
doc_split = plugin_list[plugin].__doc__.split("\n\n", 1)
short_help = doc_split[0].strip()
if len(doc_split) > 1:
additional_help = doc_split[1].strip()
plugin_parser = subparser.add_parser(
plugin,
help=plugin_list[plugin].__doc__,
description=plugin_list[plugin].__doc__,
help=short_help,
description=short_help,
epilog=additional_help,
)
self.populate_requirements_argparse(plugin_parser, plugin_list[plugin])
@@ -572,6 +583,8 @@ class CommandLine:
fulltrace = traceback.TracebackException.from_exception(excp).format(chain=True)
vollog.debug("".join(fulltrace))
file_a_bug_msg = f"Please re-run with -vvv and file a bug with the output at {constants.BUG_URL}"
if isinstance(excp, exceptions.InvalidAddressException):
general = "Volatility was unable to read a requested page:"
if isinstance(excp, exceptions.SwappedInvalidAddressException):
@@ -616,9 +629,7 @@ class CommandLine:
elif isinstance(excp, exceptions.LayerException):
general = f"Volatility experienced a layer-related issue: {excp.layer_name}"
detail = f"{excp}"
caused_by = [
"A faulty layer implementation (re-run with -vvv and file a bug)"
]
caused_by = [f"A faulty layer implementation. {file_a_bug_msg}"]
elif isinstance(excp, exceptions.MissingModuleException):
general = f"Volatility could not import a necessary module: {excp.module}"
detail = f"{excp}"
@@ -629,13 +640,17 @@ class CommandLine:
general = "Volatility experienced an issue when rendering the output:"
detail = f"{excp}"
caused_by = ["An invalid renderer option, such as no visible columns"]
elif isinstance(excp, exceptions.VersionMismatchException):
general = "A version mismatch was detected between two components:"
detail = f"{excp}"
caused_by = [
excp.failure_reason or "An outdated API caller, such as a method.",
file_a_bug_msg,
]
else:
general = "Volatility encountered an unexpected situation."
detail = ""
caused_by = [
"Please re-run using with -vvv and file a bug with the output",
f"at {constants.BUG_URL}",
]
caused_by = [file_a_bug_msg]
# Code that actually renders the exception
output = sys.stderr
+77
View File
@@ -3,6 +3,7 @@
#
from typing import Any, List, Optional, Tuple, Union
from enum import Enum
from volatility3.cli.volshell import generic
from volatility3.framework import constants, interfaces
@@ -10,6 +11,16 @@ from volatility3.framework.configuration import requirements
from volatility3.plugins.linux import pslist
# Could import the enum from psscan.py to avoid code duplication
class DescExitStateEnum(Enum):
"""Enum for linux task exit_state as defined in include/linux/sched.h"""
TASK_RUNNING = 0x00000000
EXIT_DEAD = 0x00000010
EXIT_ZOMBIE = 0x00000020
EXIT_TRACE = EXIT_ZOMBIE | EXIT_DEAD
class Volshell(generic.Volshell):
"""Shell environment to directly interact with a linux memory image."""
@@ -40,6 +51,71 @@ class Volshell(generic.Volshell):
return None
print(f"No task with task ID {pid} found")
def get_process(self, pid=None, virtaddr=None, physaddr=None):
"""Return the task_struct object that matches the pid. If a physical or a virtual address is provided, construct the task_struct object at said address. Only one parameter is allowed.
Args:
pid (int, optional): PID to search for
virtaddr (int, optional): Virtual address to construct object at
physaddr (int, optional): Physical address to construct object at
Returns:
ObjectInterface: task_struct Object
"""
if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
print("Only one parameter is accepted")
return None
vmlinux_module_name = self.config["kernel"]
vmlinux = self.context.modules[vmlinux_module_name]
kernel_layer_name = vmlinux.layer_name
kernel_layer = self.context.layers[kernel_layer_name]
memory_layer_name = kernel_layer.dependencies[0]
task_struct_symbol = vmlinux.symbol_table_name + constants.BANG + "task_struct"
if virtaddr is not None:
task = self.context.object(
task_struct_symbol,
layer_name=kernel_layer_name,
offset=virtaddr,
)
if physaddr is not None:
task = self.context.object(
task_struct_symbol,
layer_name=memory_layer_name,
offset=physaddr,
native_layer_name=kernel_layer_name,
)
if physaddr is not None or virtaddr is not None:
try:
DescExitStateEnum(task.exit_state)
except ValueError:
print(
f"task_struct @ {hex(task.vol.offset)} as exit_state {task.exit_state} is likely not valid"
)
if not (0 < task.pid < 65535):
print(
f"task_struct @ {hex(task.vol.offset)} as pid {task.pid} is likely not valid"
)
return task
if pid is not None:
tasks = self.list_tasks()
for task in tasks:
if task.pid == pid:
return task
print(f"No task with task ID {pid} found")
return None
def list_tasks(self):
"""Returns a list of task objects from the primary layer"""
# We always use the main kernel memory and associated symbols
@@ -50,6 +126,7 @@ class Volshell(generic.Volshell):
result += [
(["ct", "change_task", "cp"], self.change_task),
(["lt", "list_tasks", "ps"], self.list_tasks),
(["gp", "get_process", "get_task"], self.get_process),
(["symbols"], self.context.symbol_space[self.current_symbol_table]),
]
if self.config.get("pid", None) is not None:
+56
View File
@@ -44,11 +44,67 @@ class Volshell(generic.Volshell):
)
)
def get_process(self, pid=None, virtaddr=None, physaddr=None):
"""Returns the _EPROCESS object that matches the pid. If a physical or a virtual address is provided, construct the _EPROCESS object at said address. Only one parameter is allowed.
Args:
pid (int, optional): PID / UniqueProcessId to search for.
virtaddr (int, optional): Virtual address to construct object at
physaddr (int, optional): Physical address to construct object at
Returns:
ObjectInterface: _EPROCESS Object
"""
if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
print("Only one parameter is accepted")
return None
kernel_name = self.config["kernel"]
kernel = self.context.modules[kernel_name]
kernel_layer_name = kernel.layer_name
kernel_layer = self.context.layers[kernel_layer_name]
memory_layer_name = kernel_layer.dependencies[0]
eprocess_symbol = kernel.symbol_table_name + constants.BANG + "_EPROCESS"
if virtaddr is not None:
eproc = self.context.object(
eprocess_symbol,
layer_name=kernel_layer_name,
offset=virtaddr,
)
return eproc
if physaddr is not None:
eproc = self.context.object(
eprocess_symbol,
layer_name=memory_layer_name,
offset=physaddr,
native_layer_name=kernel_layer_name,
)
return eproc
if pid is not None:
processes = self.list_processes()
for process in processes:
if process.UniqueProcessId == pid:
return process
print(f"No process with process ID {pid} found")
return None
return None
def construct_locals(self) -> List[Tuple[List[str], Any]]:
result = super().construct_locals()
result += [
(["cp", "change_process"], self.change_process),
(["lp", "list_processes", "ps"], self.list_processes),
(["gp", "get_process"], self.get_process),
(["symbols"], self.context.symbol_space[self.current_symbol_table]),
]
if self.config.get("pid", None) is not None:
+72 -3
View File
@@ -11,10 +11,24 @@ import inspect
import logging
import os
import traceback
from typing import Any, Dict, Generator, List, Optional, Tuple, Type, TypeVar
import functools
import warnings
from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, TypeVar
from volatility3.framework import constants, interfaces
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
if (
sys.version_info.major != constants.REQUIRED_PYTHON_VERSION[0]
or sys.version_info.minor < constants.REQUIRED_PYTHON_VERSION[1]
or (
sys.version_info.minor == constants.REQUIRED_PYTHON_VERSION[1]
and sys.version_info.micro < constants.REQUIRED_PYTHON_VERSION[2]
)
):
raise RuntimeError(
f"Volatility framework requires python version {'.'.join(str(x) for x in constants.REQUIRED_PYTHON_VERSION)} or greater"
)
# ##
#
@@ -52,12 +66,67 @@ def require_interface_version(*args) -> None:
)
class Deprecation:
"""Deprecation related methods."""
@staticmethod
def deprecated_method(
replacement: Callable,
replacement_version: Tuple[int, int, int] = None,
additional_information: str = "",
):
"""A decorator for marking functions as deprecated.
Args:
replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method)
replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface.
additional_information: Information appended at the end of the deprecation message
"""
def decorator(deprecated_func):
@functools.wraps(deprecated_func)
def wrapper(*args, **kwargs):
nonlocal replacement, replacement_version, additional_information
# Prevent version mismatches between deprecated (proxy) methods and the ones they proxy
if (
replacement_version is not None
and callable(replacement)
and hasattr(replacement, "__self__")
):
replacement_base_class = replacement.__self__
# Verify that the base class inherits from VersionableInterface
if inspect.isclass(replacement_base_class) and issubclass(
replacement_base_class,
interfaces.configuration.VersionableInterface,
):
# SemVer check
if not requirements.VersionRequirement.matches_required(
replacement_version, replacement_base_class.version
):
raise exceptions.VersionMismatchException(
deprecated_func,
replacement_base_class,
replacement_version,
"This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.",
)
deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}"
warnings.warn(deprecation_msg, FutureWarning)
# Return the wrapped function with its original arguments
return deprecated_func(*args, **kwargs)
return wrapper
return decorator
class NonInheritable:
def __init__(self, value: Any, cls: Type) -> None:
self.default_value = value
self.cls = cls
def __get__(self, obj: Any, get_type: Optional[Type] = None) -> Any:
def __get__(self, obj: Any, get_type: Type = Optional[None]) -> Any:
if type is self.cls:
if hasattr(self.default_value, "__get__"):
return self.default_value.__get__(obj, get_type)
+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"
@@ -376,8 +376,74 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
valid_kernel = (virtual_layer_name, address, res[0])
return valid_kernel
def method_low_stub_offset(
self,
context: interfaces.context.ContextInterface,
vlayer: layers.intel.Intel,
progress_callback: constants.ProgressCallback = None,
) -> Optional[ValidKernelType]:
# This method is only valid for x64 systems
if not isinstance(vlayer, intel.Intel32e):
return None
kernel_hint = 0
kernel_base = 0
physical_layer = context.layers.get("memory_layer")
# Try locating kernel base via x64 Low Stub in lower 1MB starting from second page (4KB)
# If "Discard Low Memory" setting is disabled in BIOS, the Low Stub may be at the third/fourth or further pages
for offset in range(0x1000, 0x100000, 0x1000):
try:
jmp_and_completion_values = int.from_bytes(
physical_layer.read(offset, 0x8), "little"
)
if (
0xFFFFFFFFFFFF00FF & jmp_and_completion_values
!= constants.windows.JMP_AND_COMPLETION_SIGNATURE
):
continue
cr3_value = int.from_bytes(
physical_layer.read(
offset + constants.windows.PROCESSOR_START_BLOCK_CR3_OFFSET, 0x8
),
"little",
)
# Compare previously observed valid page table address that's stored in vlayer._initial_entry
# with PROCESSOR_START_BLOCK->ProcessorState->SpecialRegisters->Cr3
# which was observed to be an invalid page address, so add 1 (to make it valid too)
if (cr3_value + 1) != vlayer._initial_entry:
continue
potential_kernel_hint = int.from_bytes(
physical_layer.read(
offset
+ constants.windows.PROCESSOR_START_BLOCK_LM_TARGET_OFFSET,
0x8,
),
"little",
)
if 0x3 & potential_kernel_hint:
continue
kernel_hint = potential_kernel_hint & 0xFFFFFFFFFFFF
kernel_base = kernel_hint & (~0x1FFFFF) & 0xFFFFFFFFFFFF
break
except exceptions.InvalidAddressException:
continue
if kernel_base:
# Scanning 32mb in 2mb chunks for the 'ntoskrnl' base address
while (kernel_base + 0x2000000) > kernel_hint:
for i in range(0, 0x200000, 0x1000):
valid_kernel = self.check_kernel_offset(
context, vlayer, kernel_base, progress_callback
)
if valid_kernel:
return valid_kernel
kernel_base -= 0x200000
return None
# List of methods to be run, in order, to determine the valid kernels
methods = [
method_low_stub_offset,
method_kdbg_offset,
method_module_offset,
method_fixed_mapping,
@@ -1,14 +0,0 @@
import sys
required_python_version = (3, 8, 0)
if (
sys.version_info.major != required_python_version[0]
or sys.version_info.minor < required_python_version[1]
or (
sys.version_info.minor == required_python_version[1]
and sys.version_info.micro < required_python_version[2]
)
):
raise RuntimeError(
f"Volatility framework requires python version {required_python_version[0]}.{required_python_version[1]}.{required_python_version[2]} or greater"
)
@@ -23,6 +23,8 @@ from volatility3.framework.constants._version import (
VERSION_SUFFIX as VERSION_SUFFIX,
)
REQUIRED_PYTHON_VERSION = (3, 8, 0)
PLUGINS_PATH = [
os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "plugins")),
os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "plugins")),
+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 = 19 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
@@ -6,6 +6,7 @@
Linux-specific values that aren't found in debug symbols
"""
from enum import IntEnum, Flag
from dataclasses import dataclass
KERNEL_NAME = "__kernel__"
@@ -355,3 +356,57 @@ MODULE_MINIMUM_SIZE = 4096
# Kallsyms
KSYM_NAME_LEN = 512
@dataclass
class TaintFlag:
shift: int
desc: str
when_present: bool
module: bool
TAINT_FLAGS = {
"P": TaintFlag(
shift=1 << 0, desc="PROPRIETARY_MODULE", when_present=True, module=True
),
"G": TaintFlag(
shift=1 << 0, desc="PROPRIETARY_MODULE", when_present=False, module=True
),
"F": TaintFlag(shift=1 << 1, desc="FORCED_MODULE", when_present=True, module=False),
"S": TaintFlag(
shift=1 << 2, desc="CPU_OUT_OF_SPEC", when_present=True, module=False
),
"R": TaintFlag(shift=1 << 3, desc="FORCED_RMMOD", when_present=True, module=False),
"M": TaintFlag(shift=1 << 4, desc="MACHINE_CHECK", when_present=True, module=False),
"B": TaintFlag(shift=1 << 5, desc="BAD_PAGE", when_present=True, module=False),
"U": TaintFlag(shift=1 << 6, desc="USER", when_present=True, module=False),
"D": TaintFlag(shift=1 << 7, desc="DIE", when_present=True, module=False),
"A": TaintFlag(
shift=1 << 8, desc="OVERRIDDEN_ACPI_TABLE", when_present=True, module=False
),
"W": TaintFlag(shift=1 << 9, desc="WARN", when_present=True, module=False),
"C": TaintFlag(shift=1 << 10, desc="CRAP", when_present=True, module=True),
"I": TaintFlag(
shift=1 << 11, desc="FIRMWARE_WORKAROUND", when_present=True, module=False
),
"O": TaintFlag(shift=1 << 12, desc="OOT_MODULE", when_present=True, module=True),
"E": TaintFlag(
shift=1 << 13, desc="UNSIGNED_MODULE", when_present=True, module=True
),
"L": TaintFlag(shift=1 << 14, desc="SOFTLOCKUP", when_present=True, module=False),
"K": TaintFlag(shift=1 << 15, desc="LIVEPATCH", when_present=True, module=True),
"X": TaintFlag(shift=1 << 16, desc="AUX", when_present=True, module=True),
"T": TaintFlag(shift=1 << 17, desc="RANDSTRUCT", when_present=True, module=True),
"N": TaintFlag(shift=1 << 18, desc="TEST", when_present=True, module=True),
}
"""Flags used to taint kernel and modules, for debugging purposes.
Map based on 6.12-rc5.
Documentation :
- https://www.kernel.org/doc/Documentation/admin-guide/sysctl/kernel.rst#:~:text=guide/sysrq.rst.-,tainted,-%3D%3D%3D%3D%3D%3D%3D%0A%0ANon%2Dzero%20if
- https://www.kernel.org/doc/Documentation/admin-guide/tainted-kernels.rst#:~:text=More%20detailed%20explanation%20for%20tainting
- taint_flag kernel struct
- taint_flags kernel constant
"""
@@ -10,3 +10,21 @@ KERNEL_MODULE_NAMES = ["ntkrnlmp", "ntkrnlpa", "ntkrpamp", "ntoskrnl"]
"""The list of names that kernel modules can have within the windows OS"""
PE_MAX_EXTRACTION_SIZE = 1024 * 1024 * 256
"""
The following constants represent the layout of the Low Stub which exists only on x64 machines with no virtualization/emulation,
responsible for transitioning from Real Mode(16 bit) to Protected Mode(32 bit) and Long Mode(64 bit) on boot/return from sleep.
Contains offsets to fields and structures within the undocumented structure _PROCESSOR_START_BLOCK.
Here's a reference: https://github.com/mic101/windows/blob/master/WRK-v1.2/base/ntos/inc/amd64.h#L3334
"""
# Expected signature for validation, constructed from:
# PROCESSOR_START_BLOCK->Jmp->OpCode | PROCESSOR_START_BLOCK->Jmp->Offset | PROCESSOR_START_BLOCK->CompletionFlag
JMP_AND_COMPLETION_SIGNATURE = 0x00000001000600E9
# Address of LmTarget (Long Mode target)
PROCESSOR_START_BLOCK_LM_TARGET_OFFSET = (
0x70 # PROCESSOR_START_BLOCK->LmTarget, PVOID 8 bytes
)
# CR3 register within structures describing initial processor state to be started
PROCESSOR_START_BLOCK_CR3_OFFSET = 0xA0 # PROCESSOR_START_BLOCK->ProcessorState->SpecialRegisters->Cr3, ULONG64 8 bytes
+10 -12
View File
@@ -11,7 +11,8 @@ without them interfering with each other.
import functools
import hashlib
import logging
from typing import Callable, Iterable, List, Optional, Set, Tuple, Union
import re
from typing import Callable, Dict, Iterable, List, Optional, Set, Tuple, Union
from volatility3.framework import constants, interfaces, symbols, exceptions
from volatility3.framework.objects import templates
@@ -337,7 +338,7 @@ class Module(interfaces.context.ModuleInterface):
)
@property
def symbols(self):
def symbols(self) -> Iterable[str]:
return self.context.symbol_space[self.symbol_table_name].symbols
get_symbol = get_module_wrapper("get_symbol")
@@ -386,10 +387,8 @@ class ModuleCollection(interfaces.context.ModuleContainer):
"""Class to contain a collection of SizedModules and reason about their
contents."""
def __init__(
self, modules: Optional[List[interfaces.context.ModuleInterface]] = None
) -> None:
self._prefix_count = {}
def __init__(self, modules: Optional[List[SizedModule]] = None) -> None:
self._modules: Dict[str, SizedModule] = {}
super().__init__(modules)
def deduplicate(self) -> "ModuleCollection":
@@ -402,20 +401,19 @@ class ModuleCollection(interfaces.context.ModuleContainer):
new_modules = []
seen: Set[str] = set()
for mod in self._modules:
if mod.hash not in seen or mod.size == 0:
if self._modules[mod].hash not in seen or self._modules[mod].size == 0:
new_modules.append(mod)
seen.add(mod.hash) # type: ignore # FIXME: mypy #5107
seen.add(self._modules[mod].hash)
return ModuleCollection(new_modules)
def free_module_name(self, prefix: str = "module") -> str:
"""Returns an unused module name"""
if prefix not in self._prefix_count:
self._prefix_count[prefix] = 1
existing_names = [name for name in self if re.match(rf"^{prefix}[0-9]*$", name)]
if not existing_names:
return prefix
count = self._prefix_count[prefix]
count = len(existing_names)
while prefix + str(count) in self:
count += 1
self._prefix_count[prefix] = count
return prefix + str(count)
@property
+34 -1
View File
@@ -8,9 +8,10 @@ space or symbol tables, and by layers when an address is invalid. The
:class:`PagedInvalidAddressException` contains information about the
size of the invalid page.
"""
from typing import Dict, Optional
from typing import Callable, Dict, Optional, Tuple
from volatility3.framework import interfaces
from volatility3.framework.interfaces.configuration import VersionableInterface
class VolatilityException(Exception):
@@ -130,3 +131,35 @@ class OfflineException(VolatilityException):
class RenderException(VolatilityException):
"""Thrown if there is an error during rendering"""
class LinuxPageCacheException(VolatilityException):
"""Thrown if there is an error during Linux Page Cache processing"""
class VersionMismatchException(VolatilityException):
"""Thrown if a version mismatch has been encountered between two components."""
def __init__(
self,
source_component: Callable,
target_component: VersionableInterface,
target_version: Tuple[int, int, int],
failure_reason: str = None,
*args,
):
"""
Args:
source_component: The component that required the target component
target_component: The component that is required. Must inherit from VersionableInterface
target_version: The version of the target component that was required, and ultimately was not satisfied
failure_reason: A detailed failure reason to enhance debugging and bug tracking
"""
super().__init__(*args)
self.source_component = source_component
self.target_component = target_component
self.target_version = target_version
self.failure_reason = failure_reason
def __str__(self):
return f"{self.source_component.__module__+ '.' + self.source_component.__qualname__}: Version {self.target_version} dependency on {self.target_component.__module__+ '.' + self.target_component.__name__} {self.target_component.version} unmet."
+4 -2
View File
@@ -302,9 +302,11 @@ class ModuleInterface(interfaces.configuration.ConfigurableInterface):
def has_enumeration(self, name: str) -> bool:
"""Determines whether an enumeration is present in the module's symbol table."""
@property
@abstractmethod
def symbols(self) -> List:
"""Lists the symbols contained in the symbol table for this module"""
def symbols(self) -> Iterable[str]:
"""Returns an iterable of the symbols contained in the symbol table for this module"""
raise NotImplementedError("Symbols property has not been implemented.")
@abstractmethod
def get_symbols_by_absolute_location(self, offset: int, size: int = 0) -> List[str]:
+11 -4
View File
@@ -122,7 +122,7 @@ class BaseSymbolTableInterface:
@property
def symbols(self) -> Iterable[str]:
"""Returns an iterator of the Symbol names."""
"""Returns an iterable of the available symbol names."""
raise NotImplementedError(
"Abstract property symbols not implemented by subclass."
)
@@ -131,7 +131,7 @@ class BaseSymbolTableInterface:
@property
def types(self) -> Iterable[str]:
"""Returns an iterator of the Symbol type names."""
"""Returns an iterable of the available symbol type names."""
raise NotImplementedError(
"Abstract property types not implemented by subclass."
)
@@ -149,7 +149,7 @@ class BaseSymbolTableInterface:
@property
def enumerations(self) -> Iterable[Any]:
"""Returns an iterator of the Enumeration names."""
"""Returns an iterable of the available enumerations."""
raise NotImplementedError(
"Abstract property enumerations not implemented by subclass."
)
@@ -366,6 +366,7 @@ class NativeTableInterface(BaseSymbolTableInterface):
@property
def symbols(self) -> Iterable[str]:
"""Returns an iterable of the available symbol names."""
return []
def get_enumeration(self, name: str) -> objects.Template:
@@ -374,7 +375,13 @@ class NativeTableInterface(BaseSymbolTableInterface):
)
@property
def enumerations(self) -> Iterable[str]:
def enumerations(self) -> Iterable[Any]:
"""Returns an iterable of the available enumerations."""
return []
@property
def types(self) -> Iterable[str]:
"""Returns an iterable of the available symbol type names."""
return []
@@ -129,6 +129,8 @@ if HAS_LEECHCORE:
def readline(self, __size: Optional[int] = ...) -> bytes:
data = b""
if not __size:
__size = 0
while __size > self._chunk_size or __size < 0:
data += self.read(self._chunk_size)
index = data.find(b"\n")
+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
+5 -2
View File
@@ -33,11 +33,12 @@ def array_to_string(
) -> interfaces.objects.ObjectInterface:
"""Takes a volatility Array of characters and returns a string."""
# TODO: Consider checking the Array's target is a native char
if count is None:
count = array.vol.count
if not isinstance(array, objects.Array):
raise TypeError("Array_to_string takes an Array of char")
if count is None:
count = array.vol.count
return array.cast("string", max_length=count, errors=errors)
@@ -45,8 +46,10 @@ def pointer_to_string(pointer: "objects.Pointer", count: int, errors: str = "rep
"""Takes a volatility Pointer to characters and returns a string."""
if not isinstance(pointer, objects.Pointer):
raise TypeError("pointer_to_string takes a Pointer")
if count < 1:
raise ValueError("pointer_to_string requires a positive count")
char = pointer.dereference()
return char.cast("string", max_length=count, errors=errors)
@@ -14,8 +14,8 @@ vollog = logging.getLogger(__name__)
class ConfigWriter(plugins.PluginInterface):
"""Runs the automagics and both prints and outputs configuration in the
output directory."""
"""Runs the automagics and both prints and outputs configuration in the \
output directory."""
_required_framework_version = (2, 0, 0)
+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
@@ -5,6 +5,7 @@
import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, symbols
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
@@ -27,6 +28,11 @@ class Check_idt(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
),
@@ -99,8 +105,10 @@ class Check_idt(interfaces.plugins.PluginInterface):
idt_addr = idt_addr & address_mask
module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
vmlinux, handlers, idt_addr
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, idt_addr
)
)
yield (
@@ -18,6 +18,7 @@ vollog = logging.getLogger(__name__)
class Check_modules(plugins.PluginInterface):
"""Compares module list to sysfs info, if available"""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
@classmethod
@@ -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
@@ -83,7 +82,7 @@ class Check_syscall(plugins.PluginInterface):
return table_size
def _get_table_info_disassembly(self, ptr_sz, vmlinux):
def _get_table_info_disassembly(self, ptr_sz, vmlinux) -> int:
"""Find the size of the system call table by disassembling functions
that immediately reference it in their first instruction This is in the
form 'cmp reg,NR_syscalls'."""
@@ -108,9 +107,13 @@ class Check_syscall(plugins.PluginInterface):
return 0
vmlinux = self.context.modules[self.config["kernel"]]
data = self.context.layers.read(vmlinux.layer_name, func_addr, 6)
vmlinux_layer = self.context.layers[vmlinux.layer_name]
try:
data = vmlinux_layer.read(func_addr, 6)
except exceptions.InvalidAddressException:
return 0
for address, size, mnemonic, op_str in md.disasm_lite(data, func_addr):
for _address, _size, mnemonic, op_str in md.disasm_lite(data, func_addr):
if mnemonic == "CMP":
table_size = int(op_str.split(",")[1].strip()) & 0xFFFF
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 for enumerating memory-mapped
ELF files across all processes."""
import logging
from typing import List, Optional, Type
@@ -18,7 +18,7 @@ class Envars(plugins.PluginInterface):
"""Lists processes with their environment variables"""
_required_framework_version = (2, 13, 0)
_version = (2, 0, 0)
_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -40,8 +40,9 @@ class Envars(plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_task_env_variables(
cls,
context: interfaces.context.ContextInterface,
task: interfaces.objects.ObjectInterface,
env_area_max_size: int = 8192,
@@ -16,8 +16,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
"""Carves memory to find hidden kernel modules"""
_required_framework_version = (2, 10, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -32,8 +31,9 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_modules_memory_boundaries(
cls,
context: interfaces.context.ContextInterface,
vmlinux_module_name: str,
) -> Tuple[int]:
@@ -4,6 +4,7 @@
import logging
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
@@ -26,6 +27,11 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
@@ -66,8 +72,10 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
):
call_addr = call_back.notifier_call
module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
vmlinux, handlers, call_addr
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, call_addr
)
)
yield (0, [format_hints.Hex(call_addr), module_name, symbol_name])
+43 -27
View File
@@ -5,7 +5,7 @@ import re
import logging
from abc import ABC, abstractmethod
from enum import Enum
from typing import Generator, Iterator, List, Tuple
from typing import Generator, Iterator, List, Tuple, Union
from volatility3.framework import (
class_subclasses,
@@ -135,8 +135,14 @@ class ABCKmsg(ABC):
bool: True if the kernel being analyzed fulfill the class requirements.
"""
def get_string(self, addr: int, length: int) -> str:
txt = self._context.layers[self.layer_name].read(addr, length) # type: ignore
def get_string(self, addr: int, length: int) -> Union[str, None]:
layer = self._context.layers[self.layer_name]
if not layer.is_valid(addr, length):
vollog.warning("Failed to read log record at address 0x%x", addr)
return None
txt = layer.read(addr, length)
return txt.decode(encoding="utf8", errors="replace")
def nsec_to_sec_str(self, nsec: int) -> str:
@@ -149,7 +155,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 +172,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]:
@@ -263,7 +269,7 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
def _get_log_struct_name(self):
return "log"
def get_text_from_log(self, msg) -> str:
def get_text_from_log(self, msg) -> Union[str, None]:
log_struct_name = self._get_log_struct_name()
log_struct_size = self.vmlinux.get_type(log_struct_name).size
msg_offset = msg.vol.offset + log_struct_size
@@ -272,7 +278,8 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
def get_log_lines(self, msg) -> Generator[str, None, None]:
if msg.text_len > 0:
text = self.get_text_from_log(msg)
yield from text.splitlines()
if text:
yield from text.splitlines()
def get_dict_lines(self, msg) -> Generator[str, None, None]:
if msg.dict_len == 0:
@@ -281,9 +288,13 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
log_struct_name = self._get_log_struct_name()
log_struct_size = self.vmlinux.get_type(log_struct_name).size
dict_offset = msg.vol.offset + log_struct_size + msg.text_len
dict_data = self._context.layers[self.layer_name].read(
dict_offset, msg.dict_len
)
layer = self._context.layers[self.layer_name]
try:
dict_data = layer.read(dict_offset, msg.dict_len)
except exceptions.InvalidAddressException:
vollog.debug("Unable to read kmsg dict from 0x%x", dict_offset)
return None
for chunk in dict_data.split(b"\x00"):
yield " " + chunk.decode()
@@ -317,23 +328,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):
@@ -399,7 +414,7 @@ class Kmsg_5_10_to_(ABCKmsg):
def symtab_checks(cls, vmlinux) -> bool:
return vmlinux.has_symbol("prb")
def get_text_from_data_ring(self, text_data_ring, desc, info) -> str:
def get_text_from_data_ring(self, text_data_ring, desc, info) -> Union[str, None]:
text_data_sz = text_data_ring.size_bits
text_data_mask = 1 << text_data_sz
@@ -427,7 +442,8 @@ class Kmsg_5_10_to_(ABCKmsg):
def get_log_lines(self, text_data_ring, desc, info) -> Generator[str, None, None]:
text = self.get_text_from_data_ring(text_data_ring, desc, info)
yield from text.splitlines()
if text:
yield from text.splitlines()
def get_dict_lines(self, info) -> Generator[str, None, None]:
dict_text = utility.array_to_string(info.dev_info.subsystem)
@@ -4,6 +4,7 @@
import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
@@ -20,7 +21,7 @@ class Kthreads(plugins.PluginInterface):
"""Enumerates kthread functions"""
_required_framework_version = (2, 11, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -30,6 +31,11 @@ class Kthreads(plugins.PluginInterface):
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
),
@@ -88,8 +94,10 @@ class Kthreads(plugins.PluginInterface):
if kthread.has_member("full_name")
else task_name
)
module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
vmlinux, handlers, threadfn
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, threadfn
)
)
fields = [
+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
@@ -0,0 +1,189 @@
# 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
from typing import List, Dict, Iterator
from volatility3.plugins.linux import lsmod, check_modules, hidden_modules
from volatility3.framework import interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.constants import architectures
from volatility3.framework.symbols.linux.utilities import tainting
vollog = logging.getLogger(__name__)
class Modxview(interfaces.plugins.PluginInterface):
"""Centralize lsmod, check_modules and hidden_modules results to efficiently \
spot modules presence and taints."""
_version = (1, 0, 0)
_required_framework_version = (2, 17, 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="linux-tainting", component=tainting.Tainting, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="check_modules",
plugin=check_modules.Check_modules,
version=(1, 0, 0),
),
requirements.PluginRequirement(
name="hidden_modules",
plugin=hidden_modules.Hidden_modules,
version=(1, 0, 0),
),
requirements.BooleanRequirement(
name="plain_taints",
description="Display the plain taints string for each module.",
optional=True,
default=False,
),
]
@classmethod
def flatten_run_modules_results(
cls, run_results: Dict[str, List[extensions.module]], deduplicate: bool = True
) -> Iterator[extensions.module]:
"""Flatten a dictionary mapping plugin names and modules list, to a single merged list.
This is useful to get a generic lookup list of all the detected modules.
Args:
run_results: dictionary of plugin names mapping a list of detected modules
deduplicate: remove duplicate modules, based on their offsets
Returns:
Iterator of modules objects
"""
seen_addresses = set()
for modules in run_results.values():
for module in modules:
if deduplicate and module.vol.offset in seen_addresses:
continue
seen_addresses.add(module.vol.offset)
yield module
@classmethod
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
run_hidden_modules: bool = True,
) -> Dict[str, List[extensions.module]]:
"""Run module scanning plugins and aggregate the results. It is designed
to not operate any inter-plugin results triage.
Args:
run_hidden_modules: specify if the hidden_modules plugin should be run
Returns:
Dictionary mapping each plugin to its corresponding result
"""
kernel = context.modules[kernel_name]
run_results = {}
# lsmod
run_results["lsmod"] = list(lsmod.Lsmod.list_modules(context, kernel_name))
# check_modules
sysfs_modules: dict = check_modules.Check_modules.get_kset_modules(
context, kernel_name
)
## Convert get_kset_modules() offsets back to module objects
run_results["check_modules"] = [
kernel.object(object_type="module", offset=m_offset, absolute=True)
for m_offset in sysfs_modules.values()
]
# hidden_modules
if run_hidden_modules:
known_modules_addresses = set(
context.layers[kernel.layer_name].canonicalize(module.vol.offset)
for module in run_results["lsmod"] + run_results["check_modules"]
)
modules_memory_boundaries = (
hidden_modules.Hidden_modules.get_modules_memory_boundaries(
context, kernel_name
)
)
run_results["hidden_modules"] = list(
hidden_modules.Hidden_modules.get_hidden_modules(
context,
kernel_name,
known_modules_addresses,
modules_memory_boundaries,
)
)
return run_results
def _generator(self):
kernel_name = self.config["kernel"]
run_results = self.run_modules_scanners(self.context, kernel_name)
aggregated_modules = {}
# We want to be explicit on the plugins results we are interested in
for plugin_name in ["lsmod", "check_modules", "hidden_modules"]:
# Iterate over each recovered module
for module in run_results[plugin_name]:
# Use offsets as unique keys, whether a module
# appears in many plugin runs or not
if aggregated_modules.get(module.vol.offset, None) is not None:
# Append the plugin to the list of originating plugins
aggregated_modules[module.vol.offset][1].append(plugin_name)
else:
aggregated_modules[module.vol.offset] = (module, [plugin_name])
for module_offset, (module, originating_plugins) in aggregated_modules.items():
# Tainting parsing capabilities applied to the module
if self.config.get("plain_taints"):
taints = tainting.Tainting.get_taints_as_plain_string(
self.context,
kernel_name,
module.taints,
True,
)
else:
taints = ",".join(
tainting.Tainting.get_taints_parsed(
self.context,
kernel_name,
module.taints,
True,
)
)
yield (
0,
(
module.get_name() or NotAvailableValue(),
format_hints.Hex(module_offset),
"lsmod" in originating_plugins,
"check_modules" in originating_plugins,
"hidden_modules" in originating_plugins,
taints or NotAvailableValue(),
),
)
def run(self):
columns = [
("Name", str),
("Address", format_hints.Hex),
("In procfs", bool),
("In sysfs", bool),
("Hidden", bool),
("Taints", str),
]
return TreeGrid(
columns,
self._generator(),
)
@@ -36,7 +36,7 @@ class MountInfo(plugins.PluginInterface):
"""Lists mount points on processes mount namespaces"""
_required_framework_version = (2, 2, 0)
_version = (1, 2, 3)
_version = (1, 2, 4)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -152,9 +152,11 @@ class MountInfo(plugins.PluginInterface):
if not (
task
and task.fs
and task.fs.root
and task.fs.is_readable()
and task.nsproxy
and task.nsproxy.is_readable()
and task.nsproxy.mnt_ns
and task.nsproxy.mnt_ns.is_readable()
):
# This task doesn't have all the information required.
# It should be a kernel < 2.6.30
@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import logging
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from typing import Iterator, List, Tuple
from volatility3 import framework
from volatility3.framework import (
@@ -98,6 +99,20 @@ class AbstractNetfilter(ABC):
f"linux.LinuxUtilities version not suitable: required {linuxutils_required_version} found {linuxutils_current_version}"
)
linux_utilities_modules_required_version = (
Netfilter._required_linux_utilities_modules_version
)
linux_utilities_modules_current_version = (
linux_utilities_modules.Modules._version
)
if not requirements.VersionRequirement.matches_required(
linux_utilities_modules_required_version,
linux_utilities_modules_current_version,
):
raise exceptions.PluginRequirementException(
f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}"
)
modules = lsmod.Lsmod.list_modules(context, kernel_module_name)
self.handlers = linux.LinuxUtilities.generate_kernel_handler_info(
context, kernel_module_name, modules
@@ -263,8 +278,10 @@ class AbstractNetfilter(ABC):
"""Helper to obtain the module and symbol name in the format needed for the
output of this plugin.
"""
module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
self.vmlinux, self.handlers, addr
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self._context, self.vmlinux.name, self.handlers, addr
)
)
if module_name == "UNKNOWN":
@@ -677,6 +694,7 @@ class Netfilter(interfaces.plugins.PluginInterface):
_version = (1, 1, 0)
_required_linux_utilities_modules_version = (1, 0, 0)
_required_linuxutils_version = (2, 1, 0)
_required_lsmod_version = (2, 0, 0)
@@ -688,6 +706,11 @@ class Netfilter(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=cls._required_linux_utilities_modules_version,
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=cls._required_lsmod_version
),
@@ -6,9 +6,9 @@ import math
import logging
import datetime
from dataclasses import dataclass, astuple
from typing import List, Set, Type, Iterable
from typing import List, Set, Type, Iterable, Tuple
from volatility3.framework import renderers, interfaces
from volatility3.framework import renderers, interfaces, exceptions
from volatility3.framework.renderers import format_hints
from volatility3.framework.interfaces import plugins
from volatility3.framework.configuration import requirements
@@ -104,7 +104,7 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -147,7 +147,13 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
Otherwise, it returns the same symlink_path
"""
# i_link (fast symlinks) were introduced in 4.2
if inode and inode.is_link and inode.has_member("i_link") and inode.i_link:
if (
inode
and inode.is_link
and inode.has_member("i_link")
and inode.i_link
and inode.i_link.is_readable()
):
i_link_str = inode.i_link.dereference().cast(
"string", max_length=255, encoding="utf-8", errors="replace"
)
@@ -253,6 +259,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
if not root_inode.is_valid():
continue
if not (root_inode.i_mapping and root_inode.i_mapping.is_readable()):
# Retrieving data from the page cache requires a valid address space
continue
# Inode already processed?
if root_inode_ptr in seen_inodes:
continue
@@ -284,6 +294,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
if not file_inode.is_valid():
continue
if not (file_inode.i_mapping and file_inode.i_mapping.is_readable()):
# Retrieving data from the page cache requires a valid address space
continue
# Inode already processed?
if file_inode_ptr in seen_inodes:
continue
@@ -316,10 +330,12 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
if self.config["find"]:
if inode_in.path == self.config["find"]:
inode_out = inode_in.to_user(vmlinux_layer)
yield (0, astuple(inode_out))
break # Only the first match
else:
inode_out = inode_in.to_user(vmlinux_layer)
yield (0, astuple(inode_out))
def generate_timeline(self):
@@ -344,8 +360,8 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
yield description, timeliner.TimeLinerType.MODIFIED, inode_out.modification_time
yield description, timeliner.TimeLinerType.CHANGED, inode_out.change_time
@staticmethod
def format_fields_with_headers(headers, generator):
@classmethod
def format_fields_with_headers(cls, headers, generator):
"""Uses the headers type to cast the fields obtained from the generator"""
for level, fields in generator:
formatted_fields = []
@@ -389,7 +405,7 @@ class InodePages(plugins.PluginInterface):
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
_version = (2, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -420,8 +436,9 @@ class InodePages(plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def write_inode_content_to_file(
cls,
inode: interfaces.objects.ObjectInterface,
filename: str,
open_method: Type[interfaces.plugins.FileHandlerInterface],
@@ -443,28 +460,80 @@ class InodePages(plugins.PluginInterface):
# created, saving both disk space and I/O time.
# Additionally, using the page index will guarantee that each page is written at the
# appropriate file position.
inode_size = inode.i_size
try:
with open_method(filename) as f:
inode_size = inode.i_size
f.truncate(inode_size)
file_initialized = False
with open_method(filename) as file_obj:
for page_idx, page_content in inode.get_contents():
current_fp = page_idx * vmlinux_layer.page_size
max_length = inode_size - current_fp
page_bytes = page_content[:max_length]
if current_fp + len(page_bytes) > inode_size:
page_bytes_len = min(max_length, len(page_content))
if (
current_fp >= inode_size
or current_fp + page_bytes_len > inode_size
):
vollog.error(
"Page out of file bounds: inode 0x%x, inode size %d, page index %d",
inode.vol.offset,
inode_size,
page_idx,
)
f.seek(current_fp)
f.write(page_bytes)
continue
page_bytes = page_content[:page_bytes_len]
if not file_initialized:
# Lazy initialization to avoid truncating the file until we are
# certain there is something to write
file_obj.truncate(inode_size)
file_initialized = True
file_obj.seek(current_fp)
file_obj.write(page_bytes)
except exceptions.LinuxPageCacheException:
vollog.error(
f"Error dumping cached pages for inode at {inode.vol.offset:#x}"
)
except OSError as e:
vollog.error("Unable to write to file (%s): %s", filename, e)
def _generate_inode_fields(
self,
inode: interfaces.objects.ObjectInterface,
vmlinux_layer: interfaces.layers.TranslationLayerInterface,
) -> Iterable[Tuple[int, int, int, int, bool, str]]:
inode_size = inode.i_size
try:
for page_obj in inode.get_pages():
if page_obj.mapping != inode.i_mapping:
vollog.warning(
f"Cached page at {page_obj.vol.offset:#x} has a mismatched address space with the inode. Skipping page"
)
continue
page_vaddr = page_obj.vol.offset
page_paddr = page_obj.to_paddr()
page_mapping_addr = page_obj.mapping
page_index = page_obj.index
page_file_offset = page_index * vmlinux_layer.page_size
dump_safe = (
page_file_offset < inode_size
and page_mapping_addr
and page_mapping_addr.is_readable()
)
page_flags_list = page_obj.get_flags_list()
page_flags = ",".join([x.replace("PG_", "") for x in page_flags_list])
fields = (
page_vaddr,
page_paddr,
page_mapping_addr,
page_index,
dump_safe,
page_flags,
)
yield 0, fields
except exceptions.LinuxPageCacheException:
vollog.warning(f"Page cache for inode at {inode.vol.offset:#x} is corrupt")
def _generator(self):
vmlinux_module_name = self.config["kernel"]
vmlinux = self.context.modules[vmlinux_module_name]
@@ -486,7 +555,6 @@ class InodePages(plugins.PluginInterface):
else:
vollog.error("Unable to find inode with path %s", self.config["find"])
return None
elif self.config["inode"]:
inode = vmlinux.object("inode", self.config["inode"], absolute=True)
else:
@@ -501,27 +569,6 @@ class InodePages(plugins.PluginInterface):
vollog.error("The inode is not a regular file")
return None
inode_size = inode.i_size
for page_obj in inode.get_pages():
page_vaddr = page_obj.vol.offset
page_paddr = page_obj.to_paddr()
page_mapping_addr = page_obj.mapping
page_index = int(page_obj.index)
page_file_offset = page_index * vmlinux_layer.page_size
dump_safe = page_file_offset < inode_size
page_flags_list = page_obj.get_flags_list()
page_flags = ",".join([x.replace("PG_", "") for x in page_flags_list])
fields = (
page_vaddr,
page_paddr,
page_mapping_addr,
page_index,
dump_safe,
page_flags,
)
yield 0, fields
if self.config["dump"]:
open_method = self.open
inode_address = inode.vol.offset
@@ -530,6 +577,8 @@ class InodePages(plugins.PluginInterface):
self.write_inode_content_to_file(
inode, filename, open_method, vmlinux_layer
)
else:
yield from self._generate_inode_fields(inode, vmlinux_layer)
def run(self):
headers = [
+33 -26
View File
@@ -21,7 +21,7 @@ class Maps(plugins.PluginInterface):
"""Lists all memory maps for all processes."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
MAXSIZE_DEFAULT = 1024 * 1024 * 1024 # 1 Gb
@@ -83,18 +83,24 @@ class Maps(plugins.PluginInterface):
Returns:
Yields vmas based on the task and filtered based on the filter function
"""
if task.mm:
for vma in task.mm.get_vma_iter():
if filter_func(vma):
yield vma
else:
vollog.debug(
f"Excluded vma at offset {vma.vol.offset:#x} for pid {task.pid} due to filter_func"
)
else:
mm_pointer = task.mm
if not mm_pointer:
vollog.debug(
f"Excluded pid {task.pid} as there is no mm member. It is likely a kernel thread."
f"Excluded pid {task.pid} as there is no mm member. It is likely a kernel thread"
)
return
if not mm_pointer.is_readable():
vollog.error(f"Task {task.pid} has an invalid mm member")
return
for vma in mm_pointer.get_vma_iter():
if filter_func(vma):
yield vma
else:
vollog.debug(
f"Excluded vma at offset {vma.vol.offset:#x} for pid {task.pid} due to filter_func"
)
@classmethod
def vma_dump(
@@ -174,31 +180,32 @@ class Maps(plugins.PluginInterface):
]
# if any of the user supplied addresses would fall within this vma return true
if addrs_in_vma:
return True
else:
return False
return bool(addrs_in_vma)
vma_filter_func = vma_filter_function
for task in tasks:
if not task.mm:
if not (task.mm and task.mm.is_readable()):
continue
name = utility.array_to_string(task.comm)
for vma in self.list_vmas(task, filter_func=vma_filter_func):
flags = vma.get_protection()
page_offset = vma.get_page_offset()
major = 0
minor = 0
inode = 0
if vma.vm_file != 0:
inode_num = None
try:
dentry = vma.vm_file.get_dentry()
if dentry != 0:
inode_object = dentry.d_inode
major = inode_object.i_sb.major
minor = inode_object.i_sb.minor
inode = inode_object.i_ino
inode_ptr = dentry.d_inode
inode_num = inode_ptr.i_ino
major = inode_ptr.i_sb.major
minor = inode_ptr.i_sb.minor
except exceptions.InvalidAddressException:
if not inode_num:
inode_num = 0
major = 0
minor = 0
path = vma.get_name(self.context, task)
file_output = "Disabled"
@@ -238,7 +245,7 @@ class Maps(plugins.PluginInterface):
format_hints.Hex(page_offset),
major,
minor,
inode,
inode_num,
path,
file_output,
),
+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
@@ -9,8 +9,7 @@ from volatility3.plugins.linux import pslist
class PsTree(interfaces.plugins.PluginInterface):
"""Plugin for listing processes in a tree based on their parent process
ID."""
"""Plugin for listing processes in a tree based on their parent process ID."""
_required_framework_version = (2, 13, 0)
_version = (1, 1, 1)
@@ -438,7 +438,7 @@ class Sockstat(plugins.PluginInterface):
"""Lists all network connections for all processes."""
_required_framework_version = (2, 0, 0)
_version = (3, 0, 2)
_version = (3, 0, 3)
@classmethod
def get_requirements(cls):
@@ -514,25 +514,28 @@ class Sockstat(plugins.PluginInterface):
fd_num, filp, _full_path = fd_internal.fd_fields
task = fd_internal.task
if not (filp.f_op and filp.f_op.is_readable()):
continue
if filp.f_op not in (sfop_addr, dfop_addr):
continue
dentry = filp.get_dentry()
if not dentry:
if not (dentry and dentry.is_readable()):
continue
d_inode = dentry.d_inode
if not d_inode:
if not (d_inode and d_inode.is_readable()):
continue
socket_alloc = linux.LinuxUtilities.container_of(
d_inode, "socket_alloc", "vfs_inode", vmlinux
)
socket = socket_alloc.socket
if not (socket and socket.sk):
if not socket_alloc:
continue
socket = socket_alloc.socket
if not (socket.sk and socket.sk.is_readable()):
continue
sock = socket.sk.dereference()
sock_type = sock.get_type()
@@ -5,6 +5,7 @@
import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
@@ -29,6 +30,11 @@ class tty_check(plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
@@ -79,8 +85,10 @@ class tty_check(plugins.PluginInterface):
recv_buf = tty_dev.ldisc.ops.receive_buf
module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
vmlinux, handlers, recv_buf
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, recv_buf
)
)
yield (0, (name, format_hints.Hex(recv_buf), module_name, symbol_name))
@@ -18,7 +18,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -105,8 +105,9 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
value,
)
@staticmethod
@classmethod
def get_vma_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[int, int]]:
"""Creates a map of start/end addresses for each virtual memory area in a task.
+2 -2
View File
@@ -11,8 +11,8 @@ from volatility3.framework.symbols import mac
class Mount(plugins.PluginInterface):
"""A module containing a collection of plugins that produce data typically
found in Mac's mount command"""
"""A module containing a collection of plugins that produce data typically \
found in Mac's mount command"""
_required_framework_version = (2, 0, 0)
+1 -2
View File
@@ -10,8 +10,7 @@ from volatility3.plugins.mac import pslist
class PsTree(plugins.PluginInterface):
"""Plugin for listing processes in a tree based on their parent process
ID."""
"""Plugin for listing processes in a tree based on their parent process ID."""
_required_framework_version = (2, 0, 0)
+2 -2
View File
@@ -41,8 +41,8 @@ class TimeLinerInterface(metaclass=abc.ABCMeta):
class Timeliner(interfaces.plugins.PluginInterface):
"""Runs all relevant plugins that provide time related information and
orders the results by time."""
"""Runs all relevant plugins that provide time related information and \
orders the results by time."""
_required_framework_version = (2, 0, 0)
_version = (1, 1, 0)
@@ -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(
@@ -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
@@ -22,7 +22,7 @@ class Cachedump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -43,16 +43,16 @@ class Cachedump(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_nlkm(
sechive: registry.RegistryHive, lsakey: bytes, is_vista_or_later: bool
cls, sechive: registry.RegistryHive, lsakey: bytes, is_vista_or_later: bool
):
return lsadump.Lsadump.get_secret_by_name(
sechive, "NL$KM", lsakey, is_vista_or_later
)
@staticmethod
def decrypt_hash(edata: bytes, nlkm: bytes, ch, xp: bool):
@classmethod
def decrypt_hash(cls, edata: bytes, nlkm: bytes, ch, xp: bool):
if xp:
hmac_md5 = HMAC.new(nlkm, ch)
rc4key = hmac_md5.digest()
@@ -69,8 +69,8 @@ class Cachedump(interfaces.plugins.PluginInterface):
data += aes.decrypt(buf)
return data
@staticmethod
def parse_cache_entry(cache_data: bytes) -> Tuple[int, int, int, bytes, bytes]:
@classmethod
def parse_cache_entry(cls, cache_data: bytes) -> Tuple[int, int, int, bytes, bytes]:
(uname_len, domain_len) = unpack("<HH", cache_data[:4])
if len(cache_data[60:62]) == 0:
return (uname_len, domain_len, 0, b"", b"")
@@ -79,9 +79,9 @@ class Cachedump(interfaces.plugins.PluginInterface):
enc_data = cache_data[96:]
return (uname_len, domain_len, domain_name_len, enc_data, ch)
@staticmethod
@classmethod
def parse_decrypted_cache(
dec_data: bytes, uname_len: int, domain_len: int, domain_name_len: int
cls, dec_data: bytes, uname_len: int, domain_len: int, domain_name_len: int
) -> Tuple[str, str, str, bytes]:
"""Get the data from the cache and separate it into the username, domain name, and hash data"""
uname_offset = 72
@@ -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,
@@ -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)
@@ -53,7 +53,7 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
"""Detects the Direct System Call technique used to bypass EDRs"""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
# DLLs that are expected to host system call invocations
valid_syscall_handlers = ("ntdll.dll", "win32u.dll")
@@ -200,8 +200,8 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return disasm_bytes, end_inst
@staticmethod
def get_disasm_function(architecture: str) -> Callable:
@classmethod
def get_disasm_function(cls, architecture: str) -> Callable:
"""
Returns the disassembly handler for the given architecture
.detail is used to get full instruction information
@@ -284,8 +284,9 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return None
@staticmethod
@classmethod
def get_vad_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> List[Tuple[int, int, str]]:
"""Creates a map of start/end addresses within a virtual address
@@ -310,9 +311,9 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return vads
@staticmethod
@classmethod
def get_range_path(
ranges: List[Tuple[int, int, str]], address: int
cls, ranges: List[Tuple[int, int, str]], address: int
) -> Optional[str]:
"""
Returns the path for the range holding `address`, if found
@@ -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)
@@ -76,14 +76,14 @@ class Envars(interfaces.plugins.PluginInterface):
"CurrentControlSet\\Control\\Session Manager\\Environment"
)
sys = True
except KeyError:
with contextlib.suppress(KeyError):
except (KeyError, registry.RegistryFormatException):
with contextlib.suppress(KeyError, registry.RegistryFormatException):
key = hive.get_key(
"ControlSet001\\Control\\Session Manager\\Environment"
)
sys = True
if sys:
with contextlib.suppress(KeyError):
with contextlib.suppress(KeyError, registry.RegistryFormatException):
for node in key.get_values():
try:
value_node_name = node.get_name()
@@ -100,11 +100,11 @@ class Envars(interfaces.plugins.PluginInterface):
continue
## The user-specific variables
with contextlib.suppress(KeyError):
with contextlib.suppress(KeyError, registry.RegistryFormatException):
key = hive.get_key("Environment")
ntuser = True
if ntuser:
with contextlib.suppress(KeyError):
with contextlib.suppress(KeyError, registry.RegistryFormatException):
for node in key.get_values():
try:
value_node_name = node.get_name()
@@ -123,7 +123,7 @@ class Envars(interfaces.plugins.PluginInterface):
## The volatile user variables
try:
key = hive.get_key("Volatile Environment")
except KeyError:
except (KeyError, registry.RegistryFormatException):
continue
try:
for node in key.get_values():
@@ -10,6 +10,7 @@ from typing import List
from volatility3.framework import renderers, interfaces, constants, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
@@ -86,10 +87,18 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
# Get ControlSet\Services.
try:
services = hive.get_key(r"CurrentControlSet\Services")
except (KeyError, exceptions.InvalidAddressException):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
):
try:
services = hive.get_key(r"ControlSet001\Services")
except (KeyError, exceptions.InvalidAddressException):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
):
continue
if services:
@@ -158,7 +158,11 @@ class GetSIDs(interfaces.plugins.PluginInterface):
layers.registry.RegistryFormatException,
):
continue
except (KeyError, exceptions.InvalidAddressException):
except (
KeyError,
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
):
continue
return sids
@@ -341,7 +341,7 @@ class Handles(interfaces.plugins.PluginInterface):
try:
obj_name = entry.NameInfo.Name.String
except (ValueError, exceptions.InvalidAddressException):
obj_name = ""
obj_name = None
except exceptions.InvalidAddressException:
vollog.log(
@@ -359,7 +359,7 @@ class Handles(interfaces.plugins.PluginInterface):
format_hints.Hex(entry.HandleValue),
obj_type,
format_hints.Hex(entry.GrantedAccess),
obj_name,
obj_name or renderers.NotAvailableValue(),
),
)
@@ -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"
)
@@ -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
@@ -22,7 +22,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -37,8 +37,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
]
@staticmethod
@classmethod
def enumerate_mft_records(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
primary_layer_name: str,
@@ -128,8 +129,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
layer_name=layer.name,
)
@staticmethod
@classmethod
def parse_mft_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -191,8 +193,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_name,
)
@staticmethod
@classmethod
def parse_data_record(
cls,
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
record_map: Dict[int, Tuple[str, int, int]],
@@ -325,7 +328,7 @@ class ADS(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -343,8 +346,9 @@ class ADS(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def parse_ads_data_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -394,7 +398,7 @@ class ResidentData(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -412,8 +416,9 @@ class ResidentData(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def parse_first_data_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -23,7 +23,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for network objects present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -50,9 +50,9 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
]
@staticmethod
@classmethod
def create_netscan_constraints(
context: interfaces.context.ContextInterface, symbol_table: str
cls, context: interfaces.context.ContextInterface, symbol_table: str
) -> List[poolscanner.PoolConstraint]:
"""Creates a list of Pool Tag Constraints for network objects.
@@ -244,7 +244,7 @@ class PESymbols(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
# used for special handling of the kernel PDB file. See later notes
os_module_name = "ntoskrnl.exe"
@@ -330,9 +330,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return pe_ret
@staticmethod
@classmethod
def range_info_for_address(
ranges: ranges_type, address: int
cls, ranges: ranges_type, address: int
) -> Optional[range_type]:
"""
Helper for getting the range information for an address.
@@ -351,8 +351,8 @@ class PESymbols(interfaces.plugins.PluginInterface):
return None
@staticmethod
def filepath_for_address(ranges: ranges_type, address: int) -> Optional[str]:
@classmethod
def filepath_for_address(cls, ranges: ranges_type, address: int) -> Optional[str]:
"""
Helper to get the file path for an address
@@ -369,8 +369,8 @@ class PESymbols(interfaces.plugins.PluginInterface):
return None
@staticmethod
def filename_for_path(filepath: str) -> str:
@classmethod
def filename_for_path(cls, filepath: str) -> str:
"""
Consistent way to get the filename regardless of platform
@@ -382,8 +382,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
"""
return ntpath.basename(filepath).lower()
@staticmethod
@classmethod
def addresses_for_process_symbols(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
layer_name: str,
@@ -416,8 +417,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_symbols
@staticmethod
@classmethod
def path_and_symbol_for_address(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
collected_modules: collected_modules_type,
@@ -733,8 +735,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found, remaining
@staticmethod
@classmethod
def find_symbols(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
wanted_modules: PESymbolFinder.cached_value_dict,
@@ -775,8 +778,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_symbols, missing_symbols
@staticmethod
@classmethod
def get_kernel_modules(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
@@ -837,8 +841,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_modules
@staticmethod
@classmethod
def get_vads_for_process_cache(
cls,
vads_cache: Dict[int, ranges_type],
owner_proc: interfaces.objects.ObjectInterface,
) -> Optional[ranges_type]:
@@ -865,8 +870,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return vads
@staticmethod
@classmethod
def get_proc_vads_with_file_paths(
cls,
proc: interfaces.objects.ObjectInterface,
) -> ranges_type:
"""
@@ -928,8 +934,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
yield proc, proc_layer_name, vads
@staticmethod
@classmethod
def get_process_modules(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
@@ -127,8 +127,8 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface):
class PoolScanner(plugins.PluginInterface):
"""A generic pool scanner plugin."""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -181,9 +181,9 @@ class PoolScanner(plugins.PluginInterface):
),
)
@staticmethod
@classmethod
def builtin_constraints(
symbol_table: str, tags_filter: Optional[List[bytes]] = None
cls, symbol_table: str, tags_filter: Optional[List[bytes]] = None
) -> List[PoolConstraint]:
"""Get built-in PoolConstraints given a list of pool tags.
@@ -14,8 +14,7 @@ vollog = logging.getLogger(__name__)
class PsTree(interfaces.plugins.PluginInterface):
"""Plugin for listing processes in a tree based on their parent process
ID."""
"""Plugin for listing processes in a tree based on their parent process ID."""
_required_framework_version = (2, 0, 0)
@@ -21,9 +21,10 @@ vollog = logging.getLogger(__name__)
class PsXView(plugins.PluginInterface):
"""Lists all processes found via four of the methods described in \"The Art of Memory Forensics,\" which may help
identify processes that are trying to hide themselves. I recommend using -r pretty if you are looking at this
plugin's output in a terminal."""
"""Lists all processes found via four of the methods described in \"The Art of Memory Forensics\" which may help \
identify processes that are trying to hide themselves.
We recommend using -r pretty if you are looking at this plugin's output in a terminal."""
# I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the functionality
# which the original plugin used to do it.
@@ -12,8 +12,7 @@ from volatility3.plugins.windows import poolscanner, bigpools
class HiveScan(interfaces.plugins.PluginInterface):
"""Scans for registry hives present in a particular windows memory
image."""
"""Scans for registry hives present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@@ -13,7 +13,7 @@ from typing import Any, Generator, List, Tuple
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers.physical import BufferDataLayer
from volatility3.framework.layers.registry import RegistryHive
from volatility3.framework.layers.registry import RegistryHive, RegistryFormatException
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
from volatility3.plugins.windows.registry import hivelist
@@ -167,10 +167,21 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
self._determine_userassist_type()
userassist_node_path = hive.get_key(
"software\\microsoft\\windows\\currentversion\\explorer\\userassist",
return_list=True,
)
try:
userassist_node_path = hive.get_key(
"software\\microsoft\\windows\\currentversion\\explorer\\userassist",
return_list=True,
)
except RegistryFormatException as e:
vollog.warning(
f"Error accessing UserAssist key in {hive_name} at {hive.hive_offset:#x}: {e}"
)
return None
except KeyError:
vollog.warning(
f"UserAssist key not found in {hive_name} at {hive.hive_offset:#x}"
)
return None
if not userassist_node_path:
vollog.warning("list_userassist did not find a valid node_path (or None)")
@@ -1099,9 +1099,8 @@ class DynamicInfo:
class ScheduledTasks(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Decodes scheduled task information from the Windows registry, including
information about triggers, actions, run times, and creation times.
"""
"""Decodes scheduled task information from the Windows registry, including \
information about triggers, actions, run times, and creation times."""
_required_framework_version = (2, 11, 0)
_version = (1, 0, 0)
@@ -24,6 +24,7 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
"""Reads Shimcache entries from the ahcache.sys AVL tree"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
# These checks must be completed from newest -> oldest OS version.
_win_version_file_map: List[Tuple[versions.OsDistinguisher, bool, str]] = [
@@ -74,8 +75,9 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
),
]
@staticmethod
@classmethod
def create_shimcache_table(
cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
config_path: str,
@@ -305,14 +307,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 +326,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 +410,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 +421,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 +432,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 +442,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)
@@ -15,7 +15,7 @@ from volatility3.framework import (
symbols,
)
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import scanners
from volatility3.framework.layers import scanners, registry
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import versions
@@ -35,7 +35,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
"""Scans for windows services."""
_required_framework_version = (2, 0, 0)
_version = (3, 0, 1)
_version = (3, 0, 2)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -61,8 +61,9 @@ class SvcScan(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_record_tuple(
cls,
service_record: interfaces.objects.ObjectInterface,
binary_info: ServiceBinaryInfo,
):
@@ -159,12 +160,20 @@ class SvcScan(interfaces.plugins.PluginInterface):
return cast(
objects.StructType, hive.get_key(r"CurrentControlSet\Services")
)
except (KeyError, exceptions.InvalidAddressException):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
):
try:
return cast(
objects.StructType, hive.get_key(r"ControlSet001\Services")
)
except (KeyError, exceptions.InvalidAddressException):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
):
vollog.log(
constants.LOGLEVEL_VVVV,
"Could not retrieve any control set from SYSTEM hive",
@@ -22,7 +22,7 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
"""Lists the unloaded kernel modules."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -34,8 +34,9 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
),
]
@staticmethod
@classmethod
def create_unloadedmodules_table(
cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
config_path: str,
@@ -18,7 +18,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
"""Scans all the Virtual Address Descriptor memory maps using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 1, 1)
_version = (1, 1, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -84,7 +84,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
if not vad_maps_to_scan:
vollog.warning(
f"No VADs were found for task {task.UniqueProcessID}, not scanning"
f"No VADs were found for task {task.UniqueProcessId}, not scanning"
)
continue
@@ -104,8 +104,9 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
value,
)
@staticmethod
@classmethod
def get_vad_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[int, int]]:
"""Creates a map of start/end addresses within a virtual address
+7 -7
View File
@@ -37,7 +37,7 @@ except ImportError:
class YaraScanner(interfaces.layers.ScannerInterface):
_version = (2, 1, 0)
_version = (2, 1, 1)
# yara.Rules isn't exposed, so we can't type this properly
def __init__(self, rules) -> None:
@@ -79,23 +79,23 @@ class YaraScanner(interfaces.layers.ScannerInterface):
for offset, name, value in match.strings:
yield (offset + data_offset, match.rule, name, value)
@staticmethod
def get_rule(rule):
@classmethod
def get_rule(cls, rule):
if USE_YARA_X:
return yara_x.compile(f"rule r1 {{strings: $a = {rule} condition: $a}}")
return yara.compile(
sources={"n": f"rule r1 {{strings: $a = {rule} condition: $a}}"}
)
@staticmethod
def from_compiled_file(filepath):
@classmethod
def from_compiled_file(cls, filepath):
with resources.ResourceAccessor().open(filepath, "rb") as fp:
if USE_YARA_X:
return yara_x.Rules.deserialize_from(file=fp)
return yara.load(file=fp)
@staticmethod
def from_file(filepath):
@classmethod
def from_file(cls, filepath):
with resources.ResourceAccessor().open(filepath, "rb") as fp:
if USE_YARA_X:
return yara_x.compile(fp.read().decode())
@@ -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:
+16 -7
View File
@@ -411,18 +411,27 @@ class Version1Format(ISFormatTable):
@property
def symbols(self) -> Iterable[str]:
"""Returns an iterator of the symbol names."""
return list(self._json_object.get("symbols", {}))
"""Returns an iterable (KeysView) of the available symbol names."""
return self._json_object.get("symbols", {}).keys()
@property
def enumerations(self) -> Iterable[str]:
"""Returns an iterator of the available enumerations."""
return list(self._json_object.get("enums", {}))
def enumerations(self) -> Iterable[Any]:
"""Returns an iterable (KeysView) of the available enumerations."""
return self._json_object.get("enums", {}).keys()
@property
def types(self) -> Iterable[str]:
"""Returns an iterator of the symbol type names."""
return list(self._json_object.get("user_types", {})) + list(self.natives.types)
"""Returns an iterable (KeysView) of the available symbol type names."""
# We use ** instead of
# `set(self._json_object.get("user_types", {}).keys()).union(self.natives.types)`
# because converting user_types dict to a set is costly.
# It is more efficient to convert the (very small) self.natives.types set to a dict.
# FIXME: On Python3.8 support drop, merge the two dicts using the merge operator:
# (self._json_object.get("user_types", {}) | dict.fromkeys(self.natives.types)).keys()
return {
**self._json_object.get("user_types", {}),
**dict.fromkeys(self.natives.types),
}.keys()
def get_type_class(self, name: str) -> Type[interfaces.objects.ObjectInterface]:
return self._overrides.get(name, objects.AggregateType)
+107 -68
View File
@@ -3,15 +3,26 @@
#
import math
import contextlib
import functools
import logging
from abc import ABC, abstractmethod
from typing import Iterator, List, Tuple, Optional, Union
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3 import framework
from volatility3.framework import constants, exceptions, interfaces, objects
from volatility3.framework import (
constants,
exceptions,
interfaces,
objects,
Deprecation,
)
from volatility3.framework.objects import utility
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.linux import extensions
vollog = logging.getLogger(__name__)
class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
provides = {"type": "interface"}
@@ -43,6 +54,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
self.optional_set_type_class("bpf_prog_aux", extensions.bpf_prog_aux)
self.optional_set_type_class("kernel_cap_struct", extensions.kernel_cap_struct)
self.optional_set_type_class("kernel_cap_t", extensions.kernel_cap_t)
self.optional_set_type_class("scatterlist", extensions.scatterlist)
# kernels >= 4.18
self.optional_set_type_class("timespec64", extensions.timespec64)
@@ -76,7 +88,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
class LinuxUtilities(interfaces.configuration.VersionableInterface):
"""Class with multiple useful linux functions."""
_version = (2, 2, 0)
_version = (2, 3, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@@ -106,8 +118,8 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
Args:
task (task_struct): A reference task
mnt (vfsmount or mount): A mounted filesystem or a mount point.
- kernels < 3.3.8 type is 'vfsmount'
- kernels >= 3.3.8 type is 'mount'
- kernels < 3.3 type is 'vfsmount'
- kernels >= 3.3 type is 'mount'
Returns:
str: Pathname of the mount point relative to the task's root directory.
@@ -129,14 +141,28 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
rdentry (dentry *): A pointer to the root dentry
rmnt (vfsmount *): A pointer to the root vfsmount
dentry (dentry *): A pointer to the dentry
vfsmnt (vfsmount *): A pointer to the vfsmount
vfsmnt (vfsmount/vfsmount *): A vfsmount object (kernels >= 3.3) or a
vfsmount pointer (kernels < 3.3)
Returns:
str: Pathname of the mount point or file
"""
if not (rdentry and rdentry.is_readable() and rmnt and rmnt.is_readable()):
return ""
if isinstance(vfsmnt, objects.Pointer) and not (
vfsmnt and vfsmnt.is_readable()
):
# vfsmnt can be the vfsmount object itself (>=3.3) or a vfsmount * (<3.3)
return ""
path_reversed = []
while dentry != rdentry or not vfsmnt.is_equal(rmnt):
while (
dentry
and dentry.is_readable()
and (dentry != rdentry or not vfsmnt.is_equal(rmnt))
):
if dentry == vfsmnt.get_mnt_root() or dentry.is_root():
# Escaped?
if dentry != vfsmnt.get_mnt_root():
@@ -334,6 +360,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
yield fd_num, filp, full_path
@classmethod
@Deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.mask_mods_list,
replacement_version=(1, 0, 0),
)
def mask_mods_list(
cls,
context: interfaces.context.ContextInterface,
@@ -341,18 +371,11 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
mods: Iterator[interfaces.objects.ObjectInterface],
) -> List[Tuple[str, int, int]]:
"""
DEPRECATED: use "volatility3.framework.symbols.linux.utilities.modules.Modules.mask_mods_list" instead.
A helper function to mask the starting and end address of kernel modules
"""
mask = context.layers[layer_name].address_mask
return [
(
utility.array_to_string(mod.name),
mod.get_module_base() & mask,
(mod.get_module_base() & mask) + mod.get_core_size(),
)
for mod in mods
]
return linux_utilities_modules.Modules.mask_mods_list(context, layer_name, mods)
@classmethod
def generate_kernel_handler_info(
@@ -377,41 +400,30 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
return [
(constants.linux.KERNEL_NAME, start_addr, end_addr)
] + LinuxUtilities.mask_mods_list(context, kernel.layer_name, mods_list)
] + linux_utilities_modules.Modules.mask_mods_list(
context, kernel.layer_name, mods_list
)
@classmethod
@Deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.lookup_module_address,
replacement_version=(1, 0, 0),
)
def lookup_module_address(
cls,
kernel_module: interfaces.context.ModuleInterface,
handlers: List[Tuple[str, int, int]],
target_address: int,
):
) -> Tuple[str, str]:
"""
DEPRECATED: use "volatility3.framework.symbols.linux.utilities.modules.Modules.lookup_module_address" instead.
Searches between the start and end address of the kernel module using target_address.
Returns the module and symbol name of the address provided.
"""
mod_name = "UNKNOWN"
symbol_name = "N/A"
for name, start, end in handlers:
if start <= target_address <= end:
mod_name = name
if name == constants.linux.KERNEL_NAME:
symbols = list(
kernel_module.get_symbols_by_absolute_location(target_address)
)
if len(symbols):
symbol_name = (
symbols[0].split(constants.BANG)[1]
if constants.BANG in symbols[0]
else symbols[0]
)
break
return mod_name, symbol_name
return linux_utilities_modules.Modules.lookup_module_address(
kernel_module.context, kernel_module.name, handlers, target_address
)
@classmethod
def walk_internal_list(cls, vmlinux, struct_name, list_member, list_start):
@@ -449,6 +461,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
type_dec = vmlinux.get_type(type_name)
member_offset = type_dec.relative_child_offset(member_name)
container_addr = addr - member_offset
layer = vmlinux.context.layers[vmlinux.layer_name]
if not layer.is_valid(container_addr):
return None
return vmlinux.object(
object_type=type_name, offset=container_addr, absolute=True
)
@@ -612,7 +628,7 @@ class IDStorage(ABC):
raise NotImplementedError
def nodep_to_node(self, nodep) -> interfaces.objects.ObjectInterface:
"""Instanciates a tree node from its pointer
"""Instantiates a tree node from its pointer
Args:
nodep: Pointer to the XArray/RadixTree node
@@ -659,7 +675,7 @@ class IDStorage(ABC):
height = self.get_tree_height(root.vol.offset)
nodep = self.get_head_node(root)
if not nodep:
if not (nodep and nodep.is_readable()):
return
# Keep the internal flag before untagging it
@@ -694,7 +710,7 @@ class XArray(IDStorage):
def get_node_height(self, nodep) -> int:
node = self.nodep_to_node(nodep)
return (node.shift / self.CHUNK_SHIFT) + 1
return (node.shift // self.CHUNK_SHIFT) + 1
def get_head_node(self, tree) -> int:
return tree.xa_head
@@ -717,6 +733,7 @@ class RadixTree(IDStorage):
RADIX_TREE_INTERNAL_NODE = 1
RADIX_TREE_EXCEPTIONAL_ENTRY = 2
RADIX_TREE_ENTRY_MASK = 3
RADIX_TREE_MAP_SHIFT = 6 # CONFIG_BASE_FULL
# Dynamic values. These will be initialized later
RADIX_TREE_INDEX_BITS = None
@@ -753,43 +770,57 @@ class RadixTree(IDStorage):
def get_tree_height(self, treep) -> int:
with contextlib.suppress(exceptions.SymbolError):
if self.vmlinux.get_type("radix_tree_root").has_member("height"):
# kernels < 4.7.10
# kernels < 4.7 d0891265bbc988dc91ed8580b38eb3dac128581b
radix_tree_root = self.vmlinux.object(
"radix_tree_root", offset=treep, absolute=True
)
return radix_tree_root.height
# kernels >= 4.7.10
# kernels >= 4.7
return 0
@functools.cached_property
def _max_height_array(self):
if self.vmlinux.has_symbol("height_to_maxindex"):
# 2.6.24 26fb1589cb0aaec3a0b4418c54f30c1a2b1781f6 <= Kernels < 4.7 d0891265bbc988dc91ed8580b38eb3dac128581b
return self.vmlinux.object_from_symbol("height_to_maxindex")
elif self.vmlinux.has_symbol("height_to_maxnodes"):
# 4.8 c78c66d1ddfdbd2353f3fcfeba0268524537b096 <= kernels < 4.20 8cf2f98411e3a0865026a1061af637161b16d32b
return self.vmlinux.object_from_symbol("height_to_maxnodes")
return None
def _radix_tree_maxindex(self, node, height) -> int:
"""Return the maximum key which can be store into a radix tree with this height."""
if not self.vmlinux.has_symbol("height_to_maxindex"):
# Kernels >= 4.7
return (self.CHUNK_SIZE << node.shift) - 1
if self._max_height_array:
# 2.6.24 <= kernels <= 4.20 See _max_height_array()
return self._max_height_array[height]
else:
# Kernels < 4.7
height_to_maxindex_array = self.vmlinux.object_from_symbol(
"height_to_maxindex"
)
maxindex = height_to_maxindex_array[height]
return maxindex
# Kernels >= 4.20
return (self.CHUNK_SIZE << node.shift) - 1
def get_node_height(self, nodep) -> int:
node = self.nodep_to_node(nodep)
if hasattr(node, "shift"):
# 4.7 <= Kernels < 4.20
return (node.shift / self.CHUNK_SHIFT) + 1
height = (node.shift // self.CHUNK_SHIFT) + 1
elif hasattr(node, "path"):
# 3.15 <= Kernels < 4.7
return node.path & self.RADIX_TREE_HEIGHT_MASK
height = node.path & self.RADIX_TREE_HEIGHT_MASK
elif hasattr(node, "height"):
# Kernels < 3.15
return node.height
height = node.height
else:
raise exceptions.VolatilityException("Cannot find radix-tree node height")
if self._max_height_array and not (0 <= height < self._max_height_array.count):
error_msg = f"Radix Tree node {node.vol.offset:#x} height {height} exceeds max height of {self._max_height_array.count}"
vollog.error(error_msg)
raise exceptions.LinuxPageCacheException(error_msg)
return height
def get_head_node(self, tree) -> int:
return tree.rnode
@@ -802,14 +833,16 @@ class RadixTree(IDStorage):
def untag_node(self, nodep) -> int:
return nodep & (~self.RADIX_TREE_ENTRY_MASK)
def is_valid_node(self, nodep) -> bool:
def _is_exceptional_node(self, nodep) -> bool:
# In kernels 4.20, exceptional nodes were removed and internal entries took their bitmask
if self.vmlinux.has_type("radix_tree_root"):
return (
nodep & self.RADIX_TREE_ENTRY_MASK
) != self.RADIX_TREE_EXCEPTIONAL_ENTRY
return (
self.vmlinux.has_type("radix_tree_root")
and (nodep & self.RADIX_TREE_ENTRY_MASK)
== self.RADIX_TREE_EXCEPTIONAL_ENTRY
)
return True
def is_valid_node(self, nodep) -> bool:
return not self._is_exceptional_node(nodep)
class PageCache:
@@ -838,11 +871,17 @@ class PageCache:
Yields:
Page objects
"""
layer = self.vmlinux.context.layers[self.vmlinux.layer_name]
for page_addr in self._idstorage.get_entries(self._page_cache.i_pages):
if not page_addr:
continue
if not layer.is_valid(page_addr):
error_msg = f"Invalid cached page address at {page_addr:#x}, aborting"
vollog.error(error_msg)
raise exceptions.LinuxPageCacheException(error_msg)
page = self.vmlinux.object("page", offset=page_addr, absolute=True)
if page:
yield page
if not page.is_valid():
error_msg = f"Invalid cached page at {page_addr:#x}, aborting"
vollog.error(error_msg)
raise exceptions.LinuxPageCacheException(error_msg)
yield page
@@ -15,12 +15,11 @@ 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
vollog = logging.getLogger(__name__)
# Keep these in a basic module, to prevent import cycles when symbol providers require them
@@ -308,6 +307,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]:
@@ -325,9 +364,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
@@ -400,6 +441,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
@@ -810,23 +853,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]:
@@ -841,7 +891,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(
@@ -849,20 +903,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")
@@ -1151,19 +1212,15 @@ class struct_file(objects.StructType):
"""Returns a pointer to the dentry associated with this file"""
if self.has_member("f_path"):
return self.f_path.dentry
elif self.has_member("f_dentry"):
return self.f_dentry
else:
raise AttributeError("Unable to find file -> dentry")
raise AttributeError("Unable to find file -> dentry")
def get_vfsmnt(self) -> interfaces.objects.ObjectInterface:
"""Returns the fs (vfsmount) where this file is mounted"""
if self.has_member("f_path"):
return self.f_path.mnt
elif self.has_member("f_vfsmnt"):
return self.f_vfsmnt
else:
raise AttributeError("Unable to find file -> vfs mount")
raise AttributeError("Unable to find file -> vfs mount")
def get_inode(self) -> interfaces.objects.ObjectInterface:
"""Returns an inode associated with this file"""
@@ -1208,35 +1265,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)
@@ -1393,9 +1458,9 @@ class mount(objects.StructType):
A dentry pointer
"""
vfsmnt = self.get_vfsmnt_current()
dentry = vfsmnt.mnt_root
dentry_pointer = vfsmnt.mnt_root
return dentry
return dentry_pointer
def get_dentry_parent(self):
"""Returns the parent root of the mounted tree
@@ -1503,39 +1568,38 @@ class vfsmount(objects.StructType):
)
def _is_kernel_prior_to_struct_mount(self) -> bool:
"""Helper to distinguish between kernels prior to version 3.3.8 that
lacked the 'mount' structure and later versions that have it.
"""Helper to distinguish between kernels prior to version 3.3 which lacked the
'mount' struct, versus later versions that include it.
See 7d6fec45a5131918b51dcd76da52f2ec86a85be6.
The 'mnt_parent' member was moved from struct 'vfsmount' to struct
'mount' when the latter was introduced.
Alternatively, vmlinux.has_type('mount') can be used here but it is faster.
# Following that commit, also in kernel version 3.3 (3376f34fff5be9954fd9a9c4fd68f4a0a36d480e),
# the 'mnt_parent' member was relocated from the 'vfsmount' struct to the newly
# introduced 'mount' struct.
Returns:
bool: 'True' if the kernel
'True' if the kernel lacks the 'mount' struct, typically indicating kernel < 3.3.
"""
return self.has_member("mnt_parent")
return not self._context.symbol_space.has_type("mount")
def is_equal(self, vfsmount_ptr) -> bool:
"""Helper to make sure it is comparing two pointers to 'vfsmount'.
Depending on the kernel version, the calling object (self) could be
a 'vfsmount \\*' (<3.3.8) or a 'vfsmount' (>=3.3.8). This way we trust
in the framework "auto" dereferencing ability to assure that when we
reach this point 'self' will be a 'vfsmount' already and self.vol.offset
Depending on the kernel version, see 3376f34fff5be9954fd9a9c4fd68f4a0a36d480e,
the calling object (self) could be a 'vfsmount \\*' (<3.3) or a 'vfsmount' (>=3.3).
This way we trust in the framework "auto" dereferencing ability to assure that
when we reach this point 'self' will be a 'vfsmount' already and self.vol.offset
a 'vfsmount \\*' and not a 'vfsmount \\*\\*'. The argument must be a 'vfsmount \\*'.
Typically, it's called from do_get_path().
Args:
vfsmount_ptr (vfsmount *): A pointer to a 'vfsmount'
vfsmount_ptr: A pointer to a 'vfsmount'
Raises:
exceptions.VolatilityException: If vfsmount_ptr is not a 'vfsmount \\*'
Returns:
bool: 'True' if the given argument points to the the same 'vfsmount'
as 'self'.
'True' if the given argument points to the same 'vfsmount' as 'self'.
"""
if isinstance(vfsmount_ptr, objects.Pointer):
return self.vol.offset == vfsmount_ptr
@@ -1544,13 +1608,14 @@ class vfsmount(objects.StructType):
"Unexpected argument type. It has to be a 'vfsmount *'"
)
def _get_real_mnt(self):
def _get_real_mnt(self) -> interfaces.objects.ObjectInterface:
"""Gets the struct 'mount' containing this 'vfsmount'.
It should be only called from kernels >= 3.3.8 when 'struct mount' was introduced.
It should be only called from kernels >= 3.3 when 'struct mount' was introduced.
See 7d6fec45a5131918b51dcd76da52f2ec86a85be6
Returns:
mount: the struct 'mount' containing this 'vfsmount'.
The 'mount' object containing this 'vfsmount'.
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
return linux.LinuxUtilities.container_of(
@@ -1569,8 +1634,8 @@ class vfsmount(objects.StructType):
"""Gets the parent fs (vfsmount) to where it's mounted on
Returns:
For kernels < 3.3.8: A vfsmount pointer
For kernels >= 3.3.8: A vfsmount object
For kernels < 3.3: A vfsmount pointer
For kernels >= 3.3: A vfsmount object
"""
if self._is_kernel_prior_to_struct_mount():
return self.get_mnt_parent()
@@ -1603,8 +1668,8 @@ class vfsmount(objects.StructType):
"""Gets the mnt_parent member.
Returns:
For kernels < 3.3.8: A vfsmount pointer
For kernels >= 3.3.8: A mount pointer
For kernels < 3.3: A vfsmount pointer
For kernels >= 3.3: A mount pointer
"""
if self._is_kernel_prior_to_struct_mount():
return self.mnt_parent
@@ -1675,8 +1740,10 @@ class kobject(objects.StructType):
class mnt_namespace(objects.StructType):
def get_inode(self):
if self.has_member("proc_inum"):
# 98f842e675f96ffac96e6c50315790912b2812be 3.8 <= kernels < 3.19
return self.proc_inum
elif self.has_member("ns") and self.ns.has_member("inum"):
# kernels >= 3.19 435d5f4bb2ccba3b791d9ef61d2590e30b8e806e
return self.ns.inum
else:
raise AttributeError("Unable to find mnt_namespace inode")
@@ -2022,8 +2089,11 @@ class bpf_prog(objects.StructType):
prog_tag_addr = self.tag.vol.offset
prog_tag_size = self.tag.count
prog_tag_bytes = vmlinux_layer.read(prog_tag_addr, prog_tag_size)
if not vmlinux_layer.is_valid(prog_tag_addr, prog_tag_size):
vollog.debug("Unable to read bpf tag string from 0x%x", prog_tag_addr)
return None
prog_tag_bytes = vmlinux_layer.read(prog_tag_addr, prog_tag_size)
prog_tag = binascii.hexlify(prog_tag_bytes).decode()
return prog_tag
@@ -2488,7 +2558,12 @@ class inode(objects.StructType):
"""
if not self.i_size:
return
elif not (self.i_mapping and self.i_mapping.nrpages > 0):
if not (
self.i_mapping
and self.i_mapping.is_readable()
and self.i_mapping.nrpages > 0
):
return
page_cache = linux.PageCache(
@@ -2496,19 +2571,26 @@ class inode(objects.StructType):
kernel_module_name="kernel",
page_cache=self.i_mapping.dereference(),
)
yield from page_cache.get_cached_pages()
def get_contents(self):
def get_contents(self) -> Iterable[Tuple[int, bytes]]:
"""Get the inode cached pages from the page cache
Yields:
page_index (int): The page index in the Tree. File offset is page_index * PAGE_SIZE.
page_content (str): The page content
page_content (bytes): The page content
"""
for page_obj in self.get_pages():
if page_obj.mapping != self.i_mapping:
vollog.warning(
f"Cached page at {page_obj.vol.offset:#x} has a mismatched address space with the inode. Skipping page"
)
continue
page_index = int(page_obj.index)
page_content = page_obj.get_content()
yield page_index, page_content
if page_content:
yield page_index, page_content
class address_space(objects.StructType):
@@ -2516,7 +2598,7 @@ class address_space(objects.StructType):
def i_pages(self):
"""Returns the appropriate member containing the page cache tree"""
if self.has_member("i_pages"):
# Kernel >= 4.17
# Kernel >= 4.17 b93b016313b3ba8003c3b8bb71f569af91f19fc7
return self.member("i_pages")
elif self.has_member("page_tree"):
# Kernel < 4.17
@@ -2526,16 +2608,22 @@ class address_space(objects.StructType):
class page(objects.StructType):
@property
@functools.lru_cache
def is_valid(self) -> bool:
if self.mapping and not self.mapping.is_readable():
return False
if self.to_paddr() < 0:
return False
return True
@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"
@@ -2549,24 +2637,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]
@@ -2606,14 +2682,40 @@ 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 get_content(self) -> Union[str, None]:
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[bytes, None]:
"""Returns the page content
Returns:
@@ -2621,13 +2723,34 @@ 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
page_data = physical_layer.read(page_paddr, vmlinux_layer.page_size)
return page_data
if not physical_layer.is_valid(page_paddr, length=vmlinux_layer.page_size):
vollog.debug(
"Unable to read page 0x%x content at 0x%x", self.vol.offset, page_paddr
)
return None
return physical_layer.read(page_paddr, vmlinux_layer.page_size)
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):
@@ -2727,17 +2850,17 @@ class IDR(objects.StructType):
class rb_root(objects.StructType):
def _walk_nodes(self, root_node) -> Iterator[int]:
def _walk_nodes(self, root_node: int) -> Iterator[int]:
"""Traverses the Red-Black tree from the root node and yields a pointer to each
node in this tree.
Args:
root_node: A Red-Black tree node from which to start descending
root_node: A Red-Black tree node pointer from which to start descending
Yields:
A pointer to every node descending from the specified root node
"""
if not root_node:
if not (root_node and root_node.is_readable()):
return
yield root_node
@@ -2752,3 +2875,111 @@ class rb_root(objects.StructType):
"""
yield from self._walk_nodes(root_node=self.rb_node)
class scatterlist(objects.StructType):
SG_CHAIN = 0x01
SG_END = 0x02
SG_PAGE_LINK_MASK = SG_CHAIN | SG_END
def _sg_flags(self) -> int:
return self.page_link & self.SG_PAGE_LINK_MASK
def _sg_is_chain(self) -> int:
return self._sg_flags() & self.SG_CHAIN
def _sg_is_last(self) -> int:
return self._sg_flags() & self.SG_END
def _sg_chain_ptr(self) -> int:
"""Clears the last two bits basically."""
return self.page_link & ~self.SG_PAGE_LINK_MASK
def _sg_dma_len(self) -> int:
# Depends on CONFIG_NEED_SG_DMA_LENGTH
if self.has_member("dma_length"):
return self.dma_length
return self.length
def _get_sg_max_single_alloc(self) -> int:
"""Based on kernel's SG_MAX_SINGLE_ALLOC.
Doc. from kernel source :
* Maximum number of entries that will be allocated in one piece, if
* a list larger than this is required then chaining will be utilized.
"""
return self._context.layers[self.vol.layer_name].page_size // self.vol.size
def _sg_next(self) -> Optional[interfaces.objects.ObjectInterface]:
"""Get the next scatterlist struct from the list.
Based on kernel's sg_next.
Doc. from kernel source :
* Notes on SG table design.
*
* We use the unsigned long page_link field in the scatterlist struct to place
* the page pointer AND encode information about the sg table as well. The two
* lower bits are reserved for this information.
*
* If bit 0 is set, then the page_link contains a pointer to the next sg
* table list. Otherwise the next entry is at sg + 1.
*
* If bit 1 is set, then this sg entry is the last element in a list.
"""
if self._sg_is_last():
return None
if self._sg_is_chain():
next_address = self._sg_chain_ptr()
else:
next_address = self.vol.offset + self.vol.size
sg = self._context.object(
self.get_symbol_table_name() + constants.BANG + "scatterlist",
self.vol.layer_name,
next_address,
)
return sg
def for_each_sg(self) -> Optional[Iterator[interfaces.objects.ObjectInterface]]:
"""Iterate over each struct in the scatterlist."""
sg = self
sg_max_single_alloc = self._get_sg_max_single_alloc()
# Empty scatterlists protection
if sg.page_link == 0 and sg._sg_dma_len() == 0 and sg.dma_address == 0:
return None
else:
# Yield itself first
yield sg
entries_count = 1
# entries_count <= sg_max_single_alloc should always be true if the
# scatterlists were correctly chained.
while entries_count <= sg_max_single_alloc:
sg = sg._sg_next()
if sg is None:
break
# Points to a new scatterlist
elif sg._sg_is_chain():
entries_count = 0
else:
entries_count += 1
yield sg
def get_content(
self,
) -> Optional[Iterator[bytes]]:
"""Traverse a scatterlist to gather content located at each
dma_address position.
Returns:
An iterator of bytes
"""
# Either "physical" is layer-1 because this is a module layer, or "physical" is the current 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]
for sg in self.for_each_sg():
yield from physical_layer.read(sg.dma_address, sg._sg_dma_len())
@@ -445,7 +445,7 @@ class elf_linkmap(objects.StructType):
def get_name(self):
try:
buf = self._context.layers.read(self.vol.layer_name, self.l_name, 256)
except exceptions.PagedInvalidAddressException:
except exceptions.InvalidAddressException:
# Protection against memory smear
vollog.log(
constants.LOGLEVEL_VVVV,
@@ -0,0 +1,70 @@
from typing import Iterator, List, Tuple
from volatility3 import framework
from volatility3.framework import constants, interfaces
from volatility3.framework.objects import utility
class Modules(interfaces.configuration.VersionableInterface):
"""Kernel modules related utilities."""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@classmethod
def mask_mods_list(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
mods: Iterator[interfaces.objects.ObjectInterface],
) -> List[Tuple[str, int, int]]:
"""
A helper function to mask the starting and end address of kernel modules
"""
mask = context.layers[layer_name].address_mask
return [
(
utility.array_to_string(mod.name),
mod.get_module_base() & mask,
(mod.get_module_base() & mask) + mod.get_core_size(),
)
for mod in mods
]
@classmethod
def lookup_module_address(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
handlers: List[Tuple[str, int, int]],
target_address: int,
) -> Tuple[str, str]:
"""
Searches between the start and end address of the kernel module using target_address.
Returns the module and symbol name of the address provided.
"""
kernel_module = context.modules[kernel_module_name]
mod_name = "UNKNOWN"
symbol_name = "N/A"
for name, start, end in handlers:
if start <= target_address <= end:
mod_name = name
if name == constants.linux.KERNEL_NAME:
symbols = list(
kernel_module.get_symbols_by_absolute_location(target_address)
)
if len(symbols):
symbol_name = (
symbols[0].split(constants.BANG)[1]
if constants.BANG in symbols[0]
else symbols[0]
)
break
return mod_name, symbol_name
@@ -0,0 +1,161 @@
import functools
from volatility3 import framework
from volatility3.framework import interfaces
from volatility3.framework.constants import linux as linux_constants
from typing import List, Optional
class Tainting(interfaces.configuration.VersionableInterface):
"""Tainted kernel and modules parsing capabilities.
Relevant Linux kernel functions:
- modules: module_flags_taint
- kernel: print_tainted
"""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@classmethod
@functools.lru_cache
def _get_kernel_taint_flags_list(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
) -> Optional[List[interfaces.objects.ObjectInterface]]:
"""Determine whether the kernel embeds taint flags definition
in-memory or not.
Returns:
A list of "taint_flag" kernel objects if taint_flags symbol exists
"""
kernel = context.modules[kernel_module_name]
if kernel.has_symbol("taint_flags"):
return list(kernel.object_from_symbol("taint_flags"))
return None
@classmethod
def _module_flags_taint_pre_4_10_rc1(
cls,
taints: int,
is_module: bool = False,
) -> str:
"""Convert the module's taints value to a 1-1 character mapping.
Relies on statically defined taints mappings in the framework.
Args:
taints: The taints value, represented by an integer
is_module: Indicates if the taints value is associated with a built-in/LKM module
Returns:
The raw taints string.
"""
taints_string = ""
for char, taint_flag in linux_constants.TAINT_FLAGS.items():
if is_module and not taint_flag.module:
continue
if taints & taint_flag.shift:
taints_string += char
return taints_string
@classmethod
def _module_flags_taint_post_4_10_rc1(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
taints: int,
is_module: bool = False,
) -> str:
"""Convert the module's taints value to a 1-1 character mapping.
Relies on kernel symbol embedded taints definitions.
struct taint_flag {
char c_true; /* character printed when tainted */
char c_false; /* character printed when not tainted */
bool module; /* also show as a per-module taint flag */
};
Args:
taints: The taints value, represented by an integer
is_module: Indicates if the taints value is associated with a built-in/LKM module
Returns:
The raw taints string.
"""
taints_string = ""
for taint_bit, taint_flag in enumerate(
cls._get_kernel_taint_flags_list(context, kernel_module_name)
):
if is_module and not taint_flag.module:
continue
c_true = chr(taint_flag.c_true)
c_false = chr(taint_flag.c_false)
if taints & (1 << taint_bit):
taints_string += c_true
elif c_false != " ":
taints_string += c_false
return taints_string
@classmethod
def get_taints_as_plain_string(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
taints: int,
is_module: bool = False,
) -> str:
"""Convert the taints value to a 1-1 character mapping.
Args:
taints: The taints value, represented by an integer
is_module: Indicates if the taints value is associated with a built-in/LKM module
Returns:
The raw taints string.
Documentation:
- module_flags_taint kernel function
"""
if cls._get_kernel_taint_flags_list(context, kernel_module_name):
return cls._module_flags_taint_post_4_10_rc1(
context, kernel_module_name, taints, is_module
)
return cls._module_flags_taint_pre_4_10_rc1(taints, is_module)
@classmethod
def get_taints_parsed(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
taints: int,
is_module: bool = False,
) -> List[str]:
"""Convert the taints string to a 1-1 descriptor mapping.
Args:
taints: The taints value, represented by an integer
is_module: Indicates if the taints value is associated with a built-in/LKM module
Returns:
A comprehensive (user-friendly) taint descriptor list.
Documentation:
- module_flags_taint kernel function
"""
comprehensive_taints = []
for character in cls.get_taints_as_plain_string(
context, kernel_module_name, taints, is_module
):
taint_flag = linux_constants.TAINT_FLAGS.get(character)
if not taint_flag:
comprehensive_taints.append(f"<UNKNOWN_TAINT_CHAR_{character}>")
elif taint_flag.when_present:
comprehensive_taints.append(taint_flag.desc)
return comprehensive_taints
+1 -1
View File
@@ -30,7 +30,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
@property
def types(self) -> Iterable[str]:
"""Returns an iterator of the symbol type names."""
"""Returns an iterable (set) of the available symbol type names."""
return self._types
def get_type(self, type_name: str) -> interfaces.objects.Template:
@@ -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)
@@ -133,8 +133,17 @@ class CM_KEY_BODY(objects.StructType):
def get_full_key_name(self) -> str:
output = []
seen = set()
kcb = self.KeyControlBlock
while kcb.ParentKcb:
if kcb.ParentKcb.vol.offset in seen:
return None
seen.add(kcb.ParentKcb.vol.offset)
if len(output) > 128:
return None
if kcb.NameBlock.Name is None:
break
@@ -159,14 +168,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 +237,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 +269,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 +299,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 +342,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"):
@@ -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())
+10 -1
View File
@@ -14,6 +14,8 @@ vollog = logging.getLogger(__name__)
cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.hashcache")
validators = {}
def load_cached_validations() -> Set[str]:
"""Loads up the list of successfully cached json objects, so we don't need
@@ -93,6 +95,13 @@ def valid(
return True
try:
import jsonschema
schema_key = json.dumps(schema, sort_keys=True)
if schema_key not in validators:
validator_class = jsonschema.validators.validator_for(schema)
validator_class.check_schema(schema)
validator = validator_class(schema)
validators[schema_key] = validator
except ImportError:
vollog.info("Dependency for validation unavailable: jsonschema")
vollog.debug("All validations will report success, even with malformed input")
@@ -100,7 +109,7 @@ def valid(
try:
vollog.debug("Validating JSON against schema...")
jsonschema.validate(input, schema)
validators[schema_key].validate(input)
cached_validations.add(input_hash)
vollog.debug("JSON validated against schema (result cached)")
except jsonschema.exceptions.SchemaError: