mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-07 02:07:39 +02:00
Plugins: Change most *dump plugins to --dump
This commit is contained in:
Mike Auty
@@ -1,115 +0,0 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
import ntpath
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import interfaces, constants, exceptions
|
||||
from volatility.framework import renderers
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.objects import utility
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins.windows import pslist, vadinfo
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DllDump(interfaces.plugins.PluginInterface):
|
||||
"""Dumps process memory ranges as DLLs."""
|
||||
|
||||
@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"),
|
||||
# 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 " \
|
||||
"(all other address ranges are excluded). This must be " \
|
||||
"a base address, not an address within the desired range.",
|
||||
optional = True),
|
||||
requirements.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process IDs to include (all other processes are excluded)",
|
||||
optional = True),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.PluginRequirement(name = 'vadinfo', plugin = vadinfo.VadInfo, version = (1, 0, 0)),
|
||||
]
|
||||
|
||||
def _generator(self, procs):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = extensions.pe.class_types)
|
||||
|
||||
filter_func = lambda _: False
|
||||
if self.config.get('address', None) is not None:
|
||||
filter_func = lambda x: x.get_start() not in [self.config['address']]
|
||||
|
||||
for proc in procs:
|
||||
process_name = utility.array_to_string(proc.ImageFileName)
|
||||
|
||||
proc_id = "Unknown"
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
except exceptions.InvalidAddressException as excp:
|
||||
vollog.debug("Process {}: invalid address {} in layer {}".format(proc_id, excp.invalid_address,
|
||||
excp.layer_name))
|
||||
continue
|
||||
|
||||
for vad in vadinfo.VadInfo.list_vads(proc, filter_func = filter_func):
|
||||
|
||||
# this parameter is inherited from the VadInfo plugin. if a user specifies
|
||||
# an address, then it bypasses the DLL identification heuristics
|
||||
if self.config.get("address", None) is None:
|
||||
|
||||
# rather than relying on the PEB for DLLs, which can be swapped,
|
||||
# it requires special handling on wow64 processes, and its
|
||||
# unreliable from an integrity standpoint, let's use the VADs instead
|
||||
protection_string = vad.get_protection(
|
||||
vadinfo.VadInfo.protect_values(self.context, self.config['primary'], self.config['nt_symbols']),
|
||||
vadinfo.winnt_protections)
|
||||
|
||||
# DLLs are write copy...
|
||||
if protection_string != "PAGE_EXECUTE_WRITECOPY":
|
||||
continue
|
||||
|
||||
# DLLs have mapped files...
|
||||
if isinstance(vad.get_file_name(), interfaces.renderers.BaseAbsentValue):
|
||||
continue
|
||||
|
||||
try:
|
||||
filedata = interfaces.plugins.FileInterface("pid.{0}.{1}.{2:#x}.dmp".format(
|
||||
proc.UniqueProcessId, ntpath.basename(vad.get_file_name()), vad.get_start()))
|
||||
|
||||
dos_header = self.context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset = vad.get_start(),
|
||||
layer_name = proc_layer_name)
|
||||
|
||||
for offset, data in dos_header.reconstruct():
|
||||
filedata.data.seek(offset)
|
||||
filedata.data.write(data)
|
||||
|
||||
self.produce_file(filedata)
|
||||
result_text = "Stored {}".format(filedata.preferred_filename)
|
||||
except Exception:
|
||||
result_text = "Unable to dump PE at {0:#x}".format(vad.get_start())
|
||||
|
||||
yield (0, (proc.UniqueProcessId, process_name, result_text))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Result", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
@@ -1,18 +1,25 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
import ntpath
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import exceptions, renderers, interfaces
|
||||
from volatility.framework import exceptions, renderers, interfaces, constants
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins.windows import pslist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DllList(interfaces.plugins.PluginInterface):
|
||||
"""Lists the loaded modules in a particular windows memory image."""
|
||||
|
||||
_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
|
||||
@@ -25,13 +32,65 @@ class DllList(interfaces.plugins.PluginInterface):
|
||||
requirements.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process IDs to include (all other processes are excluded)",
|
||||
optional = True)
|
||||
optional = True),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed processes",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def dump_dll(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
pe_table_name: str,
|
||||
dll_entry: interfaces.objects.ObjectInterface,
|
||||
layer_name: str = None) -> interfaces.plugins.FileInterface:
|
||||
"""Extracts the complete data for a process as a FileInterface
|
||||
|
||||
Args:
|
||||
context: the context to operate upon
|
||||
pe_table_name: the name for the symbol table containing the PE format symbols
|
||||
dll_entry: the object representing the module
|
||||
layer_name: the layer that the DLL lives within
|
||||
|
||||
Returns:
|
||||
A FileInterface object containing the complete data for the DLL or None in the case of failure"""
|
||||
filedata = None
|
||||
try:
|
||||
try:
|
||||
name = dll_entry.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
name = 'UnreadbleDLLName'
|
||||
|
||||
if layer_name is None:
|
||||
layer_name = dll_entry.vol.layer_name
|
||||
|
||||
filedata = interfaces.plugins.FileInterface(
|
||||
"{0}.{1:#x}.{2:#x}.dmp".format(ntpath.basename(name), dll_entry.vol.offset, dll_entry.DllBase))
|
||||
|
||||
dos_header = context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset = dll_entry.DllBase,
|
||||
layer_name = layer_name)
|
||||
|
||||
for offset, data in dos_header.reconstruct():
|
||||
filedata.data.seek(offset)
|
||||
filedata.data.write(data)
|
||||
except Exception as excp:
|
||||
vollog.debug("Unable to dump dll at offset {}: {}".format(dll_entry.DllBase, excp))
|
||||
return filedata
|
||||
|
||||
def _generator(self, procs):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = extensions.pe.class_types)
|
||||
|
||||
for proc in procs:
|
||||
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
for entry in proc.load_order_modules():
|
||||
|
||||
BaseDllName = FullDllName = renderers.UnreadableValue()
|
||||
@@ -42,17 +101,25 @@ class DllList(interfaces.plugins.PluginInterface):
|
||||
except exceptions.InvalidAddressException:
|
||||
pass
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
filedata = self.dump_dll(self.context, pe_table_name, entry, proc_layer_name)
|
||||
if filedata:
|
||||
filedata.preferred_filename = "pid.{0}.".format(proc_id) + filedata.preferred_filename
|
||||
dumped = True
|
||||
self.produce_file(filedata)
|
||||
|
||||
yield (0, (proc.UniqueProcessId,
|
||||
proc.ImageFileName.cast("string",
|
||||
max_length = proc.ImageFileName.vol.count,
|
||||
errors = 'replace'), format_hints.Hex(entry.DllBase),
|
||||
format_hints.Hex(entry.SizeOfImage), BaseDllName, FullDllName))
|
||||
format_hints.Hex(entry.SizeOfImage), BaseDllName, FullDllName, dumped))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Base", format_hints.Hex),
|
||||
("Size", format_hints.Hex), ("Name", str), ("Path", str)],
|
||||
("Size", format_hints.Hex), ("Name", str), ("Path", str), ("Dumped", bool)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import exceptions, renderers, interfaces
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.plugins.windows import pslist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Memdump(interfaces.plugins.PluginInterface):
|
||||
"""Dump the addressable memory for a process"""
|
||||
|
||||
@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.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.IntRequirement(name = 'pid',
|
||||
description = "Process ID to include (all other processes are excluded)",
|
||||
optional = True)
|
||||
]
|
||||
|
||||
def _generator(self, procs):
|
||||
for proc in procs:
|
||||
data = b""
|
||||
process_name = proc.ImageFileName.cast("string",
|
||||
max_length = proc.ImageFileName.vol.count,
|
||||
errors = 'replace')
|
||||
pid = "Unknown"
|
||||
try:
|
||||
pid = proc.UniqueProcessId
|
||||
offset = format_hints.Hex(proc.vol.offset)
|
||||
filename = str(pid) + "." + str(offset)
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
proc_layer = self.context.layers[proc_layer_name]
|
||||
except exceptions.InvalidAddressException as excp:
|
||||
vollog.debug(
|
||||
"Process {}: invalid address {} in layer {}".format(pid, excp.invalid_address, excp.layer_name))
|
||||
continue
|
||||
|
||||
# Create file for writing
|
||||
filedata = interfaces.plugins.FileInterface("{}.dmp".format(filename))
|
||||
|
||||
for mapval in proc_layer.mapping(0x0, proc_layer.maximum_address, ignore_errors = True):
|
||||
offset, size, mapped_offset, _, maplayer = mapval
|
||||
data = proc_layer.read(offset, size, pad = True)
|
||||
try:
|
||||
filedata.data.write(data)
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.debug("Unable to write {}'s address {} [ {} ]to {}.dmp".format(process_name, offset,
|
||||
proc.UniqueProcessId,
|
||||
proc.UniqueProcessId))
|
||||
continue
|
||||
|
||||
try:
|
||||
result_text = "Writing {} [ {} ] to {}.dmp".format(process_name, proc.UniqueProcessId, filename)
|
||||
self.produce_file(filedata)
|
||||
except exceptions.InvalidAddressException:
|
||||
result_text = "Unable to write {} [ {} ]to {}.dmp".format(process_name, proc.UniqueProcessId, filename)
|
||||
|
||||
yield (0, (result_text,))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
|
||||
return renderers.TreeGrid([("Creating the following files:", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
@@ -26,7 +26,11 @@ class Memmap(interfaces.plugins.PluginInterface):
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.IntRequirement(name = 'pid',
|
||||
description = "Process ID to include (all other processes are excluded)",
|
||||
optional = True)
|
||||
optional = True),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed memory segments",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
def _generator(self, procs):
|
||||
@@ -43,22 +47,40 @@ class Memmap(interfaces.plugins.PluginInterface):
|
||||
pid, excp.invalid_address, excp.layer_name))
|
||||
continue
|
||||
|
||||
filename = str(pid)
|
||||
filedata = interfaces.plugins.FileInterface("pid.{}.dmp".format(filename))
|
||||
|
||||
for mapval in proc_layer.mapping(0x0, proc_layer.maximum_address, ignore_errors = True):
|
||||
offset, _, mapped_offset, mapped_size, maplayer = mapval
|
||||
offset, size, mapped_offset, mapped_size, maplayer = mapval
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
try:
|
||||
data = proc_layer.read(offset, size, pad = True)
|
||||
filedata.data.write(data)
|
||||
dumped = True
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.debug("Unable to write {}'s address {} to {}.dmp".format(proc_layer_name, offset,
|
||||
filedata.preferred_filename))
|
||||
|
||||
yield (0, (
|
||||
format_hints.Hex(offset),
|
||||
format_hints.Hex(mapped_offset),
|
||||
format_hints.Hex(mapped_size),
|
||||
format_hints.Hex(offset)))
|
||||
format_hints.Hex(offset),
|
||||
dumped))
|
||||
offset += mapped_size
|
||||
|
||||
import pdb
|
||||
pdb.set_trace()
|
||||
self.produce_file(filedata)
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
|
||||
|
||||
return renderers.TreeGrid(
|
||||
[("Virtual", format_hints.Hex), ("Physical", format_hints.Hex), ("Size", format_hints.Hex),
|
||||
("Offset", format_hints.Hex)],
|
||||
("Offset", format_hints.Hex), ("Dumped", bool)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
|
||||
@@ -33,79 +33,6 @@ class ModDump(interfaces.plugins.PluginInterface):
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def get_session_layers(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
layer_name: str,
|
||||
symbol_table: str,
|
||||
pids: List[int] = None) -> Generator[str, None, None]:
|
||||
"""Build a cache of possible virtual layers, in priority starting with
|
||||
the primary/kernel layer. Then keep one layer per session by cycling
|
||||
through the process list.
|
||||
|
||||
Args:
|
||||
context: The context to retrieve required elements (layers, symbol tables) from
|
||||
layer_name: The name of the layer on which to operate
|
||||
symbol_table: The name of the table containing the kernel symbols
|
||||
pids: A list of process identifiers to include exclusively or None for no filter
|
||||
|
||||
Returns:
|
||||
A list of session layer names
|
||||
"""
|
||||
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
for proc in pslist.PsList.list_processes(context = context,
|
||||
layer_name = layer_name,
|
||||
symbol_table = symbol_table,
|
||||
filter_func = filter_func):
|
||||
proc_id = "Unknown"
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
# create the session space object in the process' own layer.
|
||||
# not all processes have a valid session pointer.
|
||||
session_space = context.object(symbol_table + constants.BANG + "_MM_SESSION_SPACE",
|
||||
layer_name = layer_name,
|
||||
offset = proc.Session)
|
||||
|
||||
if session_space.SessionId in seen_ids:
|
||||
continue
|
||||
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV,
|
||||
"Process {} does not have a valid Session or a layer could not be constructed for it".format(
|
||||
proc_id))
|
||||
continue
|
||||
|
||||
# save the layer if we haven't seen the session yet
|
||||
seen_ids.append(session_space.SessionId)
|
||||
yield proc_layer_name
|
||||
|
||||
@classmethod
|
||||
def find_session_layer(cls, context: interfaces.context.ContextInterface, session_layers: Iterable[str],
|
||||
base_address: int):
|
||||
"""Given a base address and a list of layer names, find a layer that
|
||||
can access the specified address.
|
||||
|
||||
Args:
|
||||
context: The context to retrieve required elements (layers, symbol tables) from
|
||||
layer_name: The name of the layer on which to operate
|
||||
symbol_table: The name of the table containing the kernel symbols
|
||||
session_layers: A list of session layer names
|
||||
base_address: The base address to identify the layers that can access it
|
||||
|
||||
Returns:
|
||||
Layer name or None if no layers that contain the base address can be found
|
||||
"""
|
||||
|
||||
for layer_name in session_layers:
|
||||
if context.layers[layer_name].is_valid(base_address):
|
||||
return layer_name
|
||||
|
||||
return None
|
||||
|
||||
def _generator(self, mods):
|
||||
|
||||
|
||||
@@ -7,7 +7,9 @@ from typing import Iterable
|
||||
from volatility.framework import renderers, interfaces, exceptions
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.plugins.windows import poolscanner
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins.windows import poolscanner, dlllist
|
||||
|
||||
|
||||
class ModScan(interfaces.plugins.PluginInterface):
|
||||
@@ -20,6 +22,13 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
description = 'Memory layer for the kernel',
|
||||
architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
requirements.VersionRequirement(name = 'poolerscanner', component = poolscanner.PoolScanner,
|
||||
version = (1, 0, 0)),
|
||||
requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (1, 0, 0)),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed modules",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@@ -47,6 +56,12 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
yield mem_object
|
||||
|
||||
def _generator(self):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = extensions.pe.class_types)
|
||||
|
||||
for mod in self.scan_modules(self.context, self.config['primary'], self.config['nt_symbols']):
|
||||
|
||||
try:
|
||||
@@ -59,14 +74,22 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
except exceptions.InvalidAddressException:
|
||||
FullDllName = ""
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
filedata = dlllist.DllList.dump_dll(self.context, pe_table_name, mod)
|
||||
if filedata:
|
||||
self.produce_file(filedata)
|
||||
dumped = True
|
||||
|
||||
yield (0, (
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
dumped
|
||||
))
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Base", format_hints.Hex), ("Size", format_hints.Hex),
|
||||
("Name", str), ("Path", str)], self._generator())
|
||||
("Name", str), ("Path", str), ("Dumped", bool)], self._generator())
|
||||
|
||||
@@ -1,14 +1,19 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
from typing import List, Iterable
|
||||
import logging
|
||||
from typing import List, Iterable, Generator
|
||||
|
||||
from volatility.framework import constants
|
||||
from volatility.framework import exceptions, interfaces
|
||||
from volatility.framework import renderers
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins.windows import pslist, dlllist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Modules(interfaces.plugins.PluginInterface):
|
||||
@@ -22,10 +27,22 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
requirements.TranslationLayerRequirement(name = 'primary',
|
||||
description = 'Memory layer for the kernel',
|
||||
architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (1, 1, 0)),
|
||||
requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (1, 0, 0)),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed modules",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
def _generator(self):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = extensions.pe.class_types)
|
||||
|
||||
for mod in self.list_modules(self.context, self.config['primary'], self.config['nt_symbols']):
|
||||
|
||||
try:
|
||||
@@ -38,14 +55,96 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
except exceptions.InvalidAddressException:
|
||||
FullDllName = ""
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
filedata = dlllist.DllList.dump_dll(self.context, pe_table_name, mod)
|
||||
if filedata:
|
||||
self.produce_file(filedata)
|
||||
dumped = True
|
||||
|
||||
yield (0, (
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
dumped
|
||||
))
|
||||
|
||||
@classmethod
|
||||
def get_session_layers(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
layer_name: str,
|
||||
symbol_table: str,
|
||||
pids: List[int] = None) -> Generator[str, None, None]:
|
||||
"""Build a cache of possible virtual layers, in priority starting with
|
||||
the primary/kernel layer. Then keep one layer per session by cycling
|
||||
through the process list.
|
||||
|
||||
Args:
|
||||
context: The context to retrieve required elements (layers, symbol tables) from
|
||||
layer_name: The name of the layer on which to operate
|
||||
symbol_table: The name of the table containing the kernel symbols
|
||||
pids: A list of process identifiers to include exclusively or None for no filter
|
||||
|
||||
Returns:
|
||||
A list of session layer names
|
||||
"""
|
||||
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
for proc in pslist.PsList.list_processes(context = context,
|
||||
layer_name = layer_name,
|
||||
symbol_table = symbol_table,
|
||||
filter_func = filter_func):
|
||||
proc_id = "Unknown"
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
# create the session space object in the process' own layer.
|
||||
# not all processes have a valid session pointer.
|
||||
session_space = context.object(symbol_table + constants.BANG + "_MM_SESSION_SPACE",
|
||||
layer_name = layer_name,
|
||||
offset = proc.Session)
|
||||
|
||||
if session_space.SessionId in seen_ids:
|
||||
continue
|
||||
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV,
|
||||
"Process {} does not have a valid Session or a layer could not be constructed for it".format(
|
||||
proc_id))
|
||||
continue
|
||||
|
||||
# save the layer if we haven't seen the session yet
|
||||
seen_ids.append(session_space.SessionId)
|
||||
yield proc_layer_name
|
||||
|
||||
@classmethod
|
||||
def find_session_layer(cls, context: interfaces.context.ContextInterface, session_layers: Iterable[str],
|
||||
base_address: int):
|
||||
"""Given a base address and a list of layer names, find a layer that
|
||||
can access the specified address.
|
||||
|
||||
Args:
|
||||
context: The context to retrieve required elements (layers, symbol tables) from
|
||||
layer_name: The name of the layer on which to operate
|
||||
symbol_table: The name of the table containing the kernel symbols
|
||||
session_layers: A list of session layer names
|
||||
base_address: The base address to identify the layers that can access it
|
||||
|
||||
Returns:
|
||||
Layer name or None if no layers that contain the base address can be found
|
||||
"""
|
||||
|
||||
for layer_name in session_layers:
|
||||
if context.layers[layer_name].is_valid(base_address):
|
||||
return layer_name
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str,
|
||||
symbol_table: str) -> Iterable[interfaces.objects.ObjectInterface]:
|
||||
@@ -81,4 +180,4 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Base", format_hints.Hex), ("Size", format_hints.Hex),
|
||||
("Name", str), ("Path", str)], self._generator())
|
||||
("Name", str), ("Path", str), ("Dumped", bool)], self._generator())
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import interfaces, exceptions, constants, renderers
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.objects import utility
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows.extensions import pe
|
||||
from volatility.plugins.windows import pslist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ProcDump(interfaces.plugins.PluginInterface):
|
||||
"""Dumps process executable images."""
|
||||
|
||||
_version = (1, 1, 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.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process IDs to include (all other processes are excluded)",
|
||||
optional = True),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def process_dump(cls, context: interfaces.context.ContextInterface, kernel_table_name: str, pe_table_name: str,
|
||||
proc: interfaces.objects.ObjectInterface) -> interfaces.plugins.FileInterface:
|
||||
"""Extracts the complete data for a process as a FileInterface
|
||||
|
||||
Args:
|
||||
context: the context to operate upon
|
||||
kernel_table_name: the name for the symbol table containing the kernel's symbols
|
||||
pe_table_name: the name for the symbol table containing the PE format symbols
|
||||
proc: the process object whose memory should be output
|
||||
|
||||
Returns:
|
||||
A FileInterface object containing the complete data for the process
|
||||
"""
|
||||
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
peb = context.object(kernel_table_name + constants.BANG + "_PEB",
|
||||
layer_name = proc_layer_name,
|
||||
offset = proc.Peb)
|
||||
dos_header = context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset = peb.ImageBaseAddress,
|
||||
layer_name = proc_layer_name)
|
||||
filedata = interfaces.plugins.FileInterface("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId,
|
||||
peb.ImageBaseAddress))
|
||||
for offset, data in dos_header.reconstruct():
|
||||
filedata.data.seek(offset)
|
||||
filedata.data.write(data)
|
||||
|
||||
return filedata
|
||||
|
||||
def _generator(self, procs):
|
||||
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = pe.class_types)
|
||||
|
||||
for proc in procs:
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
process_name = utility.array_to_string(proc.ImageFileName)
|
||||
filedata = self.process_dump(self.context, self.config["nt_symbols"], pe_table_name, proc)
|
||||
self.produce_file(filedata)
|
||||
result_text = "Stored {}".format(filedata.preferred_filename)
|
||||
except ValueError:
|
||||
result_text = "PE parsing error"
|
||||
except exceptions.SwappedInvalidAddressException as exp:
|
||||
result_text = "Process {}: Required memory at {:#x} is inaccessible (swapped)".format(
|
||||
proc_id, exp.invalid_address)
|
||||
|
||||
except exceptions.PagedInvalidAddressException as exp:
|
||||
result_text = "Process {}: Required memory at {:#x} is not valid (process exited?)".format(
|
||||
proc_id, exp.invalid_address)
|
||||
|
||||
except exceptions.InvalidAddressException as exp:
|
||||
result_text = "Process {}: Required memory at {:#x} is not valid (incomplete layer {}?)".format(
|
||||
proc_id, exp.invalid_address, exp.layer_name)
|
||||
|
||||
yield (0, (proc.UniqueProcessId, process_name, result_text))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Result", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
@@ -3,19 +3,24 @@
|
||||
#
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
from typing import Callable, Iterable, List
|
||||
|
||||
from volatility.framework import renderers, interfaces, layers
|
||||
from volatility.framework import renderers, interfaces, layers, constants
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.objects import utility
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins import timeliner
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
"""Lists the processes present in a particular windows memory image."""
|
||||
|
||||
_version = (1, 0, 0)
|
||||
_version = (1, 1, 0)
|
||||
PHYSICAL_DEFAULT = False
|
||||
|
||||
@classmethod
|
||||
@@ -25,7 +30,6 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
description = 'Memory layer for the kernel',
|
||||
architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
# TODO: Convert this to a ListRequirement so that people can filter on sets of pids
|
||||
requirements.BooleanRequirement(name = 'physical',
|
||||
description = 'Display physical offsets instead of virtual',
|
||||
default = cls.PHYSICAL_DEFAULT,
|
||||
@@ -33,9 +37,48 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
requirements.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process ID to include (all other processes are excluded)",
|
||||
optional = True)
|
||||
optional = True),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed processes",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def process_dump(cls, context: interfaces.context.ContextInterface, kernel_table_name: str, pe_table_name: str,
|
||||
proc: interfaces.objects.ObjectInterface) -> interfaces.plugins.FileInterface:
|
||||
"""Extracts the complete data for a process as a FileInterface
|
||||
|
||||
Args:
|
||||
context: the context to operate upon
|
||||
kernel_table_name: the name for the symbol table containing the kernel's symbols
|
||||
pe_table_name: the name for the symbol table containing the PE format symbols
|
||||
proc: the process object whose memory should be output
|
||||
|
||||
Returns:
|
||||
A FileInterface object containing the complete data for the process or None in the case of failure
|
||||
"""
|
||||
|
||||
filedata = None
|
||||
try:
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
peb = context.object(kernel_table_name + constants.BANG + "_PEB",
|
||||
layer_name = proc_layer_name,
|
||||
offset = proc.Peb)
|
||||
|
||||
dos_header = context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset = peb.ImageBaseAddress,
|
||||
layer_name = proc_layer_name)
|
||||
filedata = interfaces.plugins.FileInterface("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId,
|
||||
peb.ImageBaseAddress))
|
||||
for offset, data in dos_header.reconstruct():
|
||||
filedata.data.seek(offset)
|
||||
filedata.data.write(data)
|
||||
except Exception as excp:
|
||||
vollog.debug("Unable to dump PE with pid {}: {}".format(proc.UniqueProcessId, excp))
|
||||
|
||||
return filedata
|
||||
|
||||
@classmethod
|
||||
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[interfaces.objects.ObjectInterface], bool]:
|
||||
"""A factory for producing filter functions that filter based on a list
|
||||
@@ -120,6 +163,11 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
yield proc
|
||||
|
||||
def _generator(self):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
|
||||
self.config_path,
|
||||
"windows",
|
||||
"pe",
|
||||
class_types = extensions.pe.class_types)
|
||||
|
||||
memory = self.context.layers[self.config['primary']]
|
||||
if not isinstance(memory, layers.intel.Intel):
|
||||
@@ -135,10 +183,17 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
else:
|
||||
(_, _, offset, _, _) = list(memory.mapping(offset = proc.vol.offset, length = 0))[0]
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
filedata = self.process_dump(self.context, self.config['nt_symbols'], pe_table_name, proc)
|
||||
if filedata:
|
||||
dumped = True
|
||||
self.produce_file(filedata)
|
||||
|
||||
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()))
|
||||
proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time(), dumped))
|
||||
|
||||
def generate_timeline(self):
|
||||
for row in self._generator():
|
||||
@@ -153,5 +208,6 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str),
|
||||
("Offset{0}".format(offsettype), format_hints.Hex), ("Threads", int),
|
||||
("Handles", int), ("SessionId", int), ("Wow64", bool),
|
||||
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime)],
|
||||
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime),
|
||||
("Dumped", bool)],
|
||||
self._generator())
|
||||
|
||||
@@ -1,67 +0,0 @@
|
||||
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import interfaces
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.interfaces import plugins
|
||||
from volatility.framework.renderers import TreeGrid, format_hints
|
||||
from volatility.plugins.windows.registry import hivelist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HiveDump(interfaces.plugins.PluginInterface):
|
||||
"""Dumps the hive files (or a specific hive) from an image."""
|
||||
|
||||
@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.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)),
|
||||
requirements.IntRequirement(name = 'offset', description = "Hive Offset", default = None, optional = True),
|
||||
]
|
||||
|
||||
def _sanitize_hive_name(self, name: str) -> str:
|
||||
return name.split('\\')[-1].replace(' ', '_').replace('.', '').replace('[', '').replace(']', '')
|
||||
|
||||
def _generator(self, layer_name, symbol_table, hive_offsets):
|
||||
chunk_size = 0x500000
|
||||
for hive in hivelist.HiveList.list_hives(self.context,
|
||||
self.config_path,
|
||||
layer_name = layer_name,
|
||||
symbol_table = symbol_table,
|
||||
hive_offsets = hive_offsets):
|
||||
|
||||
maxaddr = hive.hive.Storage[0].Length
|
||||
hive_name = self._sanitize_hive_name(hive.get_name())
|
||||
|
||||
filedata = plugins.FileInterface('registry.{}.{}.hive'.format(hive_name, hex(hive.hive_offset)))
|
||||
if hive._base_block:
|
||||
hive_data = self.context.layers[hive.dependencies[0]].read(hive.hive.BaseBlock, 1 << 12)
|
||||
else:
|
||||
hive_data = '\x00' * (1 << 12)
|
||||
filedata.data.write(hive_data)
|
||||
|
||||
for i in range(0, maxaddr, chunk_size):
|
||||
current_chunk_size = min(chunk_size, maxaddr - i)
|
||||
data = hive.read(i, current_chunk_size, pad = True)
|
||||
filedata.data.write(data)
|
||||
# if self._progress_callback:
|
||||
# self._progress_callback((i / maxaddr) * 100, 'Writing layer {}'.format(hive_name))
|
||||
self.produce_file(filedata)
|
||||
yield (0, (hive.name, format_hints.Hex(hive.hive_offset),
|
||||
'Written to {}'.format(filedata.preferred_filename)))
|
||||
|
||||
def run(self) -> interfaces.renderers.TreeGrid:
|
||||
offset = self.config.get('offset', None)
|
||||
return TreeGrid(columns = [('Hive name', str), ('Hive Offset', format_hints.Hex), ('status', str)],
|
||||
generator = self._generator(self.config['primary'],
|
||||
self.config['nt_symbols'],
|
||||
hive_offsets = None if offset is None else [offset]))
|
||||
@@ -6,6 +6,7 @@ from typing import Iterator, List, Tuple, Iterable, Optional
|
||||
|
||||
from volatility.framework import renderers, interfaces, exceptions
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.interfaces import plugins
|
||||
from volatility.framework.layers import registry
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.plugins.windows.registry import hivescan
|
||||
@@ -13,7 +14,7 @@ from volatility.plugins.windows.registry import hivescan
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HiveGenerator():
|
||||
class HiveGenerator:
|
||||
"""Walks the registry HiveList linked list in a given direction and stores an invalid offset
|
||||
if it's unable to fully walk the list"""
|
||||
|
||||
@@ -51,15 +52,52 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
optional = True,
|
||||
default = None),
|
||||
requirements.PluginRequirement(name = 'hivescan', plugin = hivescan.HiveScan, version = (1, 0, 0)),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed processes",
|
||||
default = False,
|
||||
optional = True)
|
||||
|
||||
]
|
||||
|
||||
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]:
|
||||
for hive in self.list_hive_objects(context = self.context,
|
||||
layer_name = self.config["primary"],
|
||||
symbol_table = self.config["nt_symbols"],
|
||||
filter_string = self.config.get('filter', None)):
|
||||
def _sanitize_hive_name(self, name: str) -> str:
|
||||
return name.split('\\')[-1].replace(' ', '_').replace('.', '').replace('[', '').replace(']', '')
|
||||
|
||||
yield (0, (format_hints.Hex(hive.vol.offset), hive.get_name() or ""))
|
||||
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]:
|
||||
chunk_size = 0x500000
|
||||
|
||||
for hive_object in self.list_hive_objects(context = self.context,
|
||||
layer_name = self.config["primary"],
|
||||
symbol_table = self.config["nt_symbols"],
|
||||
filter_string = self.config.get('filter', None)):
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
# Construct the hive
|
||||
hive = next(self.list_hives(self.context,
|
||||
self.config_path,
|
||||
layer_name = self.config["primary"],
|
||||
symbol_table = self.config["nt_symbols"],
|
||||
hive_offsets = [hive_object.vol.offset]))
|
||||
maxaddr = hive.hive.Storage[0].Length
|
||||
hive_name = self._sanitize_hive_name(hive.get_name())
|
||||
|
||||
filedata = plugins.FileInterface('registry.{}.{}.hive'.format(hive_name, hex(hive.hive_offset)))
|
||||
if hive._base_block:
|
||||
hive_data = self.context.layers[hive.dependencies[0]].read(hive.hive.BaseBlock, 1 << 12)
|
||||
else:
|
||||
hive_data = '\x00' * (1 << 12)
|
||||
filedata.data.write(hive_data)
|
||||
|
||||
for i in range(0, maxaddr, chunk_size):
|
||||
current_chunk_size = min(chunk_size, maxaddr - i)
|
||||
data = hive.read(i, current_chunk_size, pad = True)
|
||||
filedata.data.write(data)
|
||||
# if self._progress_callback:
|
||||
# self._progress_callback((i / maxaddr) * 100, 'Writing layer {}'.format(hive_name))
|
||||
self.produce_file(filedata)
|
||||
dumped = True
|
||||
|
||||
yield (0, (format_hints.Hex(hive_object.vol.offset), hive_object.get_name() or "", dumped))
|
||||
|
||||
@classmethod
|
||||
def list_hives(cls,
|
||||
@@ -77,7 +115,7 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
base_config_path: The configuration path for any settings required by the new table
|
||||
layer_name: The name of the layer on which to operate
|
||||
symbol_table: The name of the table containing the kernel symbols
|
||||
filter_string: An optional string which must be present in the hive name if specified
|
||||
filter_string: An optional string which must be present in the hive name if specified
|
||||
offset: An optional offset to specify a specific hive to iterate over (takes precedence over filter_string)
|
||||
|
||||
Yields:
|
||||
@@ -141,8 +179,8 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
hg = HiveGenerator(cmhive, forward = True)
|
||||
for hive in hg:
|
||||
if hive.vol.offset in seen:
|
||||
vollog.debug("Hivelist found an already seen offset {} while "\
|
||||
"traversing forwards, this should not occur".format(hex(hive.vol.offset)))
|
||||
vollog.debug("Hivelist found an already seen offset {} while " \
|
||||
"traversing forwards, this should not occur".format(hex(hive.vol.offset)))
|
||||
break
|
||||
seen.add(hive.vol.offset)
|
||||
if filter_string is None or filter_string.lower() in str(hive.get_name() or "").lower():
|
||||
@@ -156,7 +194,7 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
hg = HiveGenerator(cmhive, forward = False)
|
||||
for hive in hg:
|
||||
if hive.vol.offset in seen:
|
||||
vollog.debug("Hivelist found an already seen offset {} while "\
|
||||
vollog.debug("Hivelist found an already seen offset {} while " \
|
||||
"traversing backwards, list walking met in the middle".format(hex(hive.vol.offset)))
|
||||
break
|
||||
seen.add(hive.vol.offset)
|
||||
@@ -173,8 +211,8 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
# therefore, there must be more 2 or more invalid hives, so the middle of the list is not reachable
|
||||
# by walking the list, so revert to scanning, and walk the list forwards and backwards from each
|
||||
# found hive
|
||||
vollog.debug("Hivelist failed traversing backwards at {}, a different "\
|
||||
"location from forwards, revert to scanning".format(hex(backward_invalid)))
|
||||
vollog.debug("Hivelist failed traversing backwards at {}, a different " \
|
||||
"location from forwards, revert to scanning".format(hex(backward_invalid)))
|
||||
for hive in hivescan.HiveScan.scan_hives(context, layer_name, symbol_table):
|
||||
try:
|
||||
if hive.HiveList.Flink:
|
||||
@@ -198,4 +236,5 @@ class HiveList(interfaces.plugins.PluginInterface):
|
||||
hex(hive.vol.offset)))
|
||||
|
||||
def run(self) -> renderers.TreeGrid:
|
||||
return renderers.TreeGrid([("Offset", format_hints.Hex), ("FileFullPath", str)], self._generator())
|
||||
return renderers.TreeGrid([("Offset", format_hints.Hex), ("FileFullPath", str), ("Dumped", bool)],
|
||||
self._generator())
|
||||
|
||||
@@ -1,105 +0,0 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from volatility.framework import renderers, interfaces, exceptions
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.objects import utility
|
||||
from volatility.plugins.windows import pslist, vadinfo
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class VadDump(interfaces.plugins.PluginInterface):
|
||||
"""Dumps process memory ranges."""
|
||||
_version = (1, 1, 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"),
|
||||
# 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 " \
|
||||
"(all other address ranges are excluded). This must be " \
|
||||
"a base address, not an address within the desired range.",
|
||||
optional = True),
|
||||
requirements.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process IDs to include (all other processes are excluded)",
|
||||
optional = True),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.PluginRequirement(name = 'vadinfo', plugin = vadinfo.VadInfo, version = (1, 0, 0)),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def vad_dump(cls, context: interfaces.context.ContextInterface, layer_name: str,
|
||||
vad: interfaces.objects.ObjectInterface) -> bytes:
|
||||
"""
|
||||
Returns VAD content
|
||||
"""
|
||||
|
||||
tmp_data = b""
|
||||
proc_layer = context.layers[layer_name]
|
||||
chunk_size = 1024 * 1024 * 10
|
||||
offset = vad.get_start()
|
||||
out_of_range = vad.get_end()
|
||||
# print("walking from {:x} to {:x} | {:x}".format(offset, out_of_range, out_of_range-offset))
|
||||
while offset < out_of_range:
|
||||
to_read = min(chunk_size, out_of_range - offset)
|
||||
data = proc_layer.read(offset, to_read, pad = True)
|
||||
if not data:
|
||||
break
|
||||
tmp_data += data
|
||||
offset += to_read
|
||||
|
||||
return tmp_data
|
||||
|
||||
def _generator(self, procs):
|
||||
|
||||
filter_func = lambda _: False
|
||||
if self.config.get('address', None) is not None:
|
||||
filter_func = lambda x: x.get_start() not in [self.config['address']]
|
||||
|
||||
for proc in procs:
|
||||
process_name = utility.array_to_string(proc.ImageFileName)
|
||||
|
||||
proc_id = "Unknown"
|
||||
try:
|
||||
proc_id = proc.UniqueProcessId
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
except exceptions.InvalidAddressException as excp:
|
||||
vollog.debug("Process {}: invalid address {} in layer {}".format(proc_id, excp.invalid_address,
|
||||
excp.layer_name))
|
||||
continue
|
||||
|
||||
for vad in vadinfo.VadInfo.list_vads(proc, filter_func = filter_func):
|
||||
try:
|
||||
filedata = interfaces.plugins.FileInterface("pid.{0}.vad.{1:#x}-{2:#x}.dmp".format(
|
||||
proc.UniqueProcessId, vad.get_start(), vad.get_end()))
|
||||
|
||||
data = self.vad_dump(self.context, proc_layer_name, vad)
|
||||
filedata.data.write(data)
|
||||
|
||||
self.produce_file(filedata)
|
||||
result_text = "Stored {}".format(filedata.preferred_filename)
|
||||
except exceptions.InvalidAddressException:
|
||||
result_text = "Unable to dump {0:#x} - {1:#x}".format(vad.get_start(), vad.get_end())
|
||||
|
||||
yield (0, (proc.UniqueProcessId, process_name, result_text))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Result", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
@@ -57,6 +57,10 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
element_type = int,
|
||||
optional = True),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.BooleanRequirement(name = 'dump',
|
||||
description = "Extract listed processes",
|
||||
default = False,
|
||||
optional = True)
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@@ -96,29 +100,70 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
if not filter_func(vad):
|
||||
yield vad
|
||||
|
||||
@classmethod
|
||||
def vad_dump(cls, context: interfaces.context.ContextInterface, layer_name: str,
|
||||
vad: interfaces.objects.ObjectInterface) -> bytes:
|
||||
"""Extracts the complete data for Vad as a FileInterface
|
||||
|
||||
Args:
|
||||
context: the context to operate upon
|
||||
layer_name: the name of the layer that the VAD lives within
|
||||
vad: the virtual address descriptor to be dumped
|
||||
|
||||
Returns:
|
||||
bytes containing the data from the vad
|
||||
"""
|
||||
|
||||
tmp_data = b""
|
||||
proc_layer = context.layers[layer_name]
|
||||
chunk_size = 1024 * 1024 * 10
|
||||
offset = vad.get_start()
|
||||
out_of_range = vad.get_end()
|
||||
# print("walking from {:x} to {:x} | {:x}".format(offset, out_of_range, out_of_range-offset))
|
||||
while offset < out_of_range:
|
||||
to_read = min(chunk_size, out_of_range - offset)
|
||||
data = proc_layer.read(offset, to_read, pad = True)
|
||||
if not data:
|
||||
break
|
||||
tmp_data += data
|
||||
offset += to_read
|
||||
|
||||
return tmp_data
|
||||
|
||||
def _generator(self, procs):
|
||||
|
||||
def passthrough(_: 'interfaces.objects.ObjectInterface') -> bool:
|
||||
def passthrough(_: interfaces.objects.ObjectInterface) -> bool:
|
||||
return False
|
||||
|
||||
filter_func = passthrough
|
||||
if self.config.get('address', None) is not None:
|
||||
|
||||
def filter_function(x: 'interfaces.objects.ObjectInterface') -> bool:
|
||||
def filter_function(x: interfaces.objects.ObjectInterface) -> bool:
|
||||
return x.get_start() not in [self.config['address']]
|
||||
|
||||
filter_func = filter_function
|
||||
|
||||
for proc in procs:
|
||||
process_name = utility.array_to_string(proc.ImageFileName)
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
for vad in self.list_vads(proc, filter_func = filter_func):
|
||||
|
||||
dumped = False
|
||||
if self.config['dump']:
|
||||
data = self.vad_dump(self.context, proc_layer_name, vad)
|
||||
filedata = interfaces.plugins.FileInterface("pid.{0}.vad.{1:#x}-{2:#x}.dmp".format(
|
||||
proc.UniqueProcessId, vad.get_start(), vad.get_end()))
|
||||
filedata.data.write(data)
|
||||
self.produce_file(filedata)
|
||||
dumped = True
|
||||
|
||||
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']),
|
||||
winnt_protections), vad.get_commit_charge(), vad.get_private_memory(),
|
||||
format_hints.Hex(vad.get_parent()), vad.get_file_name()))
|
||||
format_hints.Hex(vad.get_parent()), vad.get_file_name(), dumped))
|
||||
|
||||
def run(self):
|
||||
|
||||
@@ -127,7 +172,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Offset", format_hints.Hex),
|
||||
("Start VPN", format_hints.Hex), ("End VPN", format_hints.Hex), ("Tag", str),
|
||||
("Protection", str), ("CommitCharge", int), ("PrivateMemory", int),
|
||||
("Parent", format_hints.Hex), ("File", str)],
|
||||
("Parent", format_hints.Hex), ("File", str), ("Dumped", bool)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
|
||||
@@ -11,7 +11,7 @@ from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.framework.symbols import intermed
|
||||
from volatility.framework.symbols.windows import extensions
|
||||
from volatility.plugins.windows import pslist, moddump, modules
|
||||
from volatility.plugins.windows import pslist, modules, dlllist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
@@ -32,7 +32,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
|
||||
return [
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.PluginRequirement(name = 'modules', plugin = modules.Modules, version = (1, 0, 0)),
|
||||
requirements.PluginRequirement(name = 'moddump', plugin = moddump.ModDump, version = (1, 0, 0)),
|
||||
requirements.VersionRequirement(name = 'dlllist', component = dlllist.DllList, version = (1, 0, 0)),
|
||||
requirements.TranslationLayerRequirement(name = 'primary',
|
||||
description = 'Memory layer for the kernel',
|
||||
architectures = ["Intel32", "Intel64"]),
|
||||
@@ -107,14 +107,14 @@ class VerInfo(interfaces.plugins.PluginInterface):
|
||||
except exceptions.InvalidAddressException:
|
||||
BaseDllName = renderers.UnreadableValue()
|
||||
|
||||
session_layer_name = moddump.ModDump.find_session_layer(self.context, session_layers, mod.DllBase)
|
||||
session_layer_name = modules.Modules.find_session_layer(self.context, session_layers, mod.DllBase)
|
||||
(major, minor, product, build) = [
|
||||
renderers.NotAvailableValue()
|
||||
] * 4 # type: Tuple[Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue]]
|
||||
renderers.NotAvailableValue()
|
||||
] * 4 # type: Tuple[Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue],Union[int, interfaces.renderers.BaseAbsentValue]]
|
||||
try:
|
||||
(major, minor, product, build) = self.get_version_information(self._context, pe_table_name,
|
||||
session_layer_name, mod.DllBase)
|
||||
except (exceptions.InvalidAddressException, ValueError, AttributeError):
|
||||
except (exceptions.InvalidAddressException, TypeError, AttributeError):
|
||||
(major, minor, product, build) = [renderers.UnreadableValue()] * 4
|
||||
|
||||
# the pid and process are not applicable for kernel modules
|
||||
@@ -157,7 +157,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
|
||||
mods = modules.Modules.list_modules(self.context, self.config["primary"], self.config["nt_symbols"])
|
||||
|
||||
# populate the session layers for kernel modules
|
||||
session_layers = moddump.ModDump.get_session_layers(self.context, self.config['primary'],
|
||||
session_layers = modules.Modules.get_session_layers(self.context, self.config['primary'],
|
||||
self.config['nt_symbols'])
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Base", format_hints.Hex), ("Name", str),
|
||||
|
||||
Reference in New Issue
Block a user