Merge branch 'develop' into poolscanner-thread-support

This commit is contained in:
ikelos
2024-03-03 23:48:24 +00:00
committed by GitHub
59 changed files with 3409 additions and 381 deletions
+1 -1
View File
@@ -107,7 +107,7 @@ The latest generated copy of the documentation can be found at: <https://volatil
## Licensing and Copyright
Copyright (C) 2007-2023 Volatility Foundation
Copyright (C) 2007-2024 Volatility Foundation
All Rights Reserved
+1 -1
View File
@@ -169,7 +169,7 @@ master_doc = "index"
# General information about the project.
project = "Volatility 3"
copyright = "2012-2022, Volatility Foundation"
copyright = "2012-2024, 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
+15
View File
@@ -143,3 +143,18 @@ Options
`hivescan` would match `windows.registry.hivescan.HiveScan`, but
`pslist` is ambiguous because it could match `windows.pslist` or
`linux.pslist`.
Overriding options
------------------
The default values for the command line interface are defined by constants within the code,
but can be overridden by creating a JSON file (`%APPDATA%/volatility3/vol.json` for Windows
systems, or `~/.config/volatility3/vol.json` or `volshell.json` for all others).
The format of this file is a JSON dictionary, containing the options above and their value.
It should be noted that the ordering is (`<` means is overridden by):
`in-built default value < config file value < command line parameter`
It should also be noted that boolean flags (such as `offline`) that are overridden as true will
not be unset by not specifying the command line flag.
+79 -15
View File
@@ -19,9 +19,10 @@ import os
import sys
import tempfile
import traceback
from typing import Any, Dict, Type, Union
from typing import Any, Dict, List, Tuple, Type, Union
from urllib import parse, request
from volatility3.cli import text_filter
import volatility3.plugins
import volatility3.symbols
from volatility3 import framework
@@ -105,6 +106,9 @@ class CommandLine:
]
)
# Load up system defaults
delayed_logs, default_config = self.load_system_defaults("vol.json")
parser = volargparse.HelpfulArgParser(
add_help=False,
prog=self.CLI_NAME,
@@ -230,6 +234,14 @@ class CommandLine:
default=False,
action="store_true",
)
parser.add_argument(
"--filters",
help="List of filters to apply to the output (in the form of [+-]columname,pattern[!])",
default=[],
action="append",
)
parser.set_defaults(**default_config)
# We have to filter out help, otherwise parse_known_args will trigger the help message before having
# processed the plugin choice or had the plugin subparser added.
@@ -241,19 +253,7 @@ class CommandLine:
banner_output = sys.stderr
banner_output.write(f"Volatility 3 Framework {constants.PACKAGE_VERSION}\n")
if partial_args.plugin_dirs:
volatility3.plugins.__path__ = [
os.path.abspath(p) for p in partial_args.plugin_dirs.split(";")
] + constants.PLUGINS_PATH
if partial_args.symbol_dirs:
volatility3.symbols.__path__ = [
os.path.abspath(p) for p in partial_args.symbol_dirs.split(";")
] + constants.SYMBOL_BASEPATHS
if partial_args.cache_path:
constants.CACHE_PATH = partial_args.cache_path
### Start up logging
if partial_args.log:
file_logger = logging.FileHandler(partial_args.log)
file_logger.setLevel(1)
@@ -271,6 +271,23 @@ class CommandLine:
else:
console.setLevel(10 - (partial_args.verbosity - 2))
for level, msg in delayed_logs:
vollog.log(level, msg)
### Alter constants if necessary
if partial_args.plugin_dirs:
volatility3.plugins.__path__ = [
os.path.abspath(p) for p in partial_args.plugin_dirs.split(";")
] + constants.PLUGINS_PATH
if partial_args.symbol_dirs:
volatility3.symbols.__path__ = [
os.path.abspath(p) for p in partial_args.symbol_dirs.split(";")
] + constants.SYMBOL_BASEPATHS
if partial_args.cache_path:
constants.CACHE_PATH = partial_args.cache_path
vollog.info(f"Volatility plugins path: {volatility3.plugins.__path__}")
vollog.info(f"Volatility symbols path: {volatility3.symbols.__path__}")
@@ -444,7 +461,10 @@ class CommandLine:
try:
# Construct and run the plugin
if constructed:
renderers[args.renderer]().render(constructed.run())
grid = constructed.run()
renderer = renderers[args.renderer]()
renderer.filter = text_filter.CLIFilter(grid, args.filters)
renderer.render(grid)
except exceptions.VolatilityException as excp:
self.process_exceptions(excp)
@@ -463,6 +483,50 @@ class CommandLine:
)
return requirements.URIRequirement.location_from_file(filename)
def load_system_defaults(
self, filename: str
) -> Tuple[List[Tuple[int, str]], Dict[str, Any]]:
"""Modify the main configuration based on the default configuration override"""
# Build the config path
default_config_path = os.path.join(
os.path.expanduser("~"), ".config", "volatility3", filename
)
if sys.platform == "win32":
default_config_path = os.path.join(
os.environ.get("APPDATA", os.path.expanduser("~")),
"volatility3",
filename,
)
delayed_logs = []
# Process it if the files exist
if os.path.exists(default_config_path):
with open(default_config_path, "rb") as config_json:
result = json.load(config_json)
if not isinstance(result, dict):
delayed_logs.append(
(
logging.INFO,
f"Default configuration file {default_config_path} does not contain a dictionary",
)
)
else:
delayed_logs.append(
(
logging.INFO,
f"Loading default configuration options from {default_config_path}",
)
)
delayed_logs.append(
(
logging.DEBUG,
f"Loaded configuration: {json.dumps(result, indent = 2, sort_keys = True)}",
)
)
return delayed_logs, result
return delayed_logs, {}
def process_exceptions(self, excp):
"""Provide useful feedback if an exception occurs during a run of a plugin."""
# Ensure there's nothing in the cache
+98
View File
@@ -0,0 +1,98 @@
import logging
from typing import Any, List, Optional
from volatility3.framework import constants, interfaces
import re
vollog = logging.getLogger(__name__)
class CLIFilter:
def __init__(self, treegrid, filters: List[str]):
self._filters = self._prepare(treegrid, filters)
def _prepare(self, treegrid: interfaces.renderers.TreeGrid, filters: List[str]):
"""Runs through the filter strings and creates the necessary filter objects"""
output = []
for filter in filters:
exclude = False
regex = False
pattern = None
column_name = None
if filter.startswith("-"):
exclude = True
filter = filter[1:]
elif filter.startswith("+"):
filter = filter[1:]
components = filter.split(",")
if len(components) < 2:
pattern = components[0]
else:
column_name = components[0]
pattern = ",".join(components[1:])
if pattern and pattern.endswith("!"):
regex = True
pattern = pattern[:-1]
column_num = None
if column_name:
for num, column in enumerate(treegrid.columns):
if column_name.lower() in column.name.lower():
column_num = num
break
if pattern:
output.append(ColumnFilter(column_num, pattern, regex, exclude))
vollog.log(constants.LOGLEVEL_VVV, "Filters:\n" + repr(output))
return output
def filter(
self,
row: List[Any],
) -> bool:
"""Filters the row based on each of the column_filters"""
if not self._filters:
return False
found = any([column_filter.found(row) for column_filter in self._filters])
return not found
class ColumnFilter:
def __init__(
self,
column_num: Optional[int],
pattern: str,
regex: bool = False,
exclude: bool = False,
) -> None:
self.column_num = column_num
self.pattern = pattern
self.exclude = exclude
self.regex = regex
def find(self, item) -> bool:
"""Identifies whether an item is found in the appropriate column"""
try:
if self.regex:
return re.search(self.pattern, f"{item}")
return self.pattern in f"{item}"
except IOError:
return False
def found(self, row: List[Any]) -> bool:
"""Determines whether a row should be filtered
If the classes exclude value is false, and the necessary pattern is found, the row is not filtered,
otherwise it is filtered.
"""
if self.column_num is None:
found = any([self.find(x) for x in row])
else:
found = self.find(row[self.column_num])
if self.exclude:
return not found
return found
def __repr__(self) -> str:
"""Returns a display of a column filter"""
return f"ColumnFilter(column={self.column_num},exclude={self.exclude},regex={self.regex},pattern={self.pattern})"
+14 -3
View File
@@ -10,6 +10,7 @@ import string
import sys
from functools import wraps
from typing import Any, Callable, Dict, List, Tuple
from volatility3.cli import text_filter
from volatility3.framework import interfaces, renderers
from volatility3.framework.renderers import format_hints
@@ -134,6 +135,7 @@ class CLIRenderer(interfaces.renderers.Renderer):
name = "unnamed"
structured_output = False
filter: text_filter.CLIFilter = None
class QuickTextRenderer(CLIRenderer):
@@ -172,6 +174,9 @@ class QuickTextRenderer(CLIRenderer):
outfd.write("\n{}\n".format("\t".join(line)))
def visitor(node: interfaces.renderers.TreeNode, accumulator):
if self.filter and self.filter.filter(node.values):
return accumulator
accumulator.write("\n")
# Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
accumulator.write(
@@ -306,6 +311,10 @@ class PrettyTextRenderer(CLIRenderer):
max_column_widths[tree_indent_column] = max(
max_column_widths.get(tree_indent_column, 0), node.path_depth
)
if self.filter and self.filter.filter(node.values):
return accumulator
line = {}
for column_index in range(len(grid.columns)):
column = grid.columns[column_index]
@@ -389,9 +398,11 @@ class JsonRenderer(CLIRenderer):
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
bytes: optional(lambda x: " ".join([f"{b:02x}" for b in x])),
datetime.datetime: lambda x: x.isoformat()
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
else None,
datetime.datetime: lambda x: (
x.isoformat()
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
else None
),
"default": lambda x: x,
}
+38 -16
View File
@@ -22,12 +22,14 @@ from volatility3.framework import (
)
# Make sure we log everything
rootlog = logging.getLogger()
vollog = logging.getLogger()
vollog.setLevel(0)
# Trim the console down by default
console = logging.StreamHandler()
console.setLevel(logging.WARNING)
formatter = logging.Formatter("%(levelname)-8s %(name)-12s: %(message)s")
# Trim the console down by default
console.setFormatter(formatter)
vollog.addHandler(console)
@@ -53,6 +55,9 @@ class VolShell(cli.CommandLine):
framework.require_interface_version(2, 0, 0)
# Load up system defaults
delayed_logs, default_config = self.load_system_defaults("volshell.json")
parser = argparse.ArgumentParser(
prog=self.CLI_NAME,
description="A tool for interactivate forensic analysis of memory images",
@@ -146,6 +151,12 @@ class VolShell(cli.CommandLine):
default=constants.CACHE_PATH,
type=str,
)
parser.add_argument(
"--offline",
help="Do not search online for additional JSON files",
default=False,
action="store_true",
)
# Volshell specific flags
os_specific = parser.add_mutually_exclusive_group(required=False)
@@ -167,26 +178,14 @@ class VolShell(cli.CommandLine):
"-m", "--mac", default=False, action="store_true", help="Run a Mac volshell"
)
parser.set_defaults(**default_config)
# We have to filter out help, otherwise parse_known_args will trigger the help message before having
# processed the plugin choice or had the plugin subparser added.
known_args = [arg for arg in sys.argv if arg != "--help" and arg != "-h"]
partial_args, _ = parser.parse_known_args(known_args)
if partial_args.plugin_dirs:
volatility3.plugins.__path__ = [
os.path.abspath(p) for p in partial_args.plugin_dirs.split(";")
] + constants.PLUGINS_PATH
if partial_args.symbol_dirs:
volatility3.symbols.__path__ = [
os.path.abspath(p) for p in partial_args.symbol_dirs.split(";")
] + constants.SYMBOL_BASEPATHS
if partial_args.cache_path:
constants.CACHE_PATH = partial_args.cache_path
vollog.info(f"Volatility plugins path: {volatility3.plugins.__path__}")
vollog.info(f"Volatility symbols path: {volatility3.symbols.__path__}")
### Start up logging
if partial_args.log:
file_logger = logging.FileHandler(partial_args.log)
file_logger.setLevel(0)
@@ -203,9 +202,32 @@ class VolShell(cli.CommandLine):
else:
console.setLevel(10 - (partial_args.verbosity - 2))
for level, msg in delayed_logs:
vollog.log(level, msg)
### Alter constants if necessary
if partial_args.plugin_dirs:
volatility3.plugins.__path__ = [
os.path.abspath(p) for p in partial_args.plugin_dirs.split(";")
] + constants.PLUGINS_PATH
if partial_args.symbol_dirs:
volatility3.symbols.__path__ = [
os.path.abspath(p) for p in partial_args.symbol_dirs.split(";")
] + constants.SYMBOL_BASEPATHS
if partial_args.cache_path:
constants.CACHE_PATH = partial_args.cache_path
vollog.info(f"Volatility plugins path: {volatility3.plugins.__path__}")
vollog.info(f"Volatility symbols path: {volatility3.symbols.__path__}")
if partial_args.clear_cache:
framework.clear_cache()
if partial_args.offline:
constants.OFFLINE = partial_args.offline
# Do the initialization
ctx = contexts.Context() # Construct a blank context
failures = framework.import_files(
+3 -3
View File
@@ -206,9 +206,9 @@ def _zipwalk(path: str):
if not file.is_dir():
dirlist = zip_results.get(os.path.dirname(file.filename), [])
dirlist.append(os.path.basename(file.filename))
zip_results[
os.path.join(path, os.path.dirname(file.filename))
] = dirlist
zip_results[os.path.join(path, os.path.dirname(file.filename))] = (
dirlist
)
for value in zip_results:
yield value, zip_results[value]
+3 -3
View File
@@ -138,9 +138,9 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
config_path = join("automagic", "MacIntelHelper", new_layer_name)
context.config[join(config_path, "memory_layer")] = layer_name
context.config[join(config_path, "page_map_offset")] = dtb
context.config[
join(config_path, MacSymbolFinder.banner_config_key)
] = str(banner, "latin-1")
context.config[join(config_path, MacSymbolFinder.banner_config_key)] = (
str(banner, "latin-1")
)
new_layer = intel.Intel32e(
context,
+3 -3
View File
@@ -34,9 +34,9 @@ class KernelModule(interfaces.automagic.AutomagicInterface):
return None
# The requirement is unfulfilled and is a ModuleRequirement
context.config[
interfaces.configuration.path_join(new_config_path, "class")
] = "volatility3.framework.contexts.Module"
context.config[interfaces.configuration.path_join(new_config_path, "class")] = (
"volatility3.framework.contexts.Module"
)
for req in requirement.requirements:
if (
@@ -150,12 +150,12 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
clazz = self.symbol_class
# Set the discovered options
path_join = interfaces.configuration.path_join
context.config[
path_join(config_path, requirement.name, "class")
] = clazz
context.config[
path_join(config_path, requirement.name, "isf_url")
] = isf_path
context.config[path_join(config_path, requirement.name, "class")] = (
clazz
)
context.config[path_join(config_path, requirement.name, "isf_url")] = (
isf_path
)
context.config[
path_join(config_path, requirement.name, "symbol_mask")
] = layer.address_mask
+10 -10
View File
@@ -402,19 +402,19 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
if swap_location:
context.config[current_layer_path] = current_layer_name
try:
context.config[
layer_loc_path
] = requirements.URIRequirement.location_from_file(
swap_location
context.config[layer_loc_path] = (
requirements.URIRequirement.location_from_file(
swap_location
)
)
except ValueError:
vollog.warning(
f"Volatility swap_location {swap_location} could not be validated - swap layer disabled"
)
continue
context.config[
layer_class_path
] = "volatility3.framework.layers.physical.FileLayer"
context.config[layer_class_path] = (
"volatility3.framework.layers.physical.FileLayer"
)
# Add the requirement
new_req = requirements.TranslationLayerRequirement(
@@ -424,9 +424,9 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
)
swap_req.add_requirement(new_req)
context.config[
path_join(swap_sub_config, "number_of_elements")
] = counter
context.config[path_join(swap_sub_config, "number_of_elements")] = (
counter
)
context.config[swap_sub_config] = True
swap_req.construct(context, swap_config)
@@ -550,9 +550,9 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
config_path = interfaces.configuration.path_join(config_path, self.name)
if not self.matches_required(self._version, self._component.version):
return {config_path: self}
context.config[
interfaces.configuration.path_join(config_path, self.name)
] = True
context.config[interfaces.configuration.path_join(config_path, self.name)] = (
True
)
return {}
@classmethod
+1 -1
View File
@@ -44,7 +44,7 @@ BANG = "!"
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 6 # Number of changes that only add to the interface
VERSION_MINOR = 7 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
+1 -1
View File
@@ -87,7 +87,7 @@ class ContextInterface(metaclass=ABCMeta):
offset: int,
native_layer_name: str = None,
**arguments,
):
) -> "interfaces.objects.ObjectInterface":
"""Object factory, takes a context, symbol, offset and optional
layer_name.
+1 -1
View File
@@ -60,7 +60,7 @@ class FileHandlerInterface(io.RawIOBase):
@staticmethod
def sanitize_filename(filename: str) -> str:
"""Sanititizes the filename to ensure only a specific whitelist of characters is allowed through"""
allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.- ()[]\{\}!$%^:#~?<>,|"
allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.- ()[]{}!$%^:#~?<>,|"
result = ""
for char in filename:
if char in allowed:
+3 -3
View File
@@ -224,7 +224,7 @@ class AVMLStacker(interfaces.automagic.StackerLayerInterface):
except exceptions.LayerException:
return None
new_name = context.layers.free_layer_name("AVMLLayer")
context.config[
interfaces.configuration.path_join(new_name, "base_layer")
] = layer_name
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
layer_name
)
return AVMLLayer(context, new_name, new_name)
@@ -20,7 +20,6 @@ try:
except ImportError:
HAS_GCSFS = False
from volatility3.framework import exceptions
from volatility3.framework.layers import resources
vollog = logging.getLogger(__file__)
+3 -3
View File
@@ -115,9 +115,9 @@ class Elf64Stacker(interfaces.automagic.StackerLayerInterface):
vollog.log(constants.LOGLEVEL_VVVV, f"Exception: {excp}")
return None
new_name = context.layers.free_layer_name("Elf64Layer")
context.config[
interfaces.configuration.path_join(new_name, "base_layer")
] = layer_name
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
layer_name
)
try:
return Elf64Layer(context, new_name, new_name)
+3 -3
View File
@@ -277,9 +277,9 @@ class Intel(linear.LinearlyMappedLayer):
This allows translation layers to provide maps of contiguous
regions in one layer
"""
stashed_offset = (
stashed_mapped_offset
) = stashed_size = stashed_mapped_size = stashed_map_layer = None
stashed_offset = stashed_mapped_offset = stashed_size = stashed_mapped_size = (
stashed_map_layer
) = None
for offset, size, mapped_offset, mapped_size, map_layer in self._mapping(
offset, length, ignore_errors
):
+3 -3
View File
@@ -104,7 +104,7 @@ class LimeStacker(interfaces.automagic.StackerLayerInterface):
except LimeFormatException:
return None
new_name = context.layers.free_layer_name("LimeLayer")
context.config[
interfaces.configuration.path_join(new_name, "base_layer")
] = layer_name
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
layer_name
)
return LimeLayer(context, new_name, new_name)
+3 -3
View File
@@ -486,9 +486,9 @@ class QemuStacker(interfaces.automagic.StackerLayerInterface):
except exceptions.LayerException:
return None
new_name = context.layers.free_layer_name("QemuSuspendLayer")
context.config[
interfaces.configuration.path_join(new_name, "base_layer")
] = layer_name
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
layer_name
)
layer = QemuSuspendLayer(context, new_name, new_name)
cls.stacker_slow_warning()
return layer
+3 -3
View File
@@ -173,8 +173,8 @@ class XenCoreDumpStacker(elf.Elf64Stacker):
vollog.log(constants.LOGLEVEL_VVVV, f"Exception: {excp}")
return None
new_name = context.layers.free_layer_name("XenCoreDumpLayer")
context.config[
interfaces.configuration.path_join(new_name, "base_layer")
] = layer_name
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
layer_name
)
return XenCoreDumpLayer(context, new_name, new_name)
+2 -4
View File
@@ -768,12 +768,10 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
raise IndexError(f"Member not present in array template: {child}")
@overload
def __getitem__(self, i: int) -> interfaces.objects.Template:
...
def __getitem__(self, i: int) -> interfaces.objects.Template: ...
@overload
def __getitem__(self, s: slice) -> List[interfaces.objects.Template]:
...
def __getitem__(self, s: slice) -> List[interfaces.objects.Template]: ...
def __getitem__(self, i):
"""Returns the i-th item from the array."""
+115 -6
View File
@@ -1,12 +1,121 @@
from volatility3.plugins.linux import envvars
# This file is Copyright 2022 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 volatility3.framework import exceptions, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
class Envars(envvars.Envvars):
def run(self, *args, **kwargs):
vollog.warning(
"The linux.envars plugin has been renamed to linux.envvars and will only be accessible through the new name in a future release"
class Envars(plugins.PluginInterface):
"""Lists processes with their environment variables"""
_required_framework_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
description="Filter on specific process IDs",
element_type=int,
optional=True,
),
]
def _generator(self, tasks):
"""Generates a listing of processes along with environment variables"""
# walk the process list and return the envars
for task in tasks:
pid = task.pid
# get process name as string
name = utility.array_to_string(task.comm)
# try and get task parent
try:
ppid = task.parent.pid
except exceptions.InvalidAddressException:
vollog.debug(
f"Unable to read parent pid for task {pid} {name}, setting ppid to 0."
)
ppid = 0
# kernel threads never have an mm as they do not have userland mappings
try:
mm = task.mm
except exceptions.InvalidAddressException:
# no mm so cannot get envars
vollog.debug(
f"Unable to access mm for task {pid} {name} it is likely a kernel thread, will not extract any envars."
)
mm = None
continue
# if mm exists attempt to get envars
if mm:
# get process layer to read envars from
proc_layer_name = task.add_process_layer()
if proc_layer_name is None:
vollog.debug(
f"Unable to construct process layer for task {pid} {name}, will not extract any envars."
)
continue
proc_layer = self.context.layers[proc_layer_name]
# get the size of the envars with sanity checking
envars_size = task.mm.env_end - task.mm.env_start
if not (0 < envars_size <= 8192):
vollog.debug(
f"Task {pid} {name} appears to have envars of size {envars_size} bytes which fails the sanity checking, will not extract any envars."
)
continue
# attempt to read all envars data
try:
envar_data = proc_layer.read(task.mm.env_start, envars_size)
except exceptions.InvalidAddressException:
vollog.debug(
f"Unable to read full envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)} for {envars_size} bytes, will not extract any envars."
)
continue
# parse envar data, envars are null terminated, keys and values are separated by '='
envar_data = envar_data.rstrip(b"\x00")
for envar_pair in envar_data.split(b"\x00"):
try:
key, value = envar_pair.decode().split("=", 1)
except ValueError:
vollog.debug(
f"Unable to extract envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)}, they don't appear to be '=' separated"
)
continue
yield (0, (pid, ppid, name, key, value))
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
return renderers.TreeGrid(
[("PID", int), ("PPID", int), ("COMM", str), ("KEY", str), ("VALUE", str)],
self._generator(
pslist.PsList.list_tasks(
self.context, self.config["kernel"], filter_func=filter_func
)
),
)
return super().run(*args, **kwargs)
@@ -1,121 +0,0 @@
# This file is Copyright 2022 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 volatility3.framework import exceptions, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
class Envvars(plugins.PluginInterface):
"""Lists processes with their environment variables"""
_required_framework_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
description="Filter on specific process IDs",
element_type=int,
optional=True,
),
]
def _generator(self, tasks):
"""Generates a listing of processes along with environment variables"""
# walk the process list and return the envars
for task in tasks:
pid = task.pid
# get process name as string
name = utility.array_to_string(task.comm)
# try and get task parent
try:
ppid = task.parent.pid
except exceptions.InvalidAddressException:
vollog.debug(
f"Unable to read parent pid for task {pid} {name}, setting ppid to 0."
)
ppid = 0
# kernel threads never have an mm as they do not have userland mappings
try:
mm = task.mm
except exceptions.InvalidAddressException:
# no mm so cannot get envars
vollog.debug(
f"Unable to access mm for task {pid} {name} it is likely a kernel thread, will not extract any envars."
)
mm = None
continue
# if mm exists attempt to get envars
if mm:
# get process layer to read envars from
proc_layer_name = task.add_process_layer()
if proc_layer_name is None:
vollog.debug(
f"Unable to construct process layer for task {pid} {name}, will not extract any envars."
)
continue
proc_layer = self.context.layers[proc_layer_name]
# get the size of the envars with sanity checking
envars_size = task.mm.env_end - task.mm.env_start
if not (0 < envars_size <= 8192):
vollog.debug(
f"Task {pid} {name} appears to have envars of size {envars_size} bytes which fails the sanity checking, will not extract any envars."
)
continue
# attempt to read all envars data
try:
envar_data = proc_layer.read(task.mm.env_start, envars_size)
except exceptions.InvalidAddressException:
vollog.debug(
f"Unable to read full envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)} for {envars_size} bytes, will not extract any envars."
)
continue
# parse envar data, envars are null terminated, keys and values are separated by '='
envar_data = envar_data.rstrip(b"\x00")
for envar_pair in envar_data.split(b"\x00"):
try:
key, value = envar_pair.decode().split("=", 1)
except ValueError:
vollog.debug(
f"Unable to extract envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)}, they don't appear to be '=' separated"
)
continue
yield (0, (pid, ppid, name, key, value))
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
return renderers.TreeGrid(
[("PID", int), ("PPID", int), ("COMM", str), ("KEY", str), ("VALUE", str)],
self._generator(
pslist.PsList.list_tasks(
self.context, self.config["kernel"], filter_func=filter_func
)
),
)
+12 -2
View File
@@ -10,6 +10,7 @@ from typing import Generator, Iterator, List, Tuple
from volatility3.framework import (
class_subclasses,
constants,
exceptions,
interfaces,
renderers,
)
@@ -197,7 +198,9 @@ class ABCKmsg(ABC):
class Kmsg_pre_3_5(ABCKmsg):
"""The kernel ring buffer (log_buf) is a char array that sequentially stores
log lines, each separated by newline (LF) characters. i.e:
<6>[ 9565.250411] line1!\n<6>[ 9565.250412] line2\n...
<6>[ 9565.250411] line1!\\n<6>[ 9565.250412] line2\\n...
"""
@classmethod
@@ -495,7 +498,7 @@ class Kmsg_5_10_to_(ABCKmsg):
class Kmsg(interfaces.plugins.PluginInterface):
"""Kernel log buffer reader"""
_required_framework_version = (2, 0, 0)
_required_framework_version = (2, 6, 0)
_version = (1, 0, 2)
@@ -514,6 +517,13 @@ class Kmsg(interfaces.plugins.PluginInterface):
yield (0, values)
def run(self):
if not self.context.symbol_space.verify_table_versions(
"dwarf2json", lambda version, _: (not version) or version > (0, 4, 1)
):
raise exceptions.SymbolSpaceError(
"Invalid symbol table, please ensure the ISF table produced by dwarf2json was produced using a version > 0.4.1"
)
return renderers.TreeGrid(
[
("facility", str),
@@ -17,7 +17,7 @@ class PsList(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 0, 0)
_version = (2, 2, 0)
_version = (2, 2, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -83,11 +83,11 @@ class PsList(interfaces.plugins.PluginInterface):
cls, task: interfaces.objects.ObjectInterface, decorate_comm: bool = False
) -> Tuple[int, int, int, str]:
"""Extract the fields needed for the final output
Args:
task: A task object from where to get the fields.
decorate_comm: If True, it decorates the comm string of
- User threads: in curly brackets,
- Kernel threads: in square brackets
decorate_comm: If True, it decorates the comm string of user threads in curly brackets,
and of Kernel threads in square brackets.
Defaults to False.
Returns:
A tuple with the fields to show in the plugin output.
@@ -128,7 +128,7 @@ class PsList(interfaces.plugins.PluginInterface):
else:
# Find the vma that belongs to the main ELF of the process
file_output = "Error outputting file"
for v in task.mm.get_mmap_iter():
for v in task.mm.get_vma_iter():
if v.vm_start == task.mm.start_code:
file_handle = elfs.Elfs.elf_dump(
self.context,
@@ -28,7 +28,7 @@ class PsScan(interfaces.plugins.PluginInterface):
"""Scans for processes present in a particular linux image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -139,7 +139,7 @@ class PsScan(interfaces.plugins.PluginInterface):
kernel_layer_name, f"Layer {kernel_layer_name} has no dependencies"
)
memory_layer_name = kernel_layer.dependencies[0]
memory_layer = context.layers[kernel_layer.dependencies[0]]
memory_layer = context.layers[memory_layer_name]
# scan the memory_layer for these needles
for address, _ in memory_layer.scan(
@@ -83,7 +83,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
sock: Kernel generic `sock` object
Returns a tuple with:
sock: The respective kernel's \*_sock object for that socket family
sock: The respective kernel's \\*_sock object for that socket family
sock_stat: A tuple with the source and destination (address and port) along with its state string
socket_filter: A dictionary with information about the socket filter
"""
@@ -501,7 +501,7 @@ class Sockstat(plugins.PluginInterface):
family: Socket family string (AF_UNIX, AF_INET, etc)
sock_type: Socket type string (STREAM, DGRAM, etc)
protocol: Protocol string (UDP, TCP, etc)
sock_fields: A tuple with the \*_sock object, the sock stats and the extended info dictionary
sock_fields: A tuple with the \\*_sock object, the sock stats and the extended info dictionary
"""
vmlinux = context.modules[symbol_table]
+1 -3
View File
@@ -49,9 +49,7 @@ class PsList(interfaces.plugins.PluginInterface):
]
@classmethod
def get_list_tasks(
cls, method: str
) -> Callable[
def get_list_tasks(cls, method: str) -> Callable[
[interfaces.context.ContextInterface, str, Callable[[int], bool]],
Iterable[interfaces.objects.ObjectInterface],
]:
@@ -46,9 +46,9 @@ class Crashinfo(interfaces.plugins.PluginInterface):
bitmap_size = format_hints.Hex(summary_header.BitmapSize)
bitmap_pages = format_hints.Hex(summary_header.Pages)
else:
bitmap_header_size = (
bitmap_size
) = bitmap_pages = renderers.NotApplicableValue()
bitmap_header_size = bitmap_size = bitmap_pages = (
renderers.NotApplicableValue()
)
yield (
0,
@@ -13,7 +13,7 @@ from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import pe
from volatility3.plugins import timeliner
from volatility3.plugins.windows import info, pslist
from volatility3.plugins.windows import info, pslist, psscan
vollog = logging.getLogger(__name__)
@@ -36,6 +36,9 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="psscan", component=psscan.PsScan, version=(1, 1, 0)
),
requirements.VersionRequirement(
name="info", component=info.Info, version=(1, 0, 0)
),
@@ -45,6 +48,11 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
description="Process IDs to include (all other processes are excluded)",
optional=True,
),
requirements.IntRequirement(
name="offset",
description="Process offset in the physical address space",
optional=True,
),
requirements.BooleanRequirement(
name="dump",
description="Extract listed DLLs",
@@ -221,6 +229,25 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
kernel = self.context.modules[self.config["kernel"]]
if self.config["offset"]:
procs = psscan.PsScan.scan_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func=psscan.PsScan.create_offset_filter(
self.context,
kernel.layer_name,
self.config["offset"],
),
)
else:
procs = pslist.PsList.list_processes(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
filter_func=filter_func,
)
return renderers.TreeGrid(
[
("PID", int),
@@ -232,12 +259,5 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
("LoadTime", datetime.datetime),
("File output", str),
],
self._generator(
pslist.PsList.list_processes(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
filter_func=filter_func,
)
),
self._generator(procs=procs),
)
@@ -4,11 +4,12 @@
import logging
import ntpath
import re
from typing import List, Tuple, Type, Optional, Generator
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.renderers import format_hints, UnreadableValue
from volatility3.plugins.windows import handles
from volatility3.plugins.windows import pslist
@@ -53,6 +54,17 @@ class DumpFiles(interfaces.plugins.PluginInterface):
description="Dump a single _FILE_OBJECT at this physical address",
optional=True,
),
requirements.StringRequirement(
name="filter",
description="Dump files matching regular expression FILTER",
optional=True,
),
requirements.BooleanRequirement(
name="ignore-case",
description="Ignore case in filter match",
default=False,
optional=True,
),
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(2, 0, 0)
),
@@ -208,6 +220,10 @@ class DumpFiles(interfaces.plugins.PluginInterface):
def _generator(self, procs: List, offsets: List):
kernel = self.context.modules[self.config["kernel"]]
file_re = None
if self.config["filter"]:
flags = re.I if self.config["ignore-case"] else 0
file_re = re.compile(self.config["filter"], flags)
if procs:
# The handles plugin doesn't expose any staticmethod/classmethod, and it also requires stashing
@@ -243,6 +259,14 @@ class DumpFiles(interfaces.plugins.PluginInterface):
obj_type = entry.get_object_type(type_map, cookie)
if obj_type == "File":
file_obj = entry.Body.cast("_FILE_OBJECT")
if file_re:
name = file_obj.file_name_with_device()
if isinstance(name, UnreadableValue):
continue
if not file_re.search(name):
continue
for result in self.process_file_object(
self.context, kernel.layer_name, self.open, file_obj
):
@@ -272,6 +296,13 @@ class DumpFiles(interfaces.plugins.PluginInterface):
if not file_obj.is_valid():
continue
if file_re:
name = file_obj.file_name_with_device()
if isinstance(name, UnreadableValue):
continue
if not file_re.search(name):
continue
for result in self.process_file_object(
self.context, kernel.layer_name, self.open, file_obj
):
@@ -315,6 +346,11 @@ class DumpFiles(interfaces.plugins.PluginInterface):
procs = list()
kernel = self.context.modules[self.config["kernel"]]
if self.config["filter"] and (
self.config["virtaddr"] or self.config["physaddr"]
):
raise ValueError("Cannot use filter flag with an address flag")
if self.config.get("virtaddr", None) is not None:
offsets.append((self.config["virtaddr"], True))
elif self.config.get("physaddr", None) is not None:
@@ -9,7 +9,7 @@ from volatility3.framework import constants, exceptions, renderers, interfaces,
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.plugins.windows import pslist
from volatility3.plugins.windows import pslist, psscan
vollog = logging.getLogger(__name__)
@@ -43,14 +43,22 @@ class Handles(interfaces.plugins.PluginInterface):
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="psscan", component=psscan.PsScan, version=(1, 1, 0)
),
requirements.ListRequirement(
name="pid",
element_type=int,
description="Process IDs to include (all other processes are excluded)",
optional=True,
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
requirements.IntRequirement(
name="offset",
description="Process offset in the physical address space",
optional=True,
),
]
@@ -416,6 +424,25 @@ class Handles(interfaces.plugins.PluginInterface):
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
kernel = self.context.modules[self.config["kernel"]]
if self.config["offset"]:
procs = psscan.PsScan.scan_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func=psscan.PsScan.create_offset_filter(
self.context,
kernel.layer_name,
self.config["offset"],
),
)
else:
procs = pslist.PsList.list_processes(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
filter_func=filter_func,
)
return renderers.TreeGrid(
[
("PID", int),
@@ -426,12 +453,5 @@ class Handles(interfaces.plugins.PluginInterface):
("GrantedAccess", format_hints.Hex),
("Name", str),
],
self._generator(
pslist.PsList.list_processes(
self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func=filter_func,
)
),
self._generator(procs=procs),
)
@@ -0,0 +1,150 @@
# 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, io, pefile
from volatility3.framework.symbols import intermed
from volatility3.framework import renderers, interfaces, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.plugins.windows import pslist
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols.windows.extensions import pe
vollog = logging.getLogger(__name__)
class IAT(interfaces.plugins.PluginInterface):
"""Extract Import Address Table to list API (functions) used by a program contained in external libraries"""
_required_framework_version = (2, 4, 0)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
element_type=int,
description="Process ID to include (all other processes are excluded)",
optional=True,
),
]
def _generator(self, procs):
kernel = self.context.modules[self.config["kernel"]]
for proc in procs:
try:
proc_id = proc.UniqueProcessId
proc_layer_name = proc.add_process_layer()
peb = self.context.object(
kernel.symbol_table_name + constants.BANG + "_PEB",
layer_name=proc_layer_name,
offset=proc.Peb,
)
if proc_layer_name is None:
raise TypeError("add_process_layer failed")
pe_table_name = intermed.IntermediateSymbolTable.create(
self.context,
self.config_path,
"windows",
"pe",
class_types=pe.class_types,
)
pe_data = io.BytesIO()
dos_header = self.context.object(
pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset=peb.ImageBaseAddress,
layer_name=proc_layer_name,
)
for offset, data in dos_header.reconstruct():
pe_data.seek(offset)
pe_data.write(data)
pe_obj = pefile.PE(data=pe_data.getvalue(), fast_load=True)
pe_obj.parse_data_directories(
[pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_IMPORT"]]
)
if hasattr(pe_obj, "DIRECTORY_ENTRY_IMPORT"):
for entry in pe_obj.DIRECTORY_ENTRY_IMPORT:
dll_entry = entry.dll
if dll_entry:
dll_entry = dll_entry.decode()
else:
dll_entry = renderers.NotAvailableValue
bound = True
# Initially set to 0 if not bound
time_date_stamp = entry.struct.TimeDateStamp
if not time_date_stamp:
bound = False
# Iterate over imported functions
for imp in entry.imports:
import_name = imp.name
if import_name:
import_name = imp.name.decode()
else:
import_name = renderers.NotAvailableValue()
function_address = (
pe_obj.OPTIONAL_HEADER.ImageBase + imp.address
)
if not function_address:
function_address = renderers.NotAvailableValue
yield (
0,
(
proc_id,
proc.ImageFileName.cast(
"string",
max_length=proc.ImageFileName.vol.count,
errors="replace",
),
dll_entry,
bound,
import_name,
format_hints.Hex(function_address),
),
)
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
)
continue
def run(self):
kernel = self.context.modules[self.config["kernel"]]
return renderers.TreeGrid(
[
("PID", int),
("Name", str),
("Library", str),
("Bound", bool),
("Function", str),
("Address", format_hints.Hex),
],
self._generator(
pslist.PsList.list_processes(
context=self.context,
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
filter_func=pslist.PsList.create_pid_filter(
self.config.get("pid", None)
),
)
),
)
@@ -141,16 +141,30 @@ class Malfind(interfaces.plugins.PluginInterface):
# determine if we're on a 32 or 64 bit kernel
kernel = self.context.modules[self.config["kernel"]]
# set refined criteria to know when to add to "Notes" column
refined_criteria = {
b"MZ": "MZ header",
b"\x55\x8B": "PE header",
b"\x55\x48": "Function prologue",
b"\x55\x89": "Function prologue",
}
is_32bit_arch = not symbols.symbol_table_is_64bit(
self.context, kernel.symbol_table_name
)
for proc in procs:
# by default, "Notes" column will be set to N/A
notes = renderers.NotApplicableValue()
process_name = utility.array_to_string(proc.ImageFileName)
for vad, data in self.list_injections(
self.context, kernel.layer_name, kernel.symbol_table_name, proc
):
# Check for unique headers and update "Notes" column if criteria is met
if data[0:2] in refined_criteria:
notes = refined_criteria[data[0:2]]
# if we're on a 64 bit kernel, we may still need 32 bit disasm due to wow64
if is_32bit_arch or proc.get_is_wow64():
architecture = "intel"
@@ -196,6 +210,7 @@ class Malfind(interfaces.plugins.PluginInterface):
vad.get_commit_charge(),
vad.get_private_memory(),
file_output,
notes,
format_hints.HexBytes(data),
disasm,
),
@@ -216,6 +231,7 @@ class Malfind(interfaces.plugins.PluginInterface):
("CommitCharge", int),
("PrivateMemory", int),
("File output", str),
("Notes", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
],
@@ -38,7 +38,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# Yara Rule to scan for MFT Header Signatures
rules = yarascan.YaraScan.process_yara_options(
{"yara_rules": "/FILE0|FILE\*|BAAD/"}
{"yara_rules": "/FILE0|FILE\\*|BAAD/"}
)
# Read in the Symbol File
@@ -53,7 +53,6 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# get each of the individual Field Sets
mft_object = symbol_table + constants.BANG + "MFT_ENTRY"
attribute_object = symbol_table + constants.BANG + "ATTRIBUTE"
header_object = symbol_table + constants.BANG + "ATTR_HEADER"
si_object = symbol_table + constants.BANG + "STANDARD_INFORMATION_ENTRY"
fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY"
@@ -176,7 +175,6 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
class ADS(interfaces.plugins.PluginInterface):
"""Scans for Alternate Data Stream"""
_required_framework_version = (2, 0, 0)
@@ -199,7 +197,7 @@ class ADS(interfaces.plugins.PluginInterface):
# Yara Rule to scan for MFT Header Signatures
rules = yarascan.YaraScan.process_yara_options(
{"yara_rules": "/FILE0|FILE\*|BAAD/"}
{"yara_rules": "/FILE0|FILE\\*|BAAD/"}
)
# Read in the Symbol File
@@ -487,10 +487,12 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
if not isinstance(row_data[9], datetime.datetime):
continue
row_data = [
"N/A"
if isinstance(i, renderers.UnreadableValue)
or isinstance(i, renderers.UnparsableValue)
else i
(
"N/A"
if isinstance(i, renderers.UnreadableValue)
or isinstance(i, renderers.UnparsableValue)
else i
)
for i in row_data
]
description = (
@@ -59,6 +59,75 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
]
@classmethod
def physical_offset_from_virtual(cls, context, layer_name, proc):
"""Calculate the physical offset from the virtual offset of a process.
Args:
context: The context containing layers and modules information.
layer_name: The name of the layer containing the process memory.
proc: The process object for which to calculate the physical offset.
Returns:
int: The physical offset of the process.
Raises:
TypeError: If the primary layer is not an Intel layer.
"""
memory = context.layers[layer_name]
if not isinstance(memory, layers.intel.Intel):
raise TypeError("Primary layer is not an intel layer")
(_, _, ph_offset, _, _) = list(
memory.mapping(offset=proc.vol.offset, length=0)
)[0]
return ph_offset
@classmethod
def create_offset_filter(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
offset: int = None,
physical: bool = True,
exclude: bool = False,
) -> Callable[[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on the physical offset of the process.
Args:
offset: A number that is the physical offset to be filtered out
exclude: Accept only tasks that are not the offset argument
Returns:
Filter function to be passed to the list of processes.
"""
filter_func = lambda _: False
if offset:
if physical:
if exclude:
filter_func = (
lambda proc: cls.physical_offset_from_virtual(
context, layer_name, proc
)
== offset
)
else:
filter_func = (
lambda proc: cls.physical_offset_from_virtual(
context, layer_name, proc
)
!= offset
)
else:
if exclude:
filter_func = lambda proc: proc.vol.offset == offset
else:
filter_func = lambda proc: proc.vol.offset != offset
return filter_func
@classmethod
def scan_processes(
cls,
@@ -5,7 +5,7 @@ import datetime
import logging
from typing import Callable, Dict, Set, Tuple
from volatility3.framework import objects, interfaces, renderers
from volatility3.framework import objects, interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.plugins.windows import pslist
@@ -132,6 +132,25 @@ class PsTree(interfaces.plugins.PluginInterface):
proc.get_exit_time(),
)
try:
audit = proc.SeAuditProcessCreationInfo.ImageFileName.Name
# If 'audit' is set to the empty string, display NotAvailableValue
row += (audit.get_string() or renderers.NotAvailableValue(),)
except exceptions.InvalidAddressException:
row += (renderers.NotAvailableValue(),)
try:
process_params = proc.get_peb().ProcessParameters
row += (
process_params.CommandLine.get_string(),
process_params.ImagePathName.get_string(),
)
except exceptions.InvalidAddressException:
row += (
renderers.NotAvailableValue(),
renderers.NotAvailableValue(),
)
yield (self._levels[pid] - 1, row)
for child_pid in self._children.get(pid, []):
yield from yield_processes(
@@ -161,6 +180,9 @@ class PsTree(interfaces.plugins.PluginInterface):
("Wow64", bool),
("CreateTime", datetime.datetime),
("ExitTime", datetime.datetime),
("Audit", str),
("Cmd", str),
("Path", str),
],
self._generator(
filter_func=pslist.PsList.create_pid_filter(
@@ -193,9 +193,9 @@ class PrintKey(interfaces.plugins.PluginInterface):
vollog.debug(
"Couldn't read registry value type, so data is unreadable"
)
value_data: Union[
interfaces.renderers.BaseAbsentValue, bytes
] = renderers.UnreadableValue()
value_data: Union[interfaces.renderers.BaseAbsentValue, bytes] = (
renderers.UnreadableValue()
)
else:
try:
value_data = node.decode_data()
+179 -68
View File
@@ -4,25 +4,42 @@
import logging
import os
from typing import List
from typing import Dict, List, NamedTuple, Optional, Tuple, Union, cast
from volatility3.framework import interfaces, renderers, constants, symbols, exceptions
from volatility3.framework import (
constants,
exceptions,
interfaces,
objects,
renderers,
symbols,
)
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import scanners
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import versions
from volatility3.framework.symbols.windows.extensions import services
from volatility3.plugins.windows import poolscanner, vadyarascan, pslist
from volatility3.plugins.windows import poolscanner, pslist, vadyarascan
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
ServiceBinaryInfo = NamedTuple(
"ServiceBinaryInfo",
[
("dll", Union[str, interfaces.renderers.BaseAbsentValue]),
("binary", Union[str, interfaces.renderers.BaseAbsentValue]),
],
)
class SvcScan(interfaces.plugins.PluginInterface):
"""Scans for windows services."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -42,10 +59,16 @@ class SvcScan(interfaces.plugins.PluginInterface):
requirements.PluginRequirement(
name="vadyarascan", plugin=vadyarascan.VadYaraScan, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="hivelist", plugin=hivelist.HiveList, version=(1, 0, 0)
),
]
@staticmethod
def get_record_tuple(service_record: interfaces.objects.ObjectInterface):
def get_record_tuple(
service_record: interfaces.objects.ObjectInterface,
binary_info: ServiceBinaryInfo,
):
return (
format_hints.Hex(service_record.vol.offset),
service_record.Order,
@@ -56,8 +79,32 @@ class SvcScan(interfaces.plugins.PluginInterface):
service_record.get_name(),
service_record.get_display(),
service_record.get_binary(),
binary_info.binary,
binary_info.dll,
)
# These checks must be completed from newest -> oldest OS version.
_win_version_file_map: List[Tuple[versions.OsDistinguisher, bool, str]] = [
(versions.is_win10_25398_or_later, True, "services-win10-25398-x64"),
(versions.is_win10_19041_or_later, True, "services-win10-19041-x64"),
(versions.is_win10_19041_or_later, False, "services-win10-19041-x86"),
(versions.is_win10_18362_or_later, True, "services-win10-18362-x64"),
(versions.is_win10_18362_or_later, False, "services-win10-18362-x86"),
(versions.is_win10_17763_or_later, False, "services-win10-17763-x86"),
(versions.is_win10_16299_or_later, True, "services-win10-16299-x64"),
(versions.is_win10_16299_or_later, False, "services-win10-16299-x86"),
(versions.is_win10_15063, True, "services-win10-15063-x64"),
(versions.is_win10_15063, False, "services-win10-15063-x86"),
(versions.is_win10_up_to_15063, True, "services-win8-x64"),
(versions.is_win10_up_to_15063, False, "services-win8-x86"),
(versions.is_windows_8_or_later, True, "services-win8-x64"),
(versions.is_windows_8_or_later, True, "services-win8-x86"),
(versions.is_vista_or_later, True, "services-vista-x64"),
(versions.is_vista_or_later, False, "services-vista-x86"),
(versions.is_windows_xp, False, "services-xp-x86"),
(versions.is_xp_or_2003, True, "services-xp-2003-x64"),
]
@staticmethod
def create_service_table(
context: interfaces.context.ContextInterface,
@@ -78,67 +125,14 @@ class SvcScan(interfaces.plugins.PluginInterface):
native_types = context.symbol_space[symbol_table].natives
is_64bit = symbols.symbol_table_is_64bit(context, symbol_table)
if (
versions.is_windows_xp(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-xp-x86"
elif (
versions.is_xp_or_2003(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-xp-2003-x64"
elif (
versions.is_win10_16299_or_later(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-win10-16299-x64"
elif (
versions.is_win10_16299_or_later(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-win10-16299-x86"
elif (
versions.is_win10_up_to_15063(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-win8-x64"
elif (
versions.is_win10_up_to_15063(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-win8-x86"
elif (
versions.is_win10_15063(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-win10-15063-x64"
elif (
versions.is_win10_15063(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-win10-15063-x86"
elif (
versions.is_windows_8_or_later(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-win8-x64"
elif (
versions.is_windows_8_or_later(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-win8-x86"
elif (
versions.is_vista_or_later(context=context, symbol_table=symbol_table)
and is_64bit
):
symbol_filename = "services-vista-x64"
elif (
versions.is_vista_or_later(context=context, symbol_table=symbol_table)
and not is_64bit
):
symbol_filename = "services-vista-x86"
else:
try:
symbol_filename = next(
filename
for version_check, for_64bit, filename in SvcScan._win_version_file_map
if is_64bit == for_64bit
and version_check(context=context, symbol_table=symbol_table)
)
except StopIteration:
raise NotImplementedError("This version of Windows is not supported!")
return intermed.IntermediateSymbolTable.create(
@@ -150,6 +144,94 @@ class SvcScan(interfaces.plugins.PluginInterface):
native_types=native_types,
)
def _get_service_key(self, kernel) -> Optional[objects.StructType]:
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=interfaces.configuration.path_join(
self.config_path, "hivelist"
),
layer_name=kernel.layer_name,
symbol_table=kernel.symbol_table_name,
filter_string="machine\\system",
):
# Get ControlSet\Services.
try:
return cast(
objects.StructType, hive.get_key(r"CurrentControlSet\Services")
)
except (KeyError, exceptions.InvalidAddressException):
try:
return cast(
objects.StructType, hive.get_key(r"ControlSet001\Services")
)
except (KeyError, exceptions.InvalidAddressException):
vollog.log(
constants.LOGLEVEL_VVVV,
"Could not retrieve any control set from SYSTEM hive",
)
return None
@staticmethod
def _get_service_dll(
service_key,
) -> Union[str, interfaces.renderers.BaseAbsentValue]:
try:
param_key = next(
key
for key in service_key.get_subkeys()
if key.get_name() == "Parameters"
)
return (
next(
val
for val in param_key.get_values()
if val.get_name() == "ServiceDll"
)
.decode_data()
.decode("utf-16")
.rstrip("\x00")
)
except UnicodeDecodeError:
return renderers.UnparsableValue()
except StopIteration:
return renderers.UnreadableValue()
@staticmethod
def _get_service_binary(
service_key,
) -> Union[str, interfaces.renderers.BaseAbsentValue]:
try:
return (
next(
val
for val in service_key.get_values()
if val.get_name() == "ImagePath"
)
.decode_data()
.decode("utf-16")
.rstrip("\x00")
)
except UnicodeDecodeError:
return renderers.UnparsableValue()
except StopIteration:
return renderers.UnreadableValue()
@staticmethod
def _get_service_binary_map(
services_key: interfaces.objects.ObjectInterface,
) -> Dict[str, ServiceBinaryInfo]:
services = services_key.get_subkeys()
return {
service_key.get_name(): ServiceBinaryInfo(
SvcScan._get_service_dll(service_key),
SvcScan._get_service_binary(service_key),
)
for service_key in services
}
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
@@ -157,6 +239,15 @@ class SvcScan(interfaces.plugins.PluginInterface):
self.context, kernel.symbol_table_name, self.config_path
)
# Building the dictionary ahead of time is much better for performance
# vs looking up each service's DLL individually.
services_key = self._get_service_key(kernel)
service_binary_dll_map = (
self._get_service_binary_map(services_key)
if services_key is not None
else {}
)
relative_tag_offset = self.context.symbol_space.get_type(
service_table_name + constants.BANG + "_SERVICE_RECORD"
).relative_child_offset("Tag")
@@ -209,7 +300,16 @@ class SvcScan(interfaces.plugins.PluginInterface):
if not service_record.is_valid():
continue
yield (0, self.get_record_tuple(service_record))
service_info = service_binary_dll_map.get(
service_record.get_name(),
ServiceBinaryInfo(
renderers.UnreadableValue(), renderers.UnreadableValue()
),
)
yield (
0,
self.get_record_tuple(service_record, service_info),
)
else:
service_header = self.context.object(
service_table_name + constants.BANG + "_SERVICE_HEADER",
@@ -227,7 +327,16 @@ class SvcScan(interfaces.plugins.PluginInterface):
if service_record in seen:
break
seen.append(service_record)
yield (0, self.get_record_tuple(service_record))
service_info = service_binary_dll_map.get(
service_record.get_name(),
ServiceBinaryInfo(
renderers.UnreadableValue(), renderers.UnreadableValue()
),
)
yield (
0,
self.get_record_tuple(service_record, service_info),
)
def run(self):
return renderers.TreeGrid(
@@ -241,6 +350,8 @@ class SvcScan(interfaces.plugins.PluginInterface):
("Name", str),
("Display", str),
("Binary", str),
("Binary (Registry)", str),
("Dll", str),
],
self._generator(),
)
@@ -0,0 +1,144 @@
# 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
#
from typing import Iterable, Generator, List, Tuple
from volatility3.framework import constants, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces.configuration import RequirementInterface
from volatility3.framework.interfaces.objects import ObjectInterface
from volatility3.framework.objects import Bytes, DataFormatInfo, Integer, StructType
from volatility3.framework.objects.templates import ObjectTemplate
from volatility3.framework.objects.utility import array_to_string
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import pe
from volatility3.plugins.windows import modules
class Passphrase(interfaces.plugins.PluginInterface):
"""TrueCrypt Cached Passphrase Finder"""
_version = (0, 1, 0)
_required_framework_version = (2, 5, 2)
@classmethod
def get_requirements(cls) -> List[RequirementInterface]:
return [
requirements.ModuleRequirement(
"kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="modules", component=modules.Modules, version=(1, 1, 0)
),
requirements.IntRequirement(
name="min-length",
description="Minimum length of passphrases to identify",
default=5,
optional=True,
),
]
def scan_module(
self, module_base: int, layer_name: str
) -> Generator[Tuple[int, str], None, None]:
"""Scans the TrueCrypt kernel module for cached passphrases.
Args:
module_base: the module's DLL base
layer_name: the name of the layer in which the module resides
Generates:
A tuple of the offset at which a password is found, and the password
"""
pe_table_name = intermed.IntermediateSymbolTable.create(
self.context, self.config_path, "windows", "pe", class_types=pe.class_types
)
dos_header: pe.IMAGE_DOS_HEADER = self.context.object(
pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
layer_name,
module_base,
)
data_section: StructType = next(
sec
for sec in dos_header.get_nt_header().get_sections()
if array_to_string(sec.Name) == ".data"
)
base: int = data_section.VirtualAddress + module_base
size: int = data_section.Misc.VirtualSize
# Looking at `Length` in TrueCrypt/Common/Password.h::Password struct
DWORD_SIZE_BYTES: int = 4
format = DataFormatInfo(
length=DWORD_SIZE_BYTES, byteorder="little", signed=True
)
int32 = ObjectTemplate(
Integer, pe_table_name + constants.BANG + "int", data_format=format
)
count, not_aligned = divmod(size, DWORD_SIZE_BYTES)
if not_aligned:
raise ValueError("PE data section not DWORD-aligned!")
lengths = self.context.object(
pe_table_name + constants.BANG + "array",
layer_name,
base,
count=count,
subtype=int32,
)
min_length = self.config.get("min-length")
for length in lengths:
# TrueCrypt maximum password length is 64
# (see TrueCrypt/Common/Password.h)
if not min_length <= length <= 64:
continue
offset = length.vol["offset"] + DWORD_SIZE_BYTES
passphrase: Bytes = self.context.object(
pe_table_name + constants.BANG + "bytes",
layer_name,
offset,
length=length,
)
# TrueCrypt/Common/Password.c permits chars in the range
# [0x20, 0x7F).
if not all(0x20 <= c < 0x7F for c in passphrase):
continue
# TrueCrypt/Common/Password.h::Password struct is padded with
# 3 zero bytes to keep 64-byte alignment.
buf: Bytes = self.context.object(
pe_table_name + constants.BANG + "bytes",
layer_name,
offset + length + 1, # +1 for '\0'-terminated password string
length=3,
)
if any(buf):
continue
# Password found.
yield offset, passphrase.decode(encoding="ascii")
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
mods: Iterable[ObjectInterface] = modules.Modules.list_modules(
self.context, kernel.layer_name, kernel.symbol_table_name
)
truecrypt_module_base = next(
mod.DllBase
for mod in mods
if mod.BaseDllName.get_string().lower() == "truecrypt.sys"
)
for offset, password in self.scan_module(
truecrypt_module_base, kernel.layer_name
):
yield (0, (format_hints.Hex(offset), len(password), password))
def run(self) -> renderers.TreeGrid:
return renderers.TreeGrid(
[
("Offset", format_hints.Hex),
("Length", int),
("Password", str),
],
self._generator(),
)
@@ -28,9 +28,9 @@ def wintime_to_datetime(
def unixtime_to_datetime(
unixtime: int,
) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]:
ret: Union[
interfaces.renderers.BaseAbsentValue, datetime.datetime
] = renderers.UnparsableValue()
ret: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] = (
renderers.UnparsableValue()
)
if unixtime > 0:
with contextlib.suppress(ValueError):
+61 -14
View File
@@ -4,9 +4,20 @@
import collections
import collections.abc
import datetime
import enum
import logging
from typing import Any, Dict, Iterable, Iterator, TypeVar, List
from typing import (
Any,
Callable,
Dict,
Iterable,
Iterator,
Optional,
Tuple,
TypeVar,
List,
)
from volatility3.framework import constants, exceptions, interfaces, objects
@@ -35,9 +46,9 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
def __init__(self) -> None:
super().__init__()
self._dict: Dict[
str, interfaces.symbols.BaseSymbolTableInterface
] = collections.OrderedDict()
self._dict: Dict[str, interfaces.symbols.BaseSymbolTableInterface] = (
collections.OrderedDict()
)
# Permanently cache all resolved symbols
self._resolved: Dict[str, interfaces.objects.Template] = {}
self._resolved_symbols: Dict[str, interfaces.objects.Template] = {}
@@ -73,9 +84,9 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
self, offset: int, size: int = 0, table_name: str = None
) -> Iterable[str]:
"""Returns all symbols that exist at a specific relative address."""
table_list: Iterable[
interfaces.symbols.BaseSymbolTableInterface
] = self._dict.values()
table_list: Iterable[interfaces.symbols.BaseSymbolTableInterface] = (
self._dict.values()
)
if table_name is not None:
if table_name in self._dict:
table_list = [self._dict[table_name]]
@@ -113,6 +124,42 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
self._resolved = {}
del self._dict[key]
def verify_table_versions(
self,
producer: str,
validator: Callable[[Optional[Tuple], Optional[datetime.datetime]], bool],
tables: List[str] = None,
) -> bool:
"""Verifies the producer metadata and version of tables
Args:
producer: String name of a table producer to have validation performed
validator: callable that takes an optional version and an optional datetime that returns False if table is invalid
Returns:
False if an invalid table was found or True if no invalid table was found
"""
if tables is None:
tables = self._dict.keys()
for table_name in tables:
table = self._dict[table_name]
if not table.producer:
vollog.debug(
f"Symbol table {table_name} could not be validated because no producer metadata was found"
)
continue
if table.producer.name == producer:
# Run the verification
if not validator(
table.producer.version,
table.producer.datetime,
):
vollog.debug(f"Symbol table {table_name} does not pass validator")
return False
else:
continue
return True
### Resolution functions
class UnresolvedTemplate(objects.templates.ReferenceTemplate):
@@ -179,15 +226,15 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
if child.vol.type_name not in self._resolved:
traverse_list.append(child.vol.type_name)
try:
self._resolved[
child.vol.type_name
] = self._weak_resolve(
SymbolType.TYPE, child.vol.type_name
self._resolved[child.vol.type_name] = (
self._weak_resolve(
SymbolType.TYPE, child.vol.type_name
)
)
except exceptions.SymbolError:
self._resolved[
child.vol.type_name
] = self.UnresolvedTemplate(child.vol.type_name)
self._resolved[child.vol.type_name] = (
self.UnresolvedTemplate(child.vol.type_name)
)
# Stash the replacement
replacements.add((traverser, child))
elif child.children:
@@ -183,6 +183,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
types = _construct_delegate_function("types", True)
enumerations = _construct_delegate_function("enumerations", True)
metadata = _construct_delegate_function("metadata", True)
producer = _construct_delegate_function("producer", True)
clear_symbol_cache = _construct_delegate_function("clear_symbol_cache")
get_type = _construct_delegate_function("get_type")
get_symbol = _construct_delegate_function("get_symbol")
@@ -372,6 +373,14 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass=ABCMeta):
table."""
return None
@property
def producer(self) -> Optional["metadata.ProducerMetadata"]:
"""Returns a metadata object containing information about the symbol
table."""
return metadata.ProducerMetadata(
self._json_object.get("metadata", {}).get("producer", {})
)
def clear_symbol_cache(self) -> None:
"""Clears the symbol cache of the symbol table."""
self._symbol_cache.clear()
@@ -26,14 +26,58 @@ vollog = logging.getLogger(__name__)
class module(generic.GenericIntelProcess):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._mod_mem_type = None # Initialize _mod_mem_type to None for memoization
@property
def mod_mem_type(self):
"""Return the mod_mem_type enum choices if available or an empty dict if not"""
# mod_mem_type and module_memory were added in kernel 6.4 which replaces
# module_layout for storing the information around core_layout etc.
# see commit ac3b43283923440900b4f36ca5f9f0b1ca43b70e for more information
if self._mod_mem_type is None:
try:
self._mod_mem_type = self._context.symbol_space.get_enumeration(
self.get_symbol_table_name() + constants.BANG + "mod_mem_type"
).choices
except exceptions.SymbolError:
vollog.debug(
f"Unable to find mod_mem_type enum. This message can be ignored for kernels < 6.4"
)
# set to empty dict to show that the enum was not found, and so shouldn't be searched for again
self._mod_mem_type = {}
return self._mod_mem_type
def get_module_base(self):
if self.has_member("core_layout"):
if self.has_member("mem"): # kernels 6.4+
try:
return self.mem[self.mod_mem_type["MOD_TEXT"]].base
except KeyError:
raise AttributeError(
"module -> get_module_base: Unable to get module base. Cannot read base from MOD_TEXT."
)
elif self.has_member("core_layout"):
return self.core_layout.base
else:
elif self.has_member("module_core"):
return self.module_core
raise AttributeError("module -> get_module_base: Unable to get module base")
def get_init_size(self):
if self.has_member("init_layout"):
if self.has_member("mem"): # kernels 6.4+
try:
return (
self.mem[self.mod_mem_type["MOD_INIT_TEXT"]].size
+ self.mem[self.mod_mem_type["MOD_INIT_DATA"]].size
+ self.mem[self.mod_mem_type["MOD_INIT_RODATA"]].size
)
except KeyError:
raise AttributeError(
"module -> get_init_size: Unable to determine .init section size of module. Cannot read size of MOD_INIT_TEXT, MOD_INIT_DATA, and MOD_INIT_RODATA"
)
elif self.has_member("init_layout"):
return self.init_layout.size
elif self.has_member("init_size"):
return self.init_size
@@ -42,7 +86,19 @@ class module(generic.GenericIntelProcess):
)
def get_core_size(self):
if self.has_member("core_layout"):
if self.has_member("mem"): # kernels 6.4+
try:
return (
self.mem[self.mod_mem_type["MOD_TEXT"]].size
+ self.mem[self.mod_mem_type["MOD_DATA"]].size
+ self.mem[self.mod_mem_type["MOD_RODATA"]].size
+ self.mem[self.mod_mem_type["MOD_RO_AFTER_INIT"]].size
)
except KeyError:
raise AttributeError(
"module -> get_core_size: Unable to determine core size of module. Cannot read size of MOD_TEXT, MOD_DATA, MOD_RODATA, and MOD_RO_AFTER_INIT."
)
elif self.has_member("core_layout"):
return self.core_layout.size
elif self.has_member("core_size"):
return self.core_size
@@ -51,18 +107,32 @@ class module(generic.GenericIntelProcess):
)
def get_module_core(self):
if self.has_member("core_layout"):
if self.has_member("mem"): # kernels 6.4+
try:
return self.mem[self.mod_mem_type["MOD_TEXT"]].base
except KeyError:
raise AttributeError(
"module -> get_module_core: Unable to get module core. Cannot read base from MOD_TEXT."
)
elif self.has_member("core_layout"):
return self.core_layout.base
elif self.has_member("module_core"):
return self.module_core
raise AttributeError("module -> get_module_core: Unable to get module core")
def get_module_init(self):
if self.has_member("init_layout"):
if self.has_member("mem"): # kernels 6.4+
try:
return self.mem[self.mod_mem_type["MOD_INIT_TEXT"]].base
except KeyError:
raise AttributeError(
"module -> get_module_core: Unable to get module init. Cannot read base from MOD_INIT_TEXT."
)
elif self.has_member("init_layout"):
return self.init_layout.base
elif self.has_member("module_init"):
return self.module_init
raise AttributeError("module -> get_module_core: Unable to get module init")
raise AttributeError("module -> get_module_init: Unable to get module init")
def get_name(self):
"""Get the name of the module as a string"""
@@ -362,7 +432,7 @@ class maple_tree(objects.StructType):
# None. If however you wanted to parse from a node, but ignore some parts of the tree below it then
# this could be populated with the addresses of the nodes you wish to ignore.
if seen == None:
if seen is None:
seen = set()
# protect against unlikely loop
@@ -447,12 +517,25 @@ class maple_tree(objects.StructType):
class mm_struct(objects.StructType):
# TODO: As of version 3.0.0 this method should be removed
def get_mmap_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns an iterator for the mmap list member of an mm_struct."""
"""
Deprecated: Use either get_vma_iter() or _get_mmap_iter().
"""
vollog.warning(
"This method has been deprecated in favour of using the get_vma_iter() method."
)
yield from self.get_vma_iter()
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."""
if not self.has_member("mmap"):
raise AttributeError(
"get_mmap_iter called on mm_struct where no mmap member exists."
"_get_mmap_iter called on mm_struct where no mmap member exists."
)
if not self.mmap:
return None
@@ -466,12 +549,24 @@ class mm_struct(objects.StructType):
seen.add(link.vol.offset)
link = link.vm_next
# TODO: As of version 3.0.0 this method should be removed
def get_maple_tree_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns an iterator for the mm_mt member of an mm_struct."""
"""
Deprecated: Use either get_vma_iter() or _get_maple_tree_iter().
"""
vollog.warning(
"This method has been deprecated in favour of using the get_vma_iter() method."
)
yield from self.get_vma_iter()
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."""
if not self.has_member("mm_mt"):
raise AttributeError(
"get_maple_tree_iter called on mm_struct where no mm_mt member exists."
"_get_maple_tree_iter called on mm_struct where no mm_mt member exists."
)
symbol_table_name = self.get_symbol_table_name()
for vma_pointer in self.mm_mt.get_slot_iter():
@@ -487,9 +582,9 @@ class mm_struct(objects.StructType):
"""Returns an iterator for the VMAs in an mm_struct. Automatically choosing the mmap or mm_mt as required."""
if self.has_member("mmap"):
yield from self.get_mmap_iter()
yield from self._get_mmap_iter()
elif self.has_member("mm_mt"):
yield from self.get_maple_tree_iter()
yield from self._get_maple_tree_iter()
else:
raise AttributeError("Unable to find mmap or mm_mt in mm_struct")
@@ -1042,17 +1137,17 @@ class vfsmount(objects.StructType):
"""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
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
a 'vfsmount \*' and not a 'vfsmount \*\*'. The argument must be a 'vfsmount \*'.
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 (vfsmount *): A pointer to a 'vfsmount'
Raises:
exceptions.VolatilityException: If vfsmount_ptr is not a 'vfsmount \*'
exceptions.VolatilityException: If vfsmount_ptr is not a 'vfsmount \\*'
Returns:
bool: 'True' if the given argument points to the the same 'vfsmount'
+41 -1
View File
@@ -2,9 +2,49 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import datetime
import logging
from typing import Optional, Tuple, Union
from volatility3.framework import interfaces
from volatility3.framework import constants, interfaces
vollog = logging.getLogger(__name__)
class ProducerMetadata(interfaces.symbols.MetadataInterface):
"""Class to handle the Producer metadata from an ISF"""
@property
def name(self) -> Optional[str]:
return self._json_data.get("name", None)
@property
def version(self) -> Optional[Tuple[int]]:
"""Returns the version of the ISF file producer"""
version = self._json_data.get("version", None)
if not version:
return None
if all([x in "0123456789." for x in version]):
return tuple([int(x) for x in version.split(".")])
vollog.log(
constants.LOGLEVEL_VVVV,
f"Metadata version contains unexpected characters: '{version}'",
)
return None
@property
def datetime(self) -> Optional[datetime.datetime]:
"""Returns a timestamp for when the file was produced"""
if "datetime" not in self._json_data:
return None
try:
timestamp = datetime.datetime.strptime(
self._json_data["datetime"], "YYYY-MM-DD"
)
except (TypeError, ValueError):
vollog.debug("Invalid timestamp in producer information of symbol table")
return None
return timestamp
class WindowsMetadata(interfaces.symbols.MetadataInterface):
@@ -452,9 +452,9 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
].is_valid(self.FileName.Buffer)
def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]:
name: Union[
str, interfaces.renderers.BaseAbsentValue
] = renderers.UnreadableValue()
name: Union[str, interfaces.renderers.BaseAbsentValue] = (
renderers.UnreadableValue()
)
# this pointer needs to be checked against native_layer_name because the object may
# be instantiated from a primary (virtual) layer or a memory (physical) layer.
@@ -0,0 +1,248 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 4
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 8
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 12
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 20
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 12
}
},
"kind": "struct",
"size": 12
},
"_SERVICE_RECORD": {
"fields": {
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 48
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 160
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 12
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 24
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 56
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 44
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 160
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 52
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 156
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "vtypes_to_json.py",
"datetime": "2019-04-17T13:45:16.417006"
},
"format": "4.1.0"
}
}
@@ -0,0 +1,255 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 8,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 16
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 24
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 40
}
},
"kind": "struct",
"size": 40
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_RECORD": {
"fields": {
"ServiceList": {
"type": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
},
"offset": 0
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 32
},
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 64
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 240
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 36
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 76
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 56
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 240
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 72
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 32
}
},
"kind": "struct",
"size": 248
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "vtypes_to_json.py",
"datetime": "2019-04-17T13:45:16.417006"
},
"format": "4.1.0"
}
}
@@ -0,0 +1,248 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 4
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 8
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 12
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 20
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 12
}
},
"kind": "struct",
"size": 12
},
"_SERVICE_RECORD": {
"fields": {
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 48
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 164
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 12
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 24
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 56
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 44
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 164
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 52
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 156
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "vtypes_to_json.py",
"datetime": "2019-04-17T13:45:16.417006"
},
"format": "4.1.0"
}
}
@@ -0,0 +1,255 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 8,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 16
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 24
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 40
}
},
"kind": "struct",
"size": 40
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_RECORD": {
"fields": {
"ServiceList": {
"type": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
},
"offset": 0
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 32
},
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 64
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 296
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 36
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 76
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 56
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 296
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 72
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 32
}
},
"kind": "struct",
"size": 296
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "David McDonald",
"datetime": "2023-11-16T15:05:35-06:00"
},
"format": "4.1.0"
}
}
@@ -0,0 +1,248 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 4
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 8
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 12
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 20
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 12
},
"_SERVICE_RECORD": {
"fields": {
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 48
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 192
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 12
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 24
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 56
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 44
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 192
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 52
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
}
},
"kind": "struct",
"size": 192
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "vtypes_to_json.py",
"datetime": "2019-04-17T13:45:16.417006"
},
"format": "4.1.0"
}
}
@@ -0,0 +1,255 @@
{
"symbols": {},
"enums": {
"StateEnum": {
"base": "long",
"constants": {
"SERVICE_START_PENDING": 2,
"SERVICE_STOP_PENDING": 3,
"SERVICE_STOPPED": 1,
"SERVICE_CONTINUE_PENDING": 5,
"SERVICE_PAUSE_PENDING": 6,
"SERVICE_PAUSED": 7,
"SERVICE_RUNNING": 4
},
"size": 4
},
"StartEnum": {
"base": "long",
"constants": {
"SERVICE_DEMAND_START": 3,
"SERVICE_AUTO_START": 2,
"SERVICE_BOOT_START": 0,
"SERVICE_DISABLED": 4,
"SERVICE_SYSTEM_START": 1
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "char",
"size": 1,
"signed": false,
"endian": "little"
},
"pointer": {
"kind": "int",
"size": 8,
"signed": false,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
}
},
"user_types": {
"_SERVICE_LIST_ENTRY": {
"fields": {
"Flink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 16
},
"Blink": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
}
},
"offset": 0
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_PROCESS": {
"fields": {
"BinaryPath": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 24
},
"ProcessId": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 40
}
},
"kind": "struct",
"size": 40
},
"_SERVICE_HEADER": {
"fields": {
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 0
},
"ServiceRecord": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 16
},
"_SERVICE_RECORD": {
"fields": {
"ServiceList": {
"type": {
"kind": "struct",
"name": "_SERVICE_LIST_ENTRY"
},
"offset": 0
},
"Tag": {
"type": {
"count": 4,
"subtype": {
"kind": "base",
"name": "unsigned char"
},
"kind": "array"
},
"offset": 32
},
"DisplayName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 64
},
"ServiceProcess": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_PROCESS"
}
},
"offset": 336
},
"PrevEntry": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "_SERVICE_RECORD"
}
},
"offset": 16
},
"Start": {
"type": {
"kind": "enum",
"name": "StartEnum"
},
"offset": 36
},
"State": {
"type": {
"kind": "enum",
"name": "StateEnum"
},
"offset": 84
},
"ServiceName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 56
},
"DriverName": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "unsigned short"
}
},
"offset": 296
},
"Type": {
"type": {
"kind": "base",
"name": "unsigned long"
},
"offset": 80
},
"Order": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 32
}
},
"kind": "struct",
"size": 336
}
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "David McDonald",
"datetime": "2023-11-16T15:05:35-06:00"
},
"format": "4.1.0"
}
}
@@ -151,11 +151,45 @@ is_win10_16299_or_later = OsDistinguisher(
],
)
is_win10_17763_or_later = OsDistinguisher(
version_check=lambda x: x >= (10, 0, 17763),
fallback_checks=[
("_EPROCESS", "TrustletIdentity", False),
("ParentSecurityDomain", None, True),
],
)
is_win10_18362_or_later = OsDistinguisher(
version_check=lambda x: x >= (10, 0, 18362),
fallback_checks=[
("ObHeaderCookie", None, True),
("_CM_CACHED_VALUE_INDEX", None, False),
("_WNF_PROCESS_CONTEXT", None, True),
],
)
is_win10_18363_or_later = OsDistinguisher(
version_check=lambda x: x >= (10, 0, 18363),
fallback_checks=[("_KQOS_GROUPING_SETS", None, True)],
)
is_win10_19041_or_later = OsDistinguisher(
version_check=lambda x: x >= (10, 0, 19041),
fallback_checks=[
("_EPROCESS", "TimerResolutionIgnore", True),
("_EPROCESS", "VmProcessorHostTransition", True),
("_KQOS_GROUPING_SETS", None, True),
],
)
is_win10_25398_or_later = OsDistinguisher(
version_check=lambda x: x >= (10, 0, 25398),
fallback_checks=[
("_EPROCESS", "MmSlabIdentity", True),
("_EPROCESS", "EnableProcessImpersonationLogging", True),
],
)
is_windows_10 = OsDistinguisher(
version_check=lambda x: x >= (10, 0),
fallback_checks=[("ObHeaderCookie", None, True)],
+3 -7
View File
@@ -31,13 +31,9 @@ class Statistics(plugins.PluginInterface):
# Do mass mapping and determine the number of different layers and how many pages go to each one
layer = self.context.layers[self.config["primary"]]
page_count = (
swap_count
) = (
invalid_page_count
) = (
large_page_count
) = large_swap_count = large_invalid_count = other_invalid = 0
page_count = swap_count = invalid_page_count = large_page_count = (
large_swap_count
) = large_invalid_count = other_invalid = 0
if isinstance(layer, intel.Intel):
page_addr = 0
+507
View File
@@ -0,0 +1,507 @@
{
"$schema": "http://json-schema.org/schema#",
"id": "http://volatilityfoundation.org/intermediate-format/schema",
"title": "Symbol Container",
"type": "object",
"definitions": {
"metadata_producer": {
"type": "object",
"properties": {
"name": {
"type": "string"
},
"version": {
"type": "string",
"pattern": "^[0-9]+.[0-9]+.[0-9]+$"
},
"datetime": {
"type": "string",
"format": "date-time"
}
},
"required":[
"name",
"version"
]
},
"metadata_windows_pe": {
"type": "object",
"properties": {
"major": {
"type": "integer"
},
"minor": {
"type": "integer"
},
"revision": {
"type": "integer"
},
"build": {
"type": "integer"
}
},
"additionalProperties": false,
"required": [
"major",
"minor",
"revision"
]
},
"metadata_windows_pdb": {
"type": "object",
"properties": {
"GUID": {
"type": "string"
},
"age": {
"type": "integer"
},
"database": {
"type": "string"
},
"machine_type": {
"type": "integer"
}
},
"additionalProperties": false,
"required": [
"GUID",
"age",
"database",
"machine_type"
]
},
"metadata_windows": {
"type": "object",
"properties": {
"pe": {
"$ref": "#/definitions/metadata_windows_pe"
},
"pdb": {
"$ref": "#/definitions/metadata_windows_pdb"
}
},
"additionalProperties": false
},
"metadata_nix": {
"type": "object",
"properties": {
"symbols": {
"type": "array",
"items": {
"$ref": "#/definitions/metadata_nix_item"
}
},
"types": {
"type": "array",
"items": {
"$ref": "#/definitions/metadata_nix_item"
}
}
},
"additionalProperties": false
},
"metadata_format": {
"type": "string",
"pattern": "^6.[0-9]+.[0-9]+$"
},
"metadata_nix_item": {
"type": "object",
"properties": {
"kind": {
"type": "string",
"pattern": "^(dwarf|symtab|system-map)$"
},
"name": {
"type": "string"
},
"hash_type": {
"type": "string",
"pattern": "^(sha256)$"
},
"hash_value": {
"type": "string",
"pattern": "^[a-fA-F0-9]+$"
}
},
"additionalProperties": false
},
"element_metadata": {
"type": "object",
"oneOf": [
{
"properties": {
"format": {
"$ref": "#/definitions/metadata_format"
},
"producer": {
"$ref": "#/definitions/metadata_producer"
}
},
"required": [
"format",
"producer"
],
"additionalProperties": false
},
{
"properties": {
"format": {
"$ref": "#/definitions/metadata_format"
},
"producer": {
"$ref": "#/definitions/metadata_producer"
},
"windows": {
"$ref": "#/definitions/metadata_windows"
}
},
"required": [
"format",
"producer",
"windows"
],
"additionalProperties": false
},
{
"properties": {
"format": {
"$ref": "#/definitions/metadata_format"
},
"producer": {
"$ref": "#/definitions/metadata_producer"
},
"linux": {
"$ref": "#/definitions/metadata_nix"
}
},
"required": [
"format",
"producer",
"linux"
],
"additionalProperties": false
},
{
"properties": {
"format": {
"$ref": "#/definitions/metadata_format"
},
"producer": {
"$ref": "#/definitions/metadata_producer"
},
"mac": {
"$ref": "#/definitions/metadata_nix"
}
},
"required": [
"format",
"producer",
"mac"
],
"additionalProperties": false
}
]
},
"element_enum": {
"properties": {
"size": {
"type": "integer"
},
"base": {
"type": "string"
},
"constants": {
"type": "object",
"additionalProperties": {
"type": "integer"
}
}
},
"required": [
"size",
"base",
"constants"
],
"additionalProperties": false
},
"element_symbol": {
"properties": {
"address": {
"type": "number"
},
"linkage_name": {
"type": "string"
},
"type": {
"$ref": "#/definitions/type_descriptor"
},
"constant_data": {
"type": "string",
"media": {
"binaryEncoding": "base64",
"readOnly": true
}
}
},
"required": [
"address"
],
"additionalProperties": false
},
"element_base_type": {
"properties": {
"size": {
"type": "integer"
},
"signed": {
"type": "boolean"
},
"kind": {
"type": "string",
"pattern": "^(void|int|float|char|bool)$"
},
"endian": {
"type": "string",
"pattern": "^(little|big)$"
}
},
"required": [
"size",
"kind",
"signed",
"endian"
],
"additionalProperties": false
},
"element_user_type": {
"properties": {
"kind": {
"type": "string",
"pattern": "^(struct|union|class)$"
},
"size": {
"type": "integer"
},
"fields": {
"type": "object",
"additionalProperties": {
"$ref": "#/definitions/field"
}
}
},
"required": [
"kind",
"size",
"fields"
],
"additionalProperties": false
},
"field": {
"properties": {
"type": {
"$ref": "#/definitions/type_descriptor"
},
"offset": {
"type": "integer"
},
"anonymous": {
"type": "boolean"
}
},
"required": [
"type",
"offset"
],
"additionalProperties": false
},
"type_descriptor": {
"oneOf": [
{
"$ref": "#/definitions/type_pointer"
},
{
"$ref": "#/definitions/type_base"
},
{
"$ref": "#/definitions/type_array"
},
{
"$ref": "#/definitions/type_struct"
},
{
"$ref": "#/definitions/type_enum"
},
{
"$ref": "#/definitions/type_function"
},
{
"$ref": "#/definitions/type_bitfield"
}
]
},
"type_pointer": {
"properties": {
"kind": {
"type": "string",
"pattern": "^pointer$"
},
"base": {
"type": "string"
},
"subtype": {
"$ref": "#/definitions/type_descriptor"
}
},
"required": [
"kind",
"subtype"
],
"additionalProperties": false
},
"type_base": {
"properties": {
"kind": {
"type": "string",
"pattern": "^base$"
},
"name": {
"type": "string"
}
},
"required": [
"kind",
"name"
],
"additionalProperties": false
},
"type_array": {
"properties": {
"kind": {
"type": "string",
"pattern": "^array$"
},
"subtype": {
"$ref": "#/definitions/type_descriptor"
},
"count": {
"type": "integer"
}
},
"required": [
"kind",
"subtype",
"count"
],
"additionalProperties": false
},
"type_struct": {
"properties": {
"kind": {
"type": "string",
"pattern": "^(struct|class|union)$"
},
"name": {
"type": "string"
}
},
"required": [
"kind",
"name"
],
"additionalProperties": false
},
"type_enum": {
"properties": {
"kind": {
"type": "string",
"pattern": "^enum$"
},
"name": {
"type": "string"
}
},
"required": [
"kind",
"name"
],
"additionalProperties": false
},
"type_function": {
"properties": {
"kind": {
"type": "string",
"pattern": "^function$"
}
},
"required": [
"kind"
],
"additionalProperties": false
},
"type_bitfield": {
"properties": {
"kind": {
"type": "string",
"pattern": "^bitfield$"
},
"bit_position": {
"type": "integer"
},
"bit_length": {
"type": "integer"
},
"type": {
"oneOf": [
{
"$ref": "#/definitions/type_base"
},
{
"$ref": "#/definitions/type_enum"
}
]
}
},
"required": [
"kind",
"bit_position",
"bit_length",
"type"
],
"additionalProperties": false
}
},
"properties": {
"metadata": {
"$ref": "#/definitions/element_metadata"
},
"base_types": {
"additionalProperties": {
"$ref": "#/definitions/element_base_type"
}
},
"user_types": {
"additionalProperties": {
"$ref": "#/definitions/element_user_type"
}
},
"enums": {
"additionalProperties": {
"$ref": "#/definitions/element_enum"
}
},
"symbols": {
"additionalProperties": {
"$ref": "#/definitions/element_symbol"
}
}
},
"required": [
"metadata",
"base_types",
"user_types",
"enums",
"symbols"
],
"additionalProperties": false
}