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

Merge pull request #551 from volatilityfoundation/feature/windows-kernel-module

Browse Source 
Windows: Update to kernel ModuleRequirement for ease of config
This commit is contained in:
ikelos
2021-08-22 12:28:47 +01:00
committed by GitHub
parent ab075ac667 97da37d968
commit 7bf3ac4a41
41 changed files with 522 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/windows/bigpools.py
+7 -7
View File
@@ -20,17 +20,15 @@ vollog = logging.getLogger(__name__)
class BigPools(interfaces.plugins.PluginInterface):
"""List big page pools."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.StringRequirement(name = 'tags',
description = "Comma separated list of pool tags to filter pools returned",
optional = True,
@@ -108,9 +106,11 @@ class BigPools(interfaces.plugins.PluginInterface):
else:
tags = None
kernel = self.context.modules[self.config['kernel']]
for big_pool in self.list_big_pools(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
tags = tags):
num_bytes = big_pool.get_number_of_bytes()
volatility3/framework/plugins/windows/cachedump.py
+13 -10
View File
@@ -21,16 +21,14 @@ vollog = logging.getLogger(__name__)
class Cachedump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'hashdump', plugin = hashdump.Hashdump, version = (1, 1, 0))
@@ -46,7 +44,7 @@ class Cachedump(interfaces.plugins.PluginInterface):
hmac_md5 = HMAC.new(nlkm, ch)
rc4key = hmac_md5.digest()
rc4 = ARC4.new(rc4key)
data = rc4.encrypt(edata) # lgtm [py/weak-cryptographic-algorithm]
data = rc4.encrypt(edata) # lgtm [py/weak-cryptographic-algorithm]
else:
# based on Based on code from http://lab.mediaservice.net/code/cachedump.rb
aes = AES.new(nlkm[16:32], AES.MODE_CBC, ch)
@@ -90,7 +88,10 @@ class Cachedump(interfaces.plugins.PluginInterface):
vollog.warning('Unable to find bootkey')
return
vista_or_later = versions.is_vista_or_later(context = self.context, symbol_table = self.config['nt_symbols'])
kernel = self.context.modules[self.config['kernel']]
vista_or_later = versions.is_vista_or_later(context = self.context,
symbol_table = kernel.symbol_table_name)
lsakey = lsadump.Lsadump.get_lsa_key(sechive, bootkey, vista_or_later)
if not lsakey:
@@ -129,10 +130,12 @@ class Cachedump(interfaces.plugins.PluginInterface):
syshive = sechive = None
kernel = self.context.modules[self.config['kernel']]
for hive in hivelist.HiveList.list_hives(self.context,
self.config_path,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
hive_offsets = None if offset is None else [offset]):
if hive.get_name().split('\\')[-1].upper() == 'SYSTEM':
@@ -147,5 +150,5 @@ class Cachedump(interfaces.plugins.PluginInterface):
vollog.warning('Unable to locate SECURITY hive')
return
return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hashh', bytes)],
return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hash', bytes)],
self._generator(syshive, sechive))
volatility3/framework/plugins/windows/callbacks.py
+12 -10
View File
@@ -19,16 +19,14 @@ vollog = logging.getLogger(__name__)
class Callbacks(interfaces.plugins.PluginInterface):
"""Lists kernel callbacks and notification routines."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'ssdt', plugin = ssdt.SSDT, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'svcscan', plugin = svcscan.SvcScan, version = (1, 0, 0))
]
@@ -191,7 +189,8 @@ class Callbacks(interfaces.plugins.PluginInterface):
continue
try:
component: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface] = ntkrnlmp.object(
component: Union[
interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface] = ntkrnlmp.object(
"string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace"
)
except exceptions.InvalidAddressException:
@@ -244,17 +243,20 @@ class Callbacks(interfaces.plugins.PluginInterface):
def _generator(self):
callback_table_name = self.create_callback_table(self.context, self.config["nt_symbols"], self.config_path)
kernel = self.context.modules[self.config['kernel']]
collection = ssdt.SSDT.build_module_collection(self.context, self.config['primary'], self.config['nt_symbols'])
callback_table_name = self.create_callback_table(self.context, kernel.symbol_table_name,
self.config_path)
collection = ssdt.SSDT.build_module_collection(self.context, kernel.layer_name, kernel.symbol_table_name)
callback_methods = (self.list_notify_routines, self.list_bugcheck_callbacks,
self.list_bugcheck_reason_callbacks, self.list_registry_callbacks)
for callback_method in callback_methods:
for callback_type, callback_address, callback_detail in callback_method(self.context,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
callback_table_name):
if callback_detail is None:
volatility3/framework/plugins/windows/cmdline.py
+11 -8
View File
@@ -15,17 +15,15 @@ vollog = logging.getLogger(__name__)
class CmdLine(interfaces.plugins.PluginInterface):
"""Lists process command line arguments."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.ListRequirement(name = 'pid',
element_type = int,
@@ -57,13 +55,15 @@ class CmdLine(interfaces.plugins.PluginInterface):
def _generator(self, procs):
kernel = self.context.modules[self.config['kernel']]
for proc in procs:
process_name = utility.array_to_string(proc.ImageFileName)
proc_id = "Unknown"
try:
proc_id = proc.UniqueProcessId
result_text = self.get_cmdline(self.context, self.config["nt_symbols"], proc)
result_text = self.get_cmdline(self.context, kernel.symbol_table_name, proc)
except exceptions.SwappedInvalidAddressException as exp:
result_text = f"Required memory at {exp.invalid_address:#x} is inaccessible (swapped)"
@@ -78,11 +78,14 @@ class CmdLine(interfaces.plugins.PluginInterface):
yield (0, (proc.UniqueProcessId, process_name, result_text))
def run(self):
kernel = self.context.modules[self.config['kernel']]
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
return renderers.TreeGrid([("PID", int), ("Process", str), ("Args", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/dlllist.py
+13 -10
View File
@@ -20,17 +20,15 @@ vollog = logging.getLogger(__name__)
class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Lists the loaded modules in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'info', component = info.Info, version = (1, 0, 0)),
requirements.ListRequirement(name = 'pid',
@@ -94,7 +92,9 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"pe",
class_types = pe.class_types)
kuser = info.Info.get_kuser_structure(self.context, self.config['primary'], self.config['nt_symbols'])
kernel = self.context.modules[self.config['kernel']]
kuser = info.Info.get_kuser_structure(self.context, kernel.layer_name, kernel.symbol_table_name)
nt_major_version = int(kuser.NtMajorVersion)
nt_minor_version = int(kuser.NtMinorVersion)
# LoadTime only applies to versions higher or equal to Window 7 (6.1 and higher)
@@ -144,10 +144,12 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
format_hints.Hex(entry.SizeOfImage), BaseDllName, FullDllName, DllLoadTime, file_output))
def generate_timeline(self):
kernel = self.context.modules[self.config['kernel']]
for row in self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'])):
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name)):
_depth, row_data = row
if not isinstance(row_data[6], datetime.datetime):
continue
@@ -157,12 +159,13 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("Base", format_hints.Hex),
("Size", format_hints.Hex), ("Name", str), ("Path", str),
("LoadTime", datetime.datetime), ("File output", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/driverirp.py
+6 -8
View File
@@ -22,25 +22,23 @@ MAJOR_FUNCTIONS = [
class DriverIrp(interfaces.plugins.PluginInterface):
"""List IRPs for drivers in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'ssdt', plugin = ssdt.SSDT, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'driverscan', plugin = driverscan.DriverScan, version = (1, 0, 0)),
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
]
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
collection = ssdt.SSDT.build_module_collection(self.context, self.config['primary'], self.config['nt_symbols'])
collection = ssdt.SSDT.build_module_collection(self.context, kernel.layer_name, kernel.symbol_table_name)
for driver in driverscan.DriverScan.scan_drivers(self.context, self.config['primary'],
self.config['nt_symbols']):
for driver in driverscan.DriverScan.scan_drivers(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
driver_name = driver.get_driver_name()
volatility3/framework/plugins/windows/driverscan.py
+6 -6
View File
@@ -13,16 +13,14 @@ from volatility3.plugins.windows import poolscanner
class DriverScan(interfaces.plugins.PluginInterface):
"""Scans for drivers present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
]
@@ -51,7 +49,9 @@ class DriverScan(interfaces.plugins.PluginInterface):
yield mem_object
def _generator(self):
for driver in self.scan_drivers(self.context, self.config['primary'], self.config['nt_symbols']):
kernel = self.context.modules[self.config['kernel']]
for driver in self.scan_drivers(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
driver_name = driver.get_driver_name()
volatility3/framework/plugins/windows/dumpfiles.py
+25 -22
View File
@@ -4,12 +4,13 @@
import logging
import ntpath
from typing import List, Tuple, Type, Optional, Generator
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.plugins.windows import handles
from volatility3.plugins.windows import pslist
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from typing import List, Tuple, Type, Optional, Generator
from volatility3.plugins.windows import handles
from volatility3.plugins.windows import pslist
vollog = logging.getLogger(__name__)
@@ -25,17 +26,15 @@ EXTENSION_CACHE_MAP = {
class DumpFiles(interfaces.plugins.PluginInterface):
"""Dumps cached file contents from Windows memory samples."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.IntRequirement(name = 'pid',
description = "Process ID to include (all other processes are excluded)",
optional = True),
@@ -167,6 +166,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
file_output)
def _generator(self, procs: List, offsets: List):
kernel = self.context.modules[self.config['kernel']]
if procs:
# The handles plugin doesn't expose any staticmethod/classmethod, and it also requires stashing
@@ -175,11 +175,11 @@ class DumpFiles(interfaces.plugins.PluginInterface):
# results instead of just dealing with them as direct objects here.
handles_plugin = handles.Handles(context = self.context, config_path = self._config_path)
type_map = handles_plugin.get_type_map(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"])
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name)
cookie = handles_plugin.find_cookie(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"])
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name)
for proc in procs:
@@ -195,7 +195,7 @@ 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")
for result in self.process_file_object(self.context, self.config["primary"], self.open,
for result in self.process_file_object(self.context, kernel.layer_name, self.open,
file_obj):
yield (0, result)
except exceptions.InvalidAddressException:
@@ -219,7 +219,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
if not file_obj.is_valid():
continue
for result in self.process_file_object(self.context, self.config["primary"], self.open,
for result in self.process_file_object(self.context, kernel.layer_name, self.open,
file_obj):
yield (0, result)
except exceptions.InvalidAddressException:
@@ -230,16 +230,17 @@ class DumpFiles(interfaces.plugins.PluginInterface):
# Now process any offsets explicitly requested by the user.
for offset, is_virtual in offsets:
try:
layer_name = self.config["primary"]
layer_name = kernel.layer_name
# switch to a memory layer if the user provided --physaddr instead of --virtaddr
if not is_virtual:
layer_name = self.context.layers[layer_name].config["memory_layer"]
file_obj = self.context.object(self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT",
layer_name = layer_name,
native_layer_name = self.config["primary"],
offset = offset)
for result in self.process_file_object(self.context, self.config["primary"], self.open, file_obj):
file_obj = self.context.object(
kernel.symbol_table_name + constants.BANG + "_FILE_OBJECT",
layer_name = layer_name,
native_layer_name = kernel.layer_name,
offset = offset)
for result in self.process_file_object(self.context, kernel.layer_name, self.open, file_obj):
yield (0, result)
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, f"Cannot extract file at {offset:#x}")
@@ -250,6 +251,8 @@ class DumpFiles(interfaces.plugins.PluginInterface):
# a list of processes matching the pid filter. all files for these process(es) will be dumped.
procs = []
kernel = self.context.modules[self.config['kernel']]
if self.config.get("virtaddr", None) is not None:
offsets.append((self.config["virtaddr"], True))
elif self.config.get("physaddr", None) is not None:
@@ -257,8 +260,8 @@ class DumpFiles(interfaces.plugins.PluginInterface):
else:
filter_func = pslist.PsList.create_pid_filter([self.config.get("pid", None)])
procs = pslist.PsList.list_processes(self.context,
self.config["primary"],
self.config["nt_symbols"],
kernel.layer_name,
kernel.symbol_table_name,
filter_func = filter_func)
return renderers.TreeGrid([("Cache", str), ("FileObject", format_hints.Hex), ("FileName", str),
volatility3/framework/plugins/windows/envars.py
+9 -9
View File
@@ -15,17 +15,15 @@ vollog = logging.getLogger(__name__)
class Envars(interfaces.plugins.PluginInterface):
"Display process environment variables"
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
description = 'Filter on specific process IDs',
element_type = int,
@@ -48,11 +46,12 @@ class Envars(interfaces.plugins.PluginInterface):
"""
values = []
kernel = self.context.modules[self.config['kernel']]
for hive in hivelist.HiveList.list_hives(context = self.context,
base_config_path = self.config_path,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
hive_offsets = None):
sys = False
ntuser = False
@@ -192,10 +191,11 @@ class Envars(interfaces.plugins.PluginInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("Block", str), ("Variable", str), ("Value", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/filescan.py
+6 -6
View File
@@ -13,15 +13,13 @@ from volatility3.plugins.windows import poolscanner
class FileScan(interfaces.plugins.PluginInterface):
"""Scans for file objects present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
]
@@ -50,7 +48,9 @@ class FileScan(interfaces.plugins.PluginInterface):
yield mem_object
def _generator(self):
for fileobj in self.scan_files(self.context, self.config['primary'], self.config['nt_symbols']):
kernel = self.context.modules[self.config['kernel']]
for fileobj in self.scan_files(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
file_name = fileobj.FileName.String
volatility3/framework/plugins/windows/getservicesids.py
+6 -8
View File
@@ -30,8 +30,8 @@ def createservicesid(svc) -> str:
class GetServiceSIDs(interfaces.plugins.PluginInterface):
"""Lists process token sids."""
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -53,20 +53,18 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0))
]
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
# Get the system hive
for hive in hivelist.HiveList.list_hives(context = self.context,
base_config_path = self.config_path,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_string = 'machine\\system',
hive_offsets = None):
# Get ControlSet\Services.
volatility3/framework/plugins/windows/getsids.py
+9 -9
View File
@@ -28,8 +28,8 @@ def find_sid_re(sid_string, sid_re_list) -> Union[str, interfaces.renderers.Base
class GetSIDs(interfaces.plugins.PluginInterface):
"""Print the SIDs owning each process"""
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -53,10 +53,8 @@ class GetSIDs(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
description = 'Filter on specific process IDs',
element_type = int,
@@ -75,12 +73,13 @@ class GetSIDs(interfaces.plugins.PluginInterface):
key = "Microsoft\\Windows NT\\CurrentVersion\\ProfileList"
val = "ProfileImagePath"
kernel = self.context.modules[self.config['kernel']]
sids = {}
for hive in hivelist.HiveList.list_hives(context = self.context,
base_config_path = self.config_path,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_string = 'config\\software',
hive_offsets = None):
@@ -154,10 +153,11 @@ class GetSIDs(interfaces.plugins.PluginInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("SID", str), ("Name", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/handles.py
+24 -19
View File
@@ -24,7 +24,7 @@ except ImportError:
class Handles(interfaces.plugins.PluginInterface):
"""Lists process open handles."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
def __init__(self, *args, **kwargs):
@@ -38,10 +38,8 @@ class Handles(interfaces.plugins.PluginInterface):
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
@@ -69,7 +67,9 @@ class Handles(interfaces.plugins.PluginInterface):
process' handle table, determine where the corresponding object's
_OBJECT_HEADER can be found."""
virtual = self.config["primary"]
kernel = self.context.modules[self.config['kernel']]
virtual = kernel.layer_name
try:
# before windows 7
@@ -80,7 +80,7 @@ class Handles(interfaces.plugins.PluginInterface):
object_header.GrantedAccess = handle_table_entry.GrantedAccess
except AttributeError:
# starting with windows 8
is_64bit = symbols.symbol_table_is_64bit(self.context, self.config["nt_symbols"])
is_64bit = symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name)
if is_64bit:
if handle_table_entry.LowValue == 0:
@@ -104,8 +104,7 @@ class Handles(interfaces.plugins.PluginInterface):
offset = handle_table_entry.InfoTable & ~7
# print("LowValue: {0:#x} Magic: {1:#x} Offset: {2:#x}".format(handle_table_entry.InfoTable, magic, offset))
object_header = self.context.object(self.config["nt_symbols"] + constants.BANG + "_OBJECT_HEADER",
virtual,
object_header = self.context.object(kernel.symbol_table_name + constants.BANG + "_OBJECT_HEADER", virtual,
offset = offset)
object_header.GrantedAccess = handle_table_entry.GrantedAccessBits
@@ -124,10 +123,11 @@ class Handles(interfaces.plugins.PluginInterface):
if not has_capstone:
return None
kernel = self.context.modules[self.config['kernel']]
virtual_layer_name = self.config['primary']
virtual_layer_name = kernel.layer_name
kvo = self.context.layers[virtual_layer_name].config['kernel_virtual_offset']
ntkrnlmp = self.context.module(self.config["nt_symbols"], layer_name = virtual_layer_name, offset = kvo)
ntkrnlmp = self.context.module(kernel.symbol_table_name, layer_name = virtual_layer_name, offset = kvo)
try:
func_addr = ntkrnlmp.get_symbol("ObpCaptureHandleInformationEx").address
@@ -227,10 +227,12 @@ class Handles(interfaces.plugins.PluginInterface):
"""Parse a process' handle table and yield valid handle table entries,
going as deep into the table "levels" as necessary."""
virtual = self.config["primary"]
kernel = self.context.modules[self.config['kernel']]
virtual = kernel.layer_name
kvo = self.context.layers[virtual].config['kernel_virtual_offset']
ntkrnlmp = self.context.module(self.config["nt_symbols"], layer_name = virtual, offset = kvo)
ntkrnlmp = self.context.module(kernel.symbol_table_name, layer_name = virtual, offset = kvo)
if level > 0:
subtype = ntkrnlmp.get_type("pointer")
@@ -292,13 +294,15 @@ class Handles(interfaces.plugins.PluginInterface):
def _generator(self, procs):
kernel = self.context.modules[self.config['kernel']]
type_map = self.get_type_map(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"])
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name)
cookie = self.find_cookie(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"])
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name)
for proc in procs:
try:
@@ -345,12 +349,13 @@ class Handles(interfaces.plugins.PluginInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("Offset", format_hints.Hex),
("HandleValue", format_hints.Hex), ("Type", str),
("GrantedAccess", format_hints.Hex), ("Name", str)],
self._generator(
pslist.PsList.list_processes(self.context,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/hashdump.py
+11 -11
View File
@@ -21,16 +21,14 @@ vollog = logging.getLogger(__name__)
class Hashdump(interfaces.plugins.PluginInterface):
"""Dumps user hashes from memory"""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0))
]
@@ -134,7 +132,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
hbootkey = rc4.encrypt(sam_data[0x80:0xA0]) # lgtm [py/weak-cryptographic-algorithm]
hbootkey = rc4.encrypt(sam_data[0x80:0xA0]) # lgtm [py/weak-cryptographic-algorithm]
return hbootkey
elif revision == 3:
# AES encrypted
@@ -153,7 +151,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
des2 = DES.new(des_k2, DES.MODE_ECB)
cipher = AES.new(hbootkey[:16], AES.MODE_CBC, salt)
obfkey = cipher.decrypt(enc_hash)
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:16]) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:16]) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_hashes(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive,
@@ -231,9 +229,9 @@ class Hashdump(interfaces.plugins.PluginInterface):
md5.update(hbootkey[:0x10] + pack("<L", rid) + lmntstr)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_name(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive) -> Optional[bytes]:
@@ -289,10 +287,12 @@ class Hashdump(interfaces.plugins.PluginInterface):
offset = self.config.get('offset', None)
syshive = None
samhive = None
kernel = self.context.modules[self.config['kernel']]
for hive in hivelist.HiveList.list_hives(self.context,
self.config_path,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
hive_offsets = None if offset is None else [offset]):
if hive.get_name().split('\\')[-1].upper() == 'SYSTEM':
volatility3/framework/plugins/windows/info.py
+13 -11
View File
@@ -16,16 +16,14 @@ from volatility3.framework.symbols.windows import extensions
class Info(plugins.PluginInterface):
"""Show OS & kernel details of the memory sample being analyzed."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
]
@classmethod
@@ -150,18 +148,22 @@ class Info(plugins.PluginInterface):
def _generator(self):
layer_name = self.config['primary']
symbol_table = self.config['nt_symbols']
kernel = self.context.modules[self.config['kernel']]
layer_name = kernel.layer_name
symbol_table = kernel.symbol_table_name
layer = self.context.layers[layer_name]
table = self.context.symbol_space[symbol_table]
kdbg = self.get_kdbg_structure(self.context, self.config_path, layer_name, symbol_table)
yield (0, ("Kernel Base", hex(self.config["primary.kernel_virtual_offset"])))
yield (0, ("DTB", hex(self.config["primary.page_map_offset"])))
yield (0, ("Symbols", self.config["nt_symbols.isf_url"]))
yield (0, ("Kernel Base", hex(layer.config["kernel_virtual_offset"])))
yield (0, ("DTB", hex(layer.config["page_map_offset"])))
yield (0, ("Symbols", table.config["isf_url"]))
yield (0, ("Is64Bit", str(symbols.symbol_table_is_64bit(self.context, symbol_table))))
yield (0, ("IsPAE", str(self.context.layers[layer_name].metadata.get("pae", False))))
for i, layer in self.get_depends(self.context, "primary"):
for i, layer in self.get_depends(self.context, layer_name):
yield (0, (layer.name, f"{i} {layer.__class__.__name__}"))
if kdbg.Header.OwnerTag == 0x4742444B:
volatility3/framework/plugins/windows/lsadump.py
+12 -10
View File
@@ -21,16 +21,14 @@ vollog = logging.getLogger(__name__)
class Lsadump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'hashdump', component = hashdump.Hashdump, version = (1, 1, 0)),
requirements.VersionRequirement(name = 'hivelist', component = hivelist.HiveList, version = (1, 0, 0))
]
@@ -86,7 +84,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
rc4key = md5.digest()
rc4 = ARC4.new(rc4key)
lsa_key = rc4.decrypt(obf_lsa_key[12:60]) # lgtm [py/weak-cryptographic-algorithm]
lsa_key = rc4.decrypt(obf_lsa_key[12:60]) # lgtm [py/weak-cryptographic-algorithm]
lsa_key = lsa_key[0x10:0x20]
else:
lsa_key = cls.decrypt_aes(obf_lsa_key, bootkey)
@@ -127,7 +125,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
des_key = hashdump.Hashdump.sidbytes_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
enc_block = enc_block + b"\x00" * int(abs(8 - len(enc_block)) % 8)
decrypted_data += des.decrypt(enc_block) # lgtm [py/weak-cryptographic-algorithm]
decrypted_data += des.decrypt(enc_block) # lgtm [py/weak-cryptographic-algorithm]
j += 7
if len(key[j:j + 7]) < 7:
j = len(key[j:j + 7])
@@ -138,7 +136,10 @@ class Lsadump(interfaces.plugins.PluginInterface):
def _generator(self, syshive: registry.RegistryHive, sechive: registry.RegistryHive):
vista_or_later = versions.is_vista_or_later(context = self.context, symbol_table = self.config['nt_symbols'])
kernel = self.context.modules[self.config['kernel']]
vista_or_later = versions.is_vista_or_later(context = self.context,
symbol_table = kernel.symbol_table_name)
bootkey = hashdump.Hashdump.get_bootkey(syshive)
lsakey = self.get_lsa_key(sechive, bootkey, vista_or_later)
@@ -181,11 +182,12 @@ class Lsadump(interfaces.plugins.PluginInterface):
offset = self.config.get('offset', None)
syshive = sechive = None
kernel = self.context.modules[self.config['kernel']]
for hive in hivelist.HiveList.list_hives(self.context,
self.config_path,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
hive_offsets = None if offset is None else [offset]):
if hive.get_name().split('\\')[-1].upper() == 'SYSTEM':
volatility3/framework/plugins/windows/malfind.py
+14 -14
View File
@@ -17,16 +17,14 @@ vollog = logging.getLogger(__name__)
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
@@ -105,8 +103,8 @@ class Malfind(interfaces.plugins.PluginInterface):
continue
if (vad.get_private_memory() == 1
and vad.get_tag() == "VadS") or (vad.get_private_memory() == 0
and protection_string != "PAGE_EXECUTE_WRITECOPY"):
and vad.get_tag() == "VadS") or (vad.get_private_memory() == 0
and protection_string != "PAGE_EXECUTE_WRITECOPY"):
if cls.is_vad_empty(proc_layer, vad):
continue
@@ -115,13 +113,14 @@ class Malfind(interfaces.plugins.PluginInterface):
def _generator(self, procs):
# determine if we're on a 32 or 64 bit kernel
is_32bit_arch = not symbols.symbol_table_is_64bit(self.context, self.config["nt_symbols"])
kernel = self.context.modules[self.config['kernel']]
is_32bit_arch = not symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name)
for proc in procs:
process_name = utility.array_to_string(proc.ImageFileName)
for vad, data in self.list_injections(self.context, self.config["primary"], self.config["nt_symbols"],
proc):
for vad, data in self.list_injections(self.context, kernel.layer_name, kernel.symbol_table_name, proc):
# 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():
@@ -145,13 +144,14 @@ class Malfind(interfaces.plugins.PluginInterface):
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(vad.get_start()),
format_hints.Hex(vad.get_end()), vad.get_tag(),
vad.get_protection(
vadinfo.VadInfo.protect_values(self.context, self.config["primary"],
self.config["nt_symbols"]),
vadinfo.VadInfo.protect_values(self.context, kernel.layer_name,
kernel.symbol_table_name),
vadinfo.winnt_protections), vad.get_commit_charge(), vad.get_private_memory(),
file_output, format_hints.HexBytes(data), disasm))
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("Start VPN", format_hints.Hex),
("End VPN", format_hints.Hex), ("Tag", str), ("Protection", str),
@@ -159,6 +159,6 @@ class Malfind(interfaces.plugins.PluginInterface):
("Hexdump", format_hints.HexBytes), ("Disasm", interfaces.renderers.Disassembly)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/memmap.py
+6 -7
View File
@@ -16,16 +16,14 @@ vollog = logging.getLogger(__name__)
class Memmap(interfaces.plugins.PluginInterface):
"""Prints the memory map"""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.BooleanRequirement(name = 'coalesce', description = 'Clump output where possible',
default = False, optional = True),
@@ -108,12 +106,13 @@ class Memmap(interfaces.plugins.PluginInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("Virtual", format_hints.Hex), ("Physical", format_hints.Hex),
("Size", format_hints.Hex), ("Offset in File", format_hints.Hex),
("File output", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/modscan.py
+7 -7
View File
@@ -17,16 +17,14 @@ vollog = logging.getLogger(__name__)
class ModScan(interfaces.plugins.PluginInterface):
"""Scans for modules present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'poolerscanner',
component = poolscanner.PoolScanner,
version = (1, 0, 0)),
@@ -137,14 +135,16 @@ class ModScan(interfaces.plugins.PluginInterface):
return None
def _generator(self):
session_layers = list(self.get_session_layers(self.context, self.config['primary'], self.config['nt_symbols']))
kernel = self.context.modules[self.config['kernel']]
session_layers = list(self.get_session_layers(self.context, kernel.layer_name, kernel.symbol_table_name))
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
"windows",
"pe",
class_types = pe.class_types)
for mod in self.scan_modules(self.context, self.config['primary'], self.config['nt_symbols']):
for mod in self.scan_modules(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
BaseDllName = mod.BaseDllName.get_string()
volatility3/framework/plugins/windows/modules.py
+5 -6
View File
@@ -19,16 +19,14 @@ vollog = logging.getLogger(__name__)
class Modules(interfaces.plugins.PluginInterface):
"""Lists the loaded kernel modules."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (2, 0, 0)),
requirements.BooleanRequirement(name = 'dump',
@@ -38,13 +36,14 @@ class Modules(interfaces.plugins.PluginInterface):
]
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
"windows",
"pe",
class_types = pe.class_types)
for mod in self.list_modules(self.context, self.config['primary'], self.config['nt_symbols']):
for mod in self.list_modules(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
BaseDllName = mod.BaseDllName.get_string()
volatility3/framework/plugins/windows/mutantscan.py
+6 -6
View File
@@ -13,15 +13,13 @@ from volatility3.plugins.windows import poolscanner
class MutantScan(interfaces.plugins.PluginInterface):
"""Scans for mutexes present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
]
@@ -50,7 +48,9 @@ class MutantScan(interfaces.plugins.PluginInterface):
yield mem_object
def _generator(self):
for mutant in self.scan_mutants(self.context, self.config['primary'], self.config['nt_symbols']):
kernel = self.context.modules[self.config['kernel']]
for mutant in self.scan_mutants(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
name = mutant.get_name()
volatility3/framework/plugins/windows/netscan.py
+9 -9
View File
@@ -22,16 +22,14 @@ vollog = logging.getLogger(__name__)
class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for network objects present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'poolscanner',
component = poolscanner.PoolScanner,
version = (1, 0, 0)),
@@ -279,11 +277,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def _generator(self, show_corrupt_results: Optional[bool] = None):
""" Generates the network objects for use in rendering. """
netscan_symbol_table = self.create_netscan_symbol_table(self.context, self.config["primary"],
self.config["nt_symbols"], self.config_path)
kernel = self.context.modules[self.config['kernel']]
for netw_obj in self.scan(self.context, self.config['primary'], self.config['nt_symbols'],
netscan_symbol_table):
netscan_symbol_table = self.create_netscan_symbol_table(self.context, kernel.layer_name,
kernel.symbol_table_name,
self.config_path)
for netw_obj in self.scan(self.context, kernel.layer_name, kernel.symbol_table_name, netscan_symbol_table):
vollog.debug(f"Found netw obj @ 0x{netw_obj.vol.offset:2x} of assumed type {type(netw_obj)}")
# objects passed pool header constraints. check for additional constraints if strict flag is set.
volatility3/framework/plugins/windows/netstat.py
+13 -11
View File
@@ -20,16 +20,14 @@ vollog = logging.getLogger(__name__)
class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Traverses network tracking structures present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'netscan', component = netscan.NetScan, version = (1, 0, 0)),
requirements.VersionRequirement(name = 'modules', component = modules.Modules, version = (1, 0, 0)),
requirements.VersionRequirement(name = 'pdbutil', component = pdbutil.PDBUtility, version = (1, 0, 0)),
@@ -419,19 +417,23 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def _generator(self, show_corrupt_results: Optional[bool] = None):
""" Generates the network objects for use in rendering. """
netscan_symbol_table = netscan.NetScan.create_netscan_symbol_table(self.context, self.config["primary"],
self.config["nt_symbols"], self.config_path)
kernel = self.context.modules[self.config['kernel']]
tcpip_module = self.get_tcpip_module(self.context, self.config["primary"], self.config["nt_symbols"])
netscan_symbol_table = netscan.NetScan.create_netscan_symbol_table(self.context,
kernel.layer_name,
kernel.symbol_table_name,
self.config_path)
tcpip_module = self.get_tcpip_module(self.context, kernel.layer_name, kernel.symbol_table_name)
try:
tcpip_symbol_table = pdbutil.PDBUtility.symbol_table_from_pdb(
self.context, interfaces.configuration.path_join(self.config_path, 'tcpip'), self.config["primary"],
"tcpip.pdb", tcpip_module.DllBase, tcpip_module.SizeOfImage)
self.context, interfaces.configuration.path_join(self.config_path, 'tcpip'),
kernel.layer_name, "tcpip.pdb", tcpip_module.DllBase, tcpip_module.SizeOfImage)
except exceptions.VolatilityException:
vollog.warning("Unable to locate symbols for the memory image's tcpip module")
for netw_obj in self.list_sockets(self.context, self.config['primary'], self.config['nt_symbols'],
for netw_obj in self.list_sockets(self.context, kernel.layer_name, kernel.symbol_table_name,
netscan_symbol_table, tcpip_module.DllBase, tcpip_symbol_table):
# objects passed pool header constraints. check for additional constraints if strict flag is set.
volatility3/framework/plugins/windows/poolscanner.py
+7 -7
View File
@@ -112,25 +112,25 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface):
class PoolScanner(plugins.PluginInterface):
"""A generic pool scanner plugin."""
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'handles', plugin = handles.Handles, version = (1, 0, 0)),
]
def _generator(self):
symbol_table = self.config["nt_symbols"]
kernel = self.context.modules[self.config['kernel']]
symbol_table = kernel.symbol_table_name
constraints = self.builtin_constraints(symbol_table)
for constraint, mem_object, header in self.generate_pool_scan(self.context, self.config["primary"],
for constraint, mem_object, header in self.generate_pool_scan(self.context, kernel.layer_name,
symbol_table, constraints):
# generate some type-specific info for sanity checking
if constraint.object_type == "Process":
volatility3/framework/plugins/windows/privileges.py
+6 -7
View File
@@ -16,7 +16,7 @@ vollog = logging.getLogger(__name__)
class Privs(interfaces.plugins.PluginInterface):
"""Lists process token privileges"""
_version = (1, 0, 0)
_version = (1, 2, 0)
_required_framework_version = (1, 0, 0)
def __init__(self, *args, **kwargs):
@@ -40,10 +40,8 @@ class Privs(interfaces.plugins.PluginInterface):
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.ListRequirement(name = 'pid',
description = 'Filter on specific process IDs',
element_type = int,
@@ -89,11 +87,12 @@ class Privs(interfaces.plugins.PluginInterface):
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("PID", int), ("Process", str), ("Value", int), ("Privilege", str),
("Attributes", str), ("Description", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/pslist.py
+15 -10
View File
@@ -80,7 +80,8 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
return file_handle
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None, exclude: bool = False) -> Callable[[interfaces.objects.ObjectInterface], bool]:
def create_pid_filter(cls, pid_list: List[int] = None, exclude: bool = False) -> Callable[
[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process IDs.
@@ -103,7 +104,8 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
return filter_func
@classmethod
def create_name_filter(cls, name_list: List[str] = None, exclude: bool = False) -> Callable[[interfaces.objects.ObjectInterface], bool]:
def create_name_filter(cls, name_list: List[str] = None, exclude: bool = False) -> Callable[
[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process names.
@@ -170,19 +172,21 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
yield proc
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
"windows",
"pe",
class_types = pe.class_types)
memory = self.context.layers[self.config['kernel.layer_name']]
memory = self.context.layers[kernel.layer_name]
if not isinstance(memory, layers.intel.Intel):
raise TypeError("Primary layer is not an intel layer")
for proc in self.list_processes(self.context,
self.config['kernel.layer_name'],
self.config['kernel.symbol_table_name'],
kernel.layer_name,
kernel.symbol_table_name,
filter_func = self.create_pid_filter(self.config.get('pid', None))):
if not self.config.get('physical', self.PHYSICAL_DEFAULT):
@@ -194,7 +198,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
try:
if self.config['dump']:
file_handle = self.process_dump(self.context, self.config['kernel.symbol_table_name'],
file_handle = self.process_dump(self.context, kernel.symbol_table_name,
pe_table_name, proc, self.open)
file_output = "Error outputting file"
if file_handle:
@@ -202,12 +206,13 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_output = str(file_handle.preferred_filename)
yield (0, (proc.UniqueProcessId, proc.InheritedFromUniqueProcessId,
proc.ImageFileName.cast("string", max_length = proc.ImageFileName.vol.count, errors = 'replace'),
format_hints.Hex(offset), proc.ActiveThreads, proc.get_handle_count(), proc.get_session_id(),
proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time(), file_output))
proc.ImageFileName.cast("string", max_length = proc.ImageFileName.vol.count,
errors = 'replace'),
format_hints.Hex(offset), proc.ActiveThreads, proc.get_handle_count(), proc.get_session_id(),
proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time(), file_output))
except exceptions.InvalidAddressException:
vollog.info(f"Invalid process found at address: {proc.vol.offset:x}. Skipping")
vollog.info(f"Invalid process found at address: {proc.vol.offset:x}. Skipping")
def generate_timeline(self):
for row in self._generator():
volatility3/framework/plugins/windows/psscan.py
+12 -11
View File
@@ -22,16 +22,14 @@ vollog = logging.getLogger(__name__)
class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for processes present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'info', component = info.Info, version = (1, 0, 0)),
requirements.ListRequirement(name = 'pid',
@@ -144,26 +142,29 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
return (nt_major_version, nt_minor_version, build)
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
"windows",
"pe",
class_types = pe.class_types)
for proc in self.scan_processes(self.context,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))):
file_output = "Disabled"
if self.config['dump']:
# windows 10 objects (maybe others in the future) are already in virtual memory
if proc.vol.layer_name == self.config['primary']:
if proc.vol.layer_name == kernel.layer_name:
vproc = proc
else:
vproc = self.virtual_process_from_physical(self.context, self.config['primary'],
self.config['nt_symbols'], proc)
vproc = self.virtual_process_from_physical(self.context, kernel.layer_name,
kernel.symbol_table_name, proc)
file_handle = pslist.PsList.process_dump(self.context, self.config['nt_symbols'], pe_table_name, vproc,
file_handle = pslist.PsList.process_dump(self.context, kernel.symbol_table_name,
pe_table_name, vproc,
self.open)
file_output = "Error outputting file"
if file_handle:
volatility3/framework/plugins/windows/pstree.py
+8 -7
View File
@@ -14,7 +14,7 @@ class PsTree(interfaces.plugins.PluginInterface):
"""Plugin for listing processes in a tree based on their parent process
ID."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
@@ -25,10 +25,8 @@ class PsTree(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.BooleanRequirement(name = 'physical',
description = 'Display physical offsets instead of virtual',
default = pslist.PsList.PHYSICAL_DEFAULT,
@@ -56,12 +54,15 @@ class PsTree(interfaces.plugins.PluginInterface):
def _generator(self):
"""Generates the Tree of processes."""
for proc in pslist.PsList.list_processes(self.context, self.config['primary'], self.config['nt_symbols']):
kernel = self.context.modules[self.config['kernel']]
for proc in pslist.PsList.list_processes(self.context, kernel.layer_name,
kernel.symbol_table_name):
if not self.config.get('physical', pslist.PsList.PHYSICAL_DEFAULT):
offset = proc.vol.offset
else:
layer_name = self.config['primary']
layer_name = kernel.layer_name
memory = self.context.layers[layer_name]
(_, _, offset, _, _) = list(memory.mapping(offset = proc.vol.offset, length = 0))[0]
volatility3/framework/plugins/windows/registry/hivelist.py
+9 -9
View File
@@ -39,16 +39,14 @@ class HiveGenerator:
class HiveList(interfaces.plugins.PluginInterface):
"""Lists the registry hives present in a particular memory image."""
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.StringRequirement(name = 'filter',
description = "String to filter hive names returned",
optional = True,
@@ -66,9 +64,11 @@ class HiveList(interfaces.plugins.PluginInterface):
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]:
chunk_size = 0x500000
kernel = self.context.modules[self.config['kernel']]
for hive_object in self.list_hive_objects(context = self.context,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_string = self.config.get('filter', None)):
file_output = "Disabled"
@@ -77,8 +77,8 @@ class HiveList(interfaces.plugins.PluginInterface):
hive = next(
self.list_hives(self.context,
self.config_path,
layer_name = self.config["primary"],
symbol_table = self.config["nt_symbols"],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
hive_offsets = [hive_object.vol.offset]))
maxaddr = hive.hive.Storage[0].Length
hive_name = self._sanitize_hive_name(hive.get_name())
volatility3/framework/plugins/windows/registry/hivescan.py
+8 -7
View File
@@ -15,16 +15,14 @@ class HiveScan(interfaces.plugins.PluginInterface):
"""Scans for registry hives present in a particular windows memory
image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'bigpools', plugin = bigpools.BigPools, version = (1, 0, 0)),
]
@@ -68,9 +66,12 @@ class HiveScan(interfaces.plugins.PluginInterface):
yield mem_object
def _generator(self):
for hive in self.scan_hives(self.context, self.config['primary'], self.config['nt_symbols']):
yield (0, (format_hints.Hex(hive.vol.offset), ))
kernel = self.context.modules[self.config['kernel']]
for hive in self.scan_hives(self.context, kernel.layer_name, kernel.symbol_table_name):
yield (0, (format_hints.Hex(hive.vol.offset),))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex)], self._generator())
volatility3/framework/plugins/windows/registry/printkey.py
+10 -11
View File
@@ -19,16 +19,14 @@ vollog = logging.getLogger(__name__)
class PrintKey(interfaces.plugins.PluginInterface):
"""Lists the registry keys under a hive or specific key value."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)),
requirements.IntRequirement(name = 'offset', description = "Hive Offset", default = None, optional = True),
requirements.StringRequirement(name = 'key',
@@ -43,10 +41,10 @@ class PrintKey(interfaces.plugins.PluginInterface):
@classmethod
def key_iterator(
cls,
hive: RegistryHive,
node_path: Sequence[objects.StructType] = None,
recurse: bool = False
cls,
hive: RegistryHive,
node_path: Sequence[objects.StructType] = None,
recurse: bool = False
) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, interfaces.objects.ObjectInterface]]:
"""Walks through a set of nodes from a given node (last one in
node_path). Avoids loops by not traversing into nodes already present
@@ -188,12 +186,13 @@ class PrintKey(interfaces.plugins.PluginInterface):
def run(self):
offset = self.config.get('offset', None)
kernel = self.context.modules[self.config['kernel']]
return TreeGrid(columns = [('Last Write Time', datetime.datetime), ('Hive Offset', format_hints.Hex),
('Type', str), ('Key', str), ('Name', str), ('Data', format_hints.MultiTypeData),
('Volatile', bool)],
generator = self._registry_walker(self.config['primary'],
self.config['nt_symbols'],
generator = self._registry_walker(kernel.layer_name,
kernel.symbol_table_name,
hive_offsets = None if offset is None else [offset],
key = self.config.get('key', None),
recurse = self.config.get('recurse', None)))
volatility3/framework/plugins/windows/registry/userassist.py
+13 -9
View File
@@ -23,7 +23,7 @@ vollog = logging.getLogger(__name__)
class UserAssist(interfaces.plugins.PluginInterface):
"""Print userassist registry keys and information."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -37,10 +37,8 @@ class UserAssist(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.IntRequirement(name = 'offset', description = "Hive Offset", default = None, optional = True),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0))
]
@@ -115,13 +113,17 @@ class UserAssist(interfaces.plugins.PluginInterface):
def _win7_or_later(self) -> bool:
# TODO: change this if there is a better way of determining the OS version
# _KUSER_SHARED_DATA.CookiePad is in Windows 6.1 (Win7) and later
return self.context.symbol_space.get_type(self.config['nt_symbols'] + constants.BANG +
kernel = self.context.modules[self.config['kernel']]
return self.context.symbol_space.get_type(kernel.symbol_table_name + constants.BANG +
"_KUSER_SHARED_DATA").has_member('CookiePad')
def list_userassist(self, hive: RegistryHive) -> Generator[Tuple[int, Tuple], None, None]:
"""Generate userassist data for a registry hive."""
hive_name = hive.hive.cast(self.config["nt_symbols"] + constants.BANG + "_CMHIVE").get_name()
kernel = self.context.modules[self.config['kernel']]
hive_name = hive.hive.cast(kernel.symbol_table_name + constants.BANG + "_CMHIVE").get_name()
if self._win7 is None:
try:
@@ -216,11 +218,13 @@ class UserAssist(interfaces.plugins.PluginInterface):
if self.config.get('offset', None) is not None:
hive_offsets = [self.config.get('offset', None)]
kernel = self.context.modules[self.config['kernel']]
# get all the user hive offsets or use the one specified
for hive in hivelist.HiveList.list_hives(context = self.context,
base_config_path = self.config_path,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_string = 'ntuser.dat',
hive_offsets = hive_offsets):
try:
volatility3/framework/plugins/windows/skeleton_key_check.py
+123 -118
View File
@@ -6,50 +6,49 @@
# It does this by locating the CSystems array through a variety of methods,
# and then validating the entry for RC4 HMAC (0x17 / 23)
#
# For a thorough walkthrough on how the R&D was performed to develop this plugin,
# For a thorough walkthrough on how the R&D was performed to develop this plugin,
# please see our blogpost here:
#
# <insert blog URL once published>
import logging, io
import io
import logging
from typing import Iterable, Tuple, List, Optional
from volatility3.framework.symbols.windows import pdbutil
from volatility3.framework import interfaces, symbols, exceptions
from volatility3.framework import renderers, constants
from volatility3.framework.layers import scanners
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.framework.symbols import intermed
from volatility3.framework.renderers import format_hints
from volatility3.plugins.windows import pslist, vadinfo
from volatility3.framework.symbols.windows.extensions import pe
import pefile
from volatility3.framework import interfaces, symbols, exceptions
from volatility3.framework import renderers, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import scanners
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import pdbutil
from volatility3.framework.symbols.windows.extensions import pe
from volatility3.plugins.windows import pslist, vadinfo
try:
import capstone
has_capstone = True
except ImportError:
has_capstone = False
vollog = logging.getLogger(__name__)
class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
""" Looks for signs of Skeleton Key malware """
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'vadinfo', component = vadinfo.VadInfo, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'pdbutil', component = pdbutil.PDBUtility, version = (1, 0, 0)),
@@ -71,26 +70,26 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
try:
dos_header = self.context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset = base_address,
layer_name = layer_name)
offset = base_address,
layer_name = layer_name)
for offset, data in dos_header.reconstruct():
pe_data.seek(offset)
pe_data.write(data)
pe_ret = pefile.PE(data = pe_data.getvalue(), fast_load = True)
except exceptions.InvalidAddressException:
vollog.debug("Unable to reconstruct cryptdll.dll in memory")
pe_ret = None
return pe_ret
def _check_for_skeleton_key_vad(self, csystem: interfaces.objects.ObjectInterface,
cryptdll_base: int,
cryptdll_size: int) -> bool:
def _check_for_skeleton_key_vad(self, csystem: interfaces.objects.ObjectInterface,
cryptdll_base: int,
cryptdll_size: int) -> bool:
"""
Checks if Initialize and/or Decrypt is hooked by determining if
Checks if Initialize and/or Decrypt is hooked by determining if
these function pointers reference addresses inside of the cryptdll VAD
Args:
@@ -101,11 +100,11 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
bool: if a skeleton key hook is present
"""
return not ((cryptdll_base <= csystem.Initialize <= cryptdll_base + cryptdll_size) and \
(cryptdll_base <= csystem.Decrypt <= cryptdll_base + cryptdll_size))
(cryptdll_base <= csystem.Decrypt <= cryptdll_base + cryptdll_size))
def _check_for_skeleton_key_symbols(self, csystem: interfaces.objects.ObjectInterface,
rc4HmacInitialize: int,
rc4HmacDecrypt: int) -> bool:
def _check_for_skeleton_key_symbols(self, csystem: interfaces.objects.ObjectInterface,
rc4HmacInitialize: int,
rc4HmacDecrypt: int) -> bool:
"""
Uses the PDB information to specifically check if the csystem for RC4HMAC
has an initialization pointer to rc4HmacInitialize and a decryption pointer
@@ -113,12 +112,12 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
Args:
csystem: The RC4HMAC KERB_ECRYPT instance
rc4HmacInitialize: The expected address of csystem Initialization function
rc4HmacInitialize: The expected address of csystem Initialization function
rc4HmacDecrypt: The expected address of the csystem Decryption function
Returns:
bool: if a skeleton key hook was found
"""
"""
return csystem.Initialize != rc4HmacInitialize or csystem.Decrypt != rc4HmacDecrypt
def _construct_ecrypt_array(self, array_start: int, count: int, \
@@ -137,21 +136,21 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
try:
array = cryptdll_types.object(object_type = "array",
offset = array_start,
subtype = cryptdll_types.get_type("_KERB_ECRYPT"),
count = count,
absolute = True)
offset = array_start,
subtype = cryptdll_types.get_type("_KERB_ECRYPT"),
count = count,
absolute = True)
except exceptions.InvalidAddressException:
vollog.debug("Unable to construct cSystems array at given offset: {:x}".format(array_start))
array = None
return array
def _find_array_with_pdb_symbols(self, cryptdll_symbols: str,
cryptdll_types: interfaces.context.ModuleInterface,
proc_layer_name: str,
cryptdll_base: int) -> Tuple[interfaces.objects.ObjectInterface, int, int, int]:
def _find_array_with_pdb_symbols(self, cryptdll_symbols: str,
cryptdll_types: interfaces.context.ModuleInterface,
proc_layer_name: str,
cryptdll_base: int) -> Tuple[interfaces.objects.ObjectInterface, int, int, int]:
"""
Finds the CSystems array through use of PDB symbols
@@ -177,7 +176,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
count_address = cryptdll_module.get_symbol("cCSystems").address
# we do not want to fail just because the count is not in memory
# 16 was the size on samples I tested, so I chose it as the default
# 16 was the size on samples I tested, so I chose it as the default
try:
count = cryptdll_types.object(object_type = "unsigned long", offset = count_address)
except exceptions.InvalidAddressException:
@@ -186,30 +185,31 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
array_start = cryptdll_module.get_absolute_symbol_address("CSystems")
array = self._construct_ecrypt_array(array_start, count, cryptdll_types)
if array is None:
vollog.debug("The CSystem array is not present in memory. Stopping PDB based analysis.")
return array, rc4HmacInitialize, rc4HmacDecrypt
def _get_cryptdll_types(self, context: interfaces.context.ContextInterface,
config,
config_path: str,
proc_layer_name: str,
cryptdll_base: int):
def _get_cryptdll_types(self, context: interfaces.context.ContextInterface,
config,
config_path: str,
proc_layer_name: str,
cryptdll_base: int):
"""
Builds a symbol table from the cryptdll types generated after binary analysis
Args:
context: the context to operate upon
config:
config:
config_path:
proc_layer_name: name of the lsass.exe process layer
cryptdll_base: base address of cryptdll.dll inside of lsass.exe
"""
table_mapping = {"nt_symbols": config["nt_symbols"]}
kernel = self.context.modules[self.config['kernel']]
table_mapping = {"nt_symbols": kernel.symbol_table_name}
cryptdll_symbol_table = intermed.IntermediateSymbolTable.create(context = context,
cryptdll_symbol_table = intermed.IntermediateSymbolTable.create(context = context,
config_path = config_path,
sub_path = "windows",
filename = "kerb_ecrypt",
@@ -218,7 +218,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
return context.module(cryptdll_symbol_table, proc_layer_name, offset = cryptdll_base)
def _find_lsass_proc(self, proc_list: Iterable) -> \
Tuple[interfaces.context.ContextInterface, str]:
Tuple[interfaces.context.ContextInterface, str]:
"""
Walks the process list and returns the first valid lsass instances.
There should be only one lsass process, but malware will often use the
@@ -242,11 +242,10 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
vollog.debug("Process {}: invalid address {} in layer {}".format(proc_id, excp.invalid_address,
excp.layer_name))
return None, None
def _find_cryptdll(self, lsass_proc: interfaces.context.ContextInterface) -> \
Tuple[int, int]:
Tuple[int, int]:
"""
Finds the base address of cryptdll.dll inside of lsass.exe
@@ -260,18 +259,18 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
"""
for vad in lsass_proc.get_vad_root().traverse():
filename = vad.get_file_name()
if isinstance(filename, str) and filename.lower().endswith("cryptdll.dll"):
base = vad.get_start()
return base, vad.get_end() - base
return None, None
def _find_csystems_with_symbols(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
cryptdll_size: int) -> \
Tuple[interfaces.objects.ObjectInterface, int, int]:
def _find_csystems_with_symbols(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
cryptdll_size: int) -> \
Tuple[interfaces.objects.ObjectInterface, int, int]:
"""
Attempts to find CSystems and the expected address of the handlers.
Relies on downloading and parsing of the cryptdll PDB file.
@@ -281,27 +280,28 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
cryptdll_types: The types from cryptdll binary analysis
cryptdll_base: the base address of cryptdll.dll
crytpdll_size: the size of the VAD for cryptdll.dll
Returns:
A tuple of:
array: An initialized Volatility array of _KERB_ECRYPT structures
rc4HmacInitialize: The expected address of csystem Initialization function
rc4HmacInitialize: The expected address of csystem Initialization function
rc4HmacDecrypt: The expected address of the csystem Decryption function
"""
try:
cryptdll_symbols = pdbutil.PDBUtility.symbol_table_from_pdb(self.context,
interfaces.configuration.path_join(self.config_path, 'cryptdll'),
proc_layer_name,
"cryptdll.pdb",
cryptdll_base,
cryptdll_size)
cryptdll_symbols = pdbutil.PDBUtility.symbol_table_from_pdb(self.context,
interfaces.configuration.path_join(
self.config_path, 'cryptdll'),
proc_layer_name,
"cryptdll.pdb",
cryptdll_base,
cryptdll_size)
except exceptions.VolatilityException:
vollog.debug("Unable to use the cryptdll PDB. Stopping PDB symbols based analysis.")
return None, None, None
array, rc4HmacInitialize, rc4HmacDecrypt = \
self._find_array_with_pdb_symbols(cryptdll_symbols, cryptdll_types, proc_layer_name, cryptdll_base)
self._find_array_with_pdb_symbols(cryptdll_symbols, cryptdll_types, proc_layer_name, cryptdll_base)
if array is None:
vollog.debug("The CSystem array is not present in memory. Stopping PDB symbols based analysis.")
@@ -313,7 +313,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
These instructions contain the offset of a target address
relative to the current instruction pointer.
Args:
inst: A capstone instruction instance
@@ -322,7 +322,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
"""
try:
opnd = inst.operands[1]
except capstone.CsError:
except capstone.CsError:
return None
if opnd.type != capstone.x86.X86_OP_MEM:
@@ -334,9 +334,9 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
return inst.address + inst.size + opnd.mem.disp
def _analyze_cdlocatecsystem(self, function_bytes: bytes,
function_start: int,
cryptdll_types: interfaces.context.ModuleInterface,
proc_layer_name: str) -> Optional[interfaces.objects.ObjectInterface]:
function_start: int,
cryptdll_types: interfaces.context.ModuleInterface,
proc_layer_name: str) -> Optional[interfaces.objects.ObjectInterface]:
"""
Performs static analysis on CDLocateCSystem to find the instructions that
reference CSystems as well as cCsystems
@@ -380,7 +380,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
target_address = self._get_rip_relative_target(inst)
if target_address:
array_start = target_address
array_start = target_address
# we find the count before, so we can terminate the static analysis here
break
@@ -392,10 +392,10 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
return array
def _find_csystems_with_export(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
_) -> Optional[interfaces.objects.ObjectInterface]:
def _find_csystems_with_export(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
_) -> Optional[interfaces.objects.ObjectInterface]:
"""
Uses export table analysis to locate CDLocateCsystem
This function references CSystems and cCsystems
@@ -420,8 +420,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
"windows",
"pe",
class_types = pe.class_types)
cryptdll = self._get_pefile_obj(pe_table_name, proc_layer_name, cryptdll_base)
if not cryptdll:
return None
@@ -440,7 +439,8 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
try:
function_bytes = self.context.layers[proc_layer_name].read(function_start, 0x50)
except exceptions.InvalidAddressException:
vollog.debug("The CDLocateCSystem function is not present in the lsass address space. Stopping export based analysis.")
vollog.debug(
"The CDLocateCSystem function is not present in the lsass address space. Stopping export based analysis.")
break
array = self._analyze_cdlocatecsystem(function_bytes, function_start, cryptdll_types, proc_layer_name)
@@ -451,15 +451,15 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
return None
def _find_csystems_with_scanning(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
cryptdll_size: int) -> List[interfaces.context.ModuleInterface]:
def _find_csystems_with_scanning(self, proc_layer_name: str,
cryptdll_types: interfaces.context.ModuleInterface,
cryptdll_base: int,
cryptdll_size: int) -> List[interfaces.context.ModuleInterface]:
"""
Performs scanning to find potential RC4 HMAC csystem instances
This function may return several values as it cannot validate which is the active one
Args:
proc_layer_name: the lsass.exe process layer name
cryptdll_types: the types from cryptdll binary analysis
@@ -468,22 +468,22 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
Returns:
A list of csystem instances
"""
csystems = []
cryptdll_end = cryptdll_base + cryptdll_size
proc_layer = self.context.layers[proc_layer_name]
ecrypt_size = cryptdll_types.get_type("_KERB_ECRYPT").size
# scan for potential instances of RC4 HMAC
# the signature is based on the type being 0x17
# and the block size member being 1 in all test samples
# and the block size member being 1 in all test samples
for address in proc_layer.scan(self.context,
scanners.BytesScanner(b"\x17\x00\x00\x00\x01\x00\x00\x00"),
sections = [(cryptdll_base, cryptdll_size)]):
# this occurs across page boundaries
if not proc_layer.is_valid(address, ecrypt_size):
continue
@@ -491,11 +491,11 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
kerb = cryptdll_types.object("_KERB_ECRYPT",
offset = address,
absolute = True)
# ensure the Encrypt and Finish pointers are inside the VAD
# these are not manipulated in the attack
# these are not manipulated in the attack
if (cryptdll_base < kerb.Encrypt < cryptdll_end) and \
(cryptdll_base < kerb.Finish < cryptdll_end):
(cryptdll_base < kerb.Finish < cryptdll_end):
csystems.append(kerb)
return csystems
@@ -509,35 +509,36 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
Args:
procs: the process list filtered to lsass.exe instances
"""
if not symbols.symbol_table_is_64bit(self.context, self.config["nt_symbols"]):
kernel = self.context.modules[self.config['kernel']]
if not symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name):
vollog.info("This plugin only supports 64bit Windows memory samples")
return
lsass_proc, proc_layer_name = self._find_lsass_proc(procs)
if not lsass_proc:
vollog.info("Unable to find a valid lsass.exe process in the process list. This should never happen. Analysis cannot proceed.")
vollog.info(
"Unable to find a valid lsass.exe process in the process list. This should never happen. Analysis cannot proceed.")
return
cryptdll_base, cryptdll_size = self._find_cryptdll(lsass_proc)
if not cryptdll_base:
vollog.info("Unable to find the location of cryptdll.dll inside of lsass.exe. Analysis cannot proceed.")
return
# the custom type information from binary analysis
cryptdll_types = self._get_cryptdll_types(self.context,
self.config,
cryptdll_types = self._get_cryptdll_types(self.context,
self.config,
self.config_path,
proc_layer_name,
cryptdll_base)
# attempt to find the array and symbols directly from the PDB
csystems, rc4HmacInitialize, rc4HmacDecrypt = \
self._find_csystems_with_symbols(proc_layer_name,
cryptdll_types,
cryptdll_base,
cryptdll_size)
self._find_csystems_with_symbols(proc_layer_name,
cryptdll_types,
cryptdll_base,
cryptdll_size)
csystems = None
@@ -550,7 +551,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
self._find_csystems_with_scanning]
for source in fallback_sources:
csystems = source(proc_layer_name,
csystems = source(proc_layer_name,
cryptdll_types,
cryptdll_base,
cryptdll_size)
@@ -587,13 +588,17 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
named processes to blend in or uses lsass.exe as a process hollowing target
"""
process_name = utility.array_to_string(proc.ImageFileName)
return process_name != "lsass.exe"
def run(self):
return renderers.TreeGrid([("PID", int), ("Process", str), ("Skeleton Key Found", bool), ("rc4HmacInitialize", format_hints.Hex), ("rc4HmacDecrypt", format_hints.Hex)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
filter_func = self._lsass_proc_filter)))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid(
[("PID", int), ("Process", str), ("Skeleton Key Found", bool), ("rc4HmacInitialize", format_hints.Hex),
("rc4HmacDecrypt", 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 = self._lsass_proc_filter)))
volatility3/framework/plugins/windows/ssdt.py
+9 -9
View File
@@ -18,16 +18,14 @@ from volatility3.plugins.windows import modules
class SSDT(plugins.PluginInterface):
"""Lists the system call table."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'modules', plugin = modules.Modules, version = (1, 0, 0)),
]
@@ -75,11 +73,13 @@ class SSDT(plugins.PluginInterface):
def _generator(self) -> Iterator[Tuple[int, Tuple[int, int, Any, Any]]]:
layer_name = self.config['primary']
collection = self.build_module_collection(self.context, self.config["primary"], self.config["nt_symbols"])
kernel = self.context.modules[self.config['kernel']]
layer_name = kernel.layer_name
collection = self.build_module_collection(self.context, layer_name, kernel.symbol_table_name)
kvo = self.context.layers[layer_name].config['kernel_virtual_offset']
ntkrnlmp = self.context.module(self.config["nt_symbols"], layer_name = layer_name, offset = kvo)
ntkrnlmp = self.context.module(kernel.symbol_table_name, layer_name = layer_name, offset = kvo)
# this is just one way to enumerate the native (NT) service table.
# to do the same thing for the Win32K service table, we would need Win32K.sys symbol support
@@ -91,7 +91,7 @@ class SSDT(plugins.PluginInterface):
# on 32-bit systems the table indexes are 32-bits and contain pointers (unsigned)
# on 64-bit systems the indexes are also 32-bits but they're offsets from the
# base address of the table and can be negative, so we need a signed data type
is_kernel_64 = symbols.symbol_table_is_64bit(self.context, self.config["nt_symbols"])
is_kernel_64 = symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name)
if is_kernel_64:
array_subtype = "long"
volatility3/framework/plugins/windows/strings.py
+9 -9
View File
@@ -18,18 +18,16 @@ vollog = logging.getLogger(__name__)
class Strings(interfaces.plugins.PluginInterface):
"""Reads output from the strings command and indicates which process(es) each string belongs to."""
_version = (1, 0, 0)
_version = (1, 2, 0)
_required_framework_version = (1, 0, 0)
strings_pattern = re.compile(rb"^(?:\W*)([0-9]+)(?:\W*)(\w[\w\W]+)\n?")
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process ID to include (all other processes are excluded)",
@@ -44,7 +42,7 @@ class Strings(interfaces.plugins.PluginInterface):
def _generator(self) -> Generator[Tuple, None, None]:
"""Generates results from a strings file."""
string_list: List[Tuple[int,bytes]] = []
string_list: List[Tuple[int, bytes]] = []
# Test strings file format is accurate
accessor = resources.ResourceAccessor()
@@ -60,14 +58,16 @@ class Strings(interfaces.plugins.PluginInterface):
vollog.error(f"Line in unrecognized format: line {count}")
line = strings_fp.readline()
kernel = self.context.modules[self.config['kernel']]
revmap = self.generate_mapping(self.context,
self.config['primary'],
self.config['nt_symbols'],
kernel.layer_name,
kernel.symbol_table_name,
progress_callback = self._progress_callback,
pid_list = self.config['pid'])
last_prog: float = 0
line_count: float = 0
line_count: float = 0
num_strings = len(string_list)
for offset, string in string_list:
line_count += 1
volatility3/framework/plugins/windows/svcscan.py
+10 -9
View File
@@ -21,17 +21,15 @@ vollog = logging.getLogger(__name__)
class SvcScan(interfaces.plugins.PluginInterface):
"""Scans for windows services."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
requirements.PluginRequirement(name = 'vadyarascan', plugin = vadyarascan.VadYaraScan, version = (1, 0, 0))
@@ -94,15 +92,18 @@ class SvcScan(interfaces.plugins.PluginInterface):
native_types = native_types)
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
service_table_name = self.create_service_table(self.context, self.config["nt_symbols"], self.config_path)
service_table_name = self.create_service_table(self.context, kernel.symbol_table_name,
self.config_path)
relative_tag_offset = self.context.symbol_space.get_type(service_table_name + constants.BANG +
"_SERVICE_RECORD").relative_child_offset("Tag")
filter_func = pslist.PsList.create_name_filter(["services.exe"])
is_vista_or_later = versions.is_vista_or_later(context = self.context, symbol_table = self.config["nt_symbols"])
is_vista_or_later = versions.is_vista_or_later(context = self.context,
symbol_table = kernel.symbol_table_name)
if is_vista_or_later:
service_tag = b"serH"
@@ -112,8 +113,8 @@ class SvcScan(interfaces.plugins.PluginInterface):
seen = []
for task in pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func):
proc_id = "Unknown"
volatility3/framework/plugins/windows/symlinkscan.py
+6 -6
View File
@@ -15,15 +15,13 @@ from volatility3.plugins.windows import poolscanner
class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for links present in a particular windows memory image."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
]
@classmethod
@@ -51,7 +49,9 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
yield mem_object
def _generator(self):
for link in self.scan_symlinks(self.context, self.config['primary'], self.config['nt_symbols']):
kernel = self.context.modules[self.config['kernel']]
for link in self.scan_symlinks(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
from_name = link.get_link_name()
volatility3/framework/plugins/windows/vadinfo.py
+10 -8
View File
@@ -33,7 +33,7 @@ winnt_protections = {
class VadInfo(interfaces.plugins.PluginInterface):
"""Lists process memory ranges."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (2, 0, 0)
MAXSIZE_DEFAULT = 0
@@ -44,10 +44,8 @@ class VadInfo(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
return [requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
# TODO: Convert this to a ListRequirement so that people can filter on sets of ranges
requirements.IntRequirement(name = 'address',
description = "Process virtual memory address to include " \
@@ -169,6 +167,8 @@ class VadInfo(interfaces.plugins.PluginInterface):
def _generator(self, procs):
kernel = self.context.modules[self.config['kernel']]
def passthrough(_: interfaces.objects.ObjectInterface) -> bool:
return False
@@ -196,12 +196,14 @@ class VadInfo(interfaces.plugins.PluginInterface):
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(vad.vol.offset),
format_hints.Hex(vad.get_start()), format_hints.Hex(vad.get_end()), vad.get_tag(),
vad.get_protection(
self.protect_values(self.context, self.config['primary'], self.config['nt_symbols']),
self.protect_values(self.context, kernel.layer_name, kernel.symbol_table_name),
winnt_protections), vad.get_commit_charge(), vad.get_private_memory(),
format_hints.Hex(vad.get_parent()), vad.get_file_name(), file_output))
def run(self):
kernel = self.context.modules[self.config['kernel']]
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
return renderers.TreeGrid([("PID", int), ("Process", str), ("Offset", format_hints.Hex),
@@ -210,6 +212,6 @@ class VadInfo(interfaces.plugins.PluginInterface):
("Parent", format_hints.Hex), ("File", str), ("File output", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
volatility3/framework/plugins/windows/vadyarascan.py
+7 -7
View File
@@ -17,16 +17,14 @@ vollog = logging.getLogger(__name__)
class VadYaraScan(interfaces.plugins.PluginInterface):
"""Scans all the Virtual Address Descriptor memory maps using yara."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = "Memory layer for the kernel",
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.BooleanRequirement(name = "wide",
description = "Match wide (unicode) strings",
default = False,
@@ -55,13 +53,15 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
rules = yarascan.YaraScan.process_yara_options(dict(self.config))
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
for task in pslist.PsList.list_processes(context = self.context,
layer_name = self.config['primary'],
symbol_table = self.config['nt_symbols'],
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func):
layer_name = task.add_process_layer()
layer = self.context.layers[layer_name]
volatility3/framework/plugins/windows/verinfo.py
+10 -10
View File
@@ -27,21 +27,19 @@ except ImportError:
class VerInfo(interfaces.plugins.PluginInterface):
"""Lists version information from PE files."""
_required_framework_version = (1, 2, 0)
_version = (1, 0, 0)
_required_framework_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
## TODO: we might add a regex option on the name later, but otherwise we're good
## TODO: and we don't want any CLI options from pslist, modules, or moddump
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.PluginRequirement(name = 'modules', plugin = modules.Modules, version = (1, 0, 0)),
requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (2, 0, 0)),
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
requirements.BooleanRequirement(name = "extensive",
description = "Search physical layer for version information",
optional = True,
@@ -123,6 +121,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
mods: <generator> of modules
session_layers: <generator> of layers in the session to be checked
"""
kernel = self.context.modules[self.config['kernel']]
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
@@ -131,7 +130,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
class_types = pe.class_types)
# TODO: Fix this so it works with more than just intel layers
physical_layer_name = self.context.layers[self.config['primary']].config.get('memory_layer', None)
physical_layer_name = self.context.layers[kernel.layer_name].config.get('memory_layer', None)
for mod in mods:
try:
@@ -191,13 +190,14 @@ class VerInfo(interfaces.plugins.PluginInterface):
build))
def run(self):
procs = pslist.PsList.list_processes(self.context, self.config["primary"], self.config["nt_symbols"])
kernel = self.context.modules[self.config['kernel']]
mods = modules.Modules.list_modules(self.context, self.config["primary"], self.config["nt_symbols"])
procs = pslist.PsList.list_processes(self.context, kernel.layer_name, kernel.symbol_table_name)
mods = modules.Modules.list_modules(self.context, kernel.layer_name, kernel.symbol_table_name)
# populate the session layers for kernel modules
session_layers = modules.Modules.get_session_layers(self.context, self.config['primary'],
self.config['nt_symbols'])
session_layers = modules.Modules.get_session_layers(self.context, kernel.layer_name, kernel.symbol_table_name)
return renderers.TreeGrid([("PID", int), ("Process", str), ("Base", format_hints.Hex), ("Name", str),
("Major", int), ("Minor", int), ("Product", int), ("Build", int)],
volatility3/framework/plugins/windows/virtmap.py
+7 -7
View File
@@ -16,16 +16,14 @@ vollog = logging.getLogger(__name__)
class VirtMap(interfaces.plugins.PluginInterface):
"""Lists virtual mapped sections."""
_required_framework_version = (1, 0, 0)
_required_framework_version = (1, 2, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"])
]
def _generator(self, map):
@@ -112,8 +110,10 @@ class VirtMap(interfaces.plugins.PluginInterface):
yield value
def run(self):
layer = self.context.layers[self.config['primary']]
module = self.context.module(self.config['nt_symbols'],
kernel = self.context.modules[self.config['kernel']]
layer = self.context.layers[kernel.layer_name]
module = self.context.module(kernel.symbol_table_name,
layer_name = layer.name,
offset = layer.config['kernel_virtual_offset'])